<commit_before># Copyright (C) 2014 Linaro Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Test module for the DefConfHandler handler.."""

import json
import mongomock

from concurrent.futures import ThreadPoolExecutor
from mock import (
    MagicMock,
    patch,
)
from tornado import (
    ioloop,
    testing,
    web,
)

from handlers.app import AppHandler
from urls import _SUBSCRIPTION_URL

# Default Content-Type header returned by Tornado.
DEFAULT_CONTENT_TYPE = 'application/json; charset=UTF-8'


class TestSubscriptionHandler(
        testing.AsyncHTTPTestCase, testing.LogTrapTestCase):

    def setUp(self):
        self.mongodb_client = mongomock.Connection()

        super(TestSubscriptionHandler, self).setUp()

        patched_find_token = patch("handlers.base.BaseHandler._find_token")
        self.find_token = patched_find_token.start()
        self.find_token.return_value = "token"

        patched_validate_token = patch("handlers.base.validate_token")
        self.validate_token = patched_validate_token.start()
        self.validate_token.return_value = True

        self.addCleanup(patched_find_token.stop)
        self.addCleanup(patched_validate_token.stop)

    def get_app(self):
        dboptions = {
            'dbpassword': "",
            'dbuser': ""
        }

        settings = {
            'dboptions': dboptions,
            'client': self.mongodb_client,
            'executor': ThreadPoolExecutor(max_workers=2),
            'default_handler_class': AppHandler,
            'debug': False,
        }

        return web.Application([_SUBSCRIPTION_URL], **settings)

    def get_new_ioloop(self):
        return ioloop.IOLoop.instance()

    @patch('utils.db.find')
    @patch('utils.db.count')
    def test_get(self, mock_count, mock_find):
        mock_count.return_value = 0
        mock_find.return_value = []

        expected_body = (
            '{"count": 0, "code": 200, "limit": 0, "result": []}'
        )

        headers = {'Authorization': 'foo'}
        response = self.fetch('/api/subscription', headers=headers)

        self.assertEqual(response.code, 200)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)
        self.assertEqual(response.body, expected_body)

    @patch('handlers.subscription.SubscriptionHandler.collection')
    def test_get_by_id_not_found(self, mock_collection):
        mock_collection.find_one = MagicMock()
        mock_collection.find_one.return_value = None

        headers = {'Authorization': 'foo'}
        response = self.fetch('/api/subscription/sub', headers=headers)

        self.assertEqual(response.code, 404)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_post_without_token(self):

        body = json.dumps(dict(job='job', kernel='kernel'))

        response = self.fetch('/api/subscription', method='POST', body=body)

        self.assertEqual(response.code, 403)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_post_not_json(self):
        headers = {'Authorization': 'foo'}

        response = self.fetch(
            '/api/subscription', method='POST', body='', headers=headers
        )

        self.assertEqual(response.code, 415)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    @patch('utils.subscription.find_one')
    def test_post_valid(self, mock_find_one):
        mock_find_one.return_value = dict(_id='sub', job_id='job', emails=[])

        headers = {'Authorization': 'foo', 'Content-Type': 'application/json'}

        body = json.dumps(dict(job='job', email='email'))

        response = self.fetch(
            '/api/subscription', method='POST', body=body, headers=headers
        )

        self.assertEqual(response.code, 201)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    @patch('utils.subscription.find_one')
    def test_delete_valid_with_payload(self, mock_find_one):
        mock_find_one.return_value = dict(
            _id='sub', emails=['email'], job_id='job'
        )

        headers = {'Authorization': 'foo', 'Content-Type': 'application/json'}

        body = json.dumps(dict(email='email'))

        response = self.fetch(
            '/api/subscription/sub', method='DELETE', body=body,
            headers=headers, allow_nonstandard_methods=True,
        )

        self.assertEqual(response.code, 200)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_delete_valid_without_payload(self):
        headers = {'Authorization': 'foo'}

        response = self.fetch(
            '/api/subscription/sub', method='DELETE', headers=headers,
        )

        self.assertEqual(response.code, 200)<commit_msg>Fix API URLs - subscription tests.<commit_after># Copyright (C) 2014 Linaro Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Test module for the DefConfHandler handler.."""

import json
import mongomock

from concurrent.futures import ThreadPoolExecutor
from mock import (
    MagicMock,
    patch,
)
from tornado import (
    ioloop,
    testing,
    web,
)

from handlers.app import AppHandler
from urls import _SUBSCRIPTION_URL

# Default Content-Type header returned by Tornado.
DEFAULT_CONTENT_TYPE = 'application/json; charset=UTF-8'


class TestSubscriptionHandler(
        testing.AsyncHTTPTestCase, testing.LogTrapTestCase):

    def setUp(self):
        self.mongodb_client = mongomock.Connection()

        super(TestSubscriptionHandler, self).setUp()

        patched_find_token = patch("handlers.base.BaseHandler._find_token")
        self.find_token = patched_find_token.start()
        self.find_token.return_value = "token"

        patched_validate_token = patch("handlers.base.validate_token")
        self.validate_token = patched_validate_token.start()
        self.validate_token.return_value = True

        self.addCleanup(patched_find_token.stop)
        self.addCleanup(patched_validate_token.stop)

    def get_app(self):
        dboptions = {
            'dbpassword': "",
            'dbuser': ""
        }

        settings = {
            'dboptions': dboptions,
            'client': self.mongodb_client,
            'executor': ThreadPoolExecutor(max_workers=2),
            'default_handler_class': AppHandler,
            'debug': False,
        }

        return web.Application([_SUBSCRIPTION_URL], **settings)

    def get_new_ioloop(self):
        return ioloop.IOLoop.instance()

    @patch('utils.db.find')
    @patch('utils.db.count')
    def test_get(self, mock_count, mock_find):
        mock_count.return_value = 0
        mock_find.return_value = []

        expected_body = (
            '{"count": 0, "code": 200, "limit": 0, "result": []}'
        )

        headers = {'Authorization': 'foo'}
        response = self.fetch('/subscription', headers=headers)

        self.assertEqual(response.code, 200)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)
        self.assertEqual(response.body, expected_body)

    @patch('handlers.subscription.SubscriptionHandler.collection')
    def test_get_by_id_not_found(self, mock_collection):
        mock_collection.find_one = MagicMock()
        mock_collection.find_one.return_value = None

        headers = {'Authorization': 'foo'}
        response = self.fetch('/subscription/sub', headers=headers)

        self.assertEqual(response.code, 404)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_post_without_token(self):

        body = json.dumps(dict(job='job', kernel='kernel'))

        response = self.fetch('/subscription', method='POST', body=body)

        self.assertEqual(response.code, 403)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_post_not_json(self):
        headers = {'Authorization': 'foo'}

        response = self.fetch(
            '/subscription', method='POST', body='', headers=headers
        )

        self.assertEqual(response.code, 415)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    @patch('utils.subscription.find_one')
    def test_post_valid(self, mock_find_one):
        mock_find_one.return_value = dict(_id='sub', job_id='job', emails=[])

        headers = {'Authorization': 'foo', 'Content-Type': 'application/json'}

        body = json.dumps(dict(job='job', email='email'))

        response = self.fetch(
            '/subscription', method='POST', body=body, headers=headers
        )

        self.assertEqual(response.code, 201)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    @patch('utils.subscription.find_one')
    def test_delete_valid_with_payload(self, mock_find_one):
        mock_find_one.return_value = dict(
            _id='sub', emails=['email'], job_id='job'
        )

        headers = {'Authorization': 'foo', 'Content-Type': 'application/json'}

        body = json.dumps(dict(email='email'))

        response = self.fetch(
            '/subscription/sub', method='DELETE', body=body,
            headers=headers, allow_nonstandard_methods=True,
        )

        self.assertEqual(response.code, 200)
        self.assertEqual(
            response.headers['Content-Type'], DEFAULT_CONTENT_TYPE)

    def test_delete_valid_without_payload(self):
        headers = {'Authorization': 'foo'}

        response = self.fetch(
            '/subscription/sub', method='DELETE', headers=headers,
        )

        self.assertEqual(response.code, 200)
{'fraction_non_alphanumeric': 0.11372120549940817, 'fraction_numerical': 0.00619138668851862, 'mean_word_length': 2.969642211781713, 'pattern_counts': {'":': 8, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4661573', 'n_tokens_mistral': 3171, 'n_tokens_neox': 3050, 'n_words': 689}