<commit_before>"""Unit tests for reviewbotext.views.WorkerStatusView."""

from __future__ import unicode_literals

import json
from collections import OrderedDict

import kgb
from django.urls import reverse

from reviewbotext.tests.testcase import TestCase


class WorkerStatusViewTests(kgb.SpyAgency, TestCase):
    """Unit tests for reviewbotext.views.WorkerStatusView."""

    def test_get(self):
        """Testing WorkerStatusView.get"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [],
        }

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'success',
                'hosts': [
                    {
                        'hostname': 'bot1.example.com',
                        'tools': [
                            {
                                'name': 'tool1',
                                'entry_point': 'path.to.tool1:Tool1',
                                'version': '1.0',
                                'description': 'Test tool 1',
                                'tool_options': [
                                    {
                                        'name': 'option1',
                                        'field_type': ('django.forms.'
                                                       'BooleanField'),
                                        'default': True,
                                        'field_options': {
                                            'label': 'Option 1',
                                            'required': False,
                                            'help_text': 'Test.',
                                        },
                                    },
                                ],
                                'timeout': 100,
                                'working_directory_required': False,
                            },
                        ],
                    },
                    {
                        'hostname': 'bot2.example.com',
                        'tools': [],
                    },
                ],
            })

    def test_get_with_worker_status_error(self):
        """Testing WorkerStatusView.get with worker status=error"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {
            'status': 'error',
            'error': 'Oh no.',
        }

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Error from <EMAIL>: Oh no.',
            })

    def test_get_with_worker_status_unknown(self):
        """Testing WorkerStatusView.get with worker status unknown"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {}

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': (
                    "Unexpected result when querying worker status for "
                    "<EMAIL>. Please check the worker's "
                    "logs for information."
                ),
            })

    def test_get_with_ioerror(self):
        """Testing WorkerStatusView.get with IOError querying workers"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpRaise(IOError('Oh no.')))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Unable to connect to broker: Oh no.',
            })

    def test_get_with_not_configured(self):
        """Testing WorkerStatusView.get with Review Bot not configured"""
        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Review Bot is not yet configured.',
            })
<commit_msg>Fix Review Bot extension unit tests on Review Board 3.<commit_after>"""Unit tests for reviewbotext.views.WorkerStatusView."""

from __future__ import unicode_literals

import json
from collections import OrderedDict

import kgb

try:
    # Django >= 1.11, Review Board >= 4.0
    from django.urls import reverse
except ImportError:
    # Django 1.6, Review Board 3.0
    from django.core.urlresolvers import reverse

from reviewbotext.tests.testcase import TestCase


class WorkerStatusViewTests(kgb.SpyAgency, TestCase):
    """Unit tests for reviewbotext.views.WorkerStatusView."""

    def test_get(self):
        """Testing WorkerStatusView.get"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [],
        }

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'success',
                'hosts': [
                    {
                        'hostname': 'bot1.example.com',
                        'tools': [
                            {
                                'name': 'tool1',
                                'entry_point': 'path.to.tool1:Tool1',
                                'version': '1.0',
                                'description': 'Test tool 1',
                                'tool_options': [
                                    {
                                        'name': 'option1',
                                        'field_type': ('django.forms.'
                                                       'BooleanField'),
                                        'default': True,
                                        'field_options': {
                                            'label': 'Option 1',
                                            'required': False,
                                            'help_text': 'Test.',
                                        },
                                    },
                                ],
                                'timeout': 100,
                                'working_directory_required': False,
                            },
                        ],
                    },
                    {
                        'hostname': 'bot2.example.com',
                        'tools': [],
                    },
                ],
            })

    def test_get_with_worker_status_error(self):
        """Testing WorkerStatusView.get with worker status=error"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {
            'status': 'error',
            'error': 'Oh no.',
        }

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Error from <EMAIL>: Oh no.',
            })

    def test_get_with_worker_status_unknown(self):
        """Testing WorkerStatusView.get with worker status unknown"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        hosts = OrderedDict()
        hosts['<EMAIL>'] = {
            'status': 'ok',
            'tools': [
                {
                    'name': 'tool1',
                    'entry_point': 'path.to.tool1:Tool1',
                    'version': '1.0',
                    'description': 'Test tool 1',
                    'tool_options': [
                        {
                            'name': 'option1',
                            'field_type': ('django.forms.'
                                           'BooleanField'),
                            'default': True,
                            'field_options': {
                                'label': 'Option 1',
                                'required': False,
                                'help_text': 'Test.',
                            },
                        },
                    ],
                    'timeout': 100,
                    'working_directory_required': False,
                },
            ],
        }
        hosts['<EMAIL>'] = {}

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([hosts]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': (
                    "Unexpected result when querying worker status for "
                    "<EMAIL>. Please check the worker's "
                    "logs for information."
                ),
            })

    def test_get_with_ioerror(self):
        """Testing WorkerStatusView.get with IOError querying workers"""
        user = self.create_user()

        extension = self.extension
        extension.settings['user'] = user.pk
        extension.settings['broker_url'] = 'example.com'

        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpRaise(IOError('Oh no.')))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Unable to connect to broker: Oh no.',
            })

    def test_get_with_not_configured(self):
        """Testing WorkerStatusView.get with Review Bot not configured"""
        self.spy_on(self.extension.celery.control.broadcast,
                    op=kgb.SpyOpReturn([]))

        response = self.client.get(reverse('reviewbot-worker-status'))

        self.assertEqual(
            json.loads(response.content.decode('utf-8')),
            {
                'state': 'error',
                'error': 'Review Bot is not yet configured.',
            })

{'fraction_non_alphanumeric': 0.11677157479851756, 'fraction_numerical': 0.006588622860168245, 'mean_word_length': 0.9838954370404948, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3867906', 'n_tokens_mistral': 4125, 'n_tokens_neox': 3916, 'n_words': 692}