Blame tests/test_stream_server.py

Adam Williamson 01c1a4
#!/usr/bin/env python
Adam Williamson 01c1a4
# -*- coding: utf-8 -*-
Adam Williamson 01c1a4
Adam Williamson 01c1a4
"""
Adam Williamson 01c1a4
 (c) 2016 - Copyright Red Hat Inc
Adam Williamson 01c1a4
Adam Williamson 01c1a4
 Authors:
Adam Williamson 01c1a4
   Adam Williamson <awilliam@redhat.com></awilliam@redhat.com>
Adam Williamson 01c1a4
Adam Williamson 01c1a4
Tests for the Pagure streaming server.
Adam Williamson 01c1a4
Adam Williamson 01c1a4
"""
Adam Williamson 01c1a4
Adam Williamson 01c1a4
# obviously this is fine for testing.
Adam Williamson 01c1a4
# pylint: disable=locally-disabled, protected-access
Adam Williamson 01c1a4
Aurélien Bompard dcf6f6
from __future__ import unicode_literals
Aurélien Bompard dcf6f6
Adam Williamson 01c1a4
import logging
Adam Williamson 01c1a4
import os
Adam Williamson 01c1a4
import sys
Adam Williamson 01c1a4
import unittest
Adam Williamson 01c1a4
Adam Williamson 01c1a4
import mock
Aurélien Bompard 626417
import six
Adam Williamson 01c1a4
Adam Williamson 01c1a4
sys.path.insert(0, os.path.join(os.path.dirname(
Adam Williamson 01c1a4
    os.path.abspath(__file__)), '..'))
Adam Williamson 01c1a4
sys.path.insert(0, os.path.join(os.path.dirname(
Pierre-Yves Chibon 132515
    os.path.abspath(__file__)), '../pagure-ev'))
Adam Williamson 01c1a4
Pierre-Yves Chibon 76b0c9
if six.PY3:
Pierre-Yves Chibon 76b0c9
    raise unittest.case.SkipTest('Skipping on python3')
Pierre-Yves Chibon 76b0c9
Pierre-Yves Chibon 930073
import pagure.lib.query                             # pylint: disable=wrong-import-position
Pierre-Yves Chibon c13fca
from pagure.exceptions import PagureException, PagureEvException     # pylint: disable=wrong-import-position
Adam Williamson 01c1a4
import tests                                        # pylint: disable=wrong-import-position
Adam Williamson 01c1a4
# comes from ev-server/
Adam Williamson 01c1a4
import pagure_stream_server as pss                  # pylint: disable=wrong-import-position, import-error
Adam Williamson 01c1a4
Adam Williamson 01c1a4
logging.basicConfig(stream=sys.stderr)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
Adam Williamson 01c1a4
class StreamingServerTests(tests.Modeltests):
Adam Williamson 01c1a4
    """Tests for the streaming server."""
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def setUp(self):
Adam Williamson 01c1a4
        """Set up the environnment, run before every test."""
Adam Williamson 01c1a4
        super(StreamingServerTests, self).setUp()
Pierre-Yves Chibon b130e5
Pierre-Yves Chibon b130e5
        # Make sure the server uses the existing session
Pierre-Yves Chibon b130e5
        pss.SESSION = self.session
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Mock send_email, we never want to send or see emails here.
Adam Williamson 01c1a4
        self.mailpatcher = mock.patch('pagure.lib.notify.send_email')
Adam Williamson 01c1a4
        self.mailpatcher.start()
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Setup projects
Adam Williamson 01c1a4
        tests.create_projects(self.session)
Pierre-Yves Chibon 930073
        self.repo = pagure.lib.query._get_project(self.session, 'test')
Pierre-Yves Chibon 930073
        self.repo2 = pagure.lib.query._get_project(self.session, 'test2')
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Disable repo 2's issue tracker and PR tracker
Pierre-Yves Chibon 930073
        pagure.lib.query.update_project_settings(
Adam Williamson 01c1a4
            session=self.session,
Adam Williamson 01c1a4
            repo=self.repo2,
Adam Williamson 01c1a4
            user='pingou',
Adam Williamson 01c1a4
            settings={
Adam Williamson 01c1a4
                'issue_tracker': False,
Adam Williamson 01c1a4
                'pull_requests': False,
Adam Williamson 01c1a4
            }
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Create a public issue
Pierre-Yves Chibon 930073
        pagure.lib.query.new_issue(
Adam Williamson 01c1a4
            session=self.session,
Adam Williamson 01c1a4
            repo=self.repo,
Adam Williamson 01c1a4
            title='Test issue',
Adam Williamson 01c1a4
            content='We should work on this',
Adam Williamson 01c1a4
            user='pingou',
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Create a private issue
Pierre-Yves Chibon 930073
        pagure.lib.query.new_issue(
Adam Williamson 01c1a4
            session=self.session,
Adam Williamson 01c1a4
            repo=self.repo,
Adam Williamson 01c1a4
            title='Private issue #2',
Adam Williamson 01c1a4
            content='The world can see my porn folder',
Adam Williamson 01c1a4
            user='pingou',
Adam Williamson 01c1a4
            private=True,
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Create a PR
Pierre-Yves Chibon 930073
        pagure.lib.query.new_pull_request(
Adam Williamson 01c1a4
            session=self.session,
Adam Williamson 01c1a4
            repo_from=self.repo,
Adam Williamson 01c1a4
            repo_to=self.repo,
Adam Williamson 01c1a4
            branch_from='feature',
Adam Williamson 01c1a4
            branch_to='master',
Adam Williamson 01c1a4
            title='Test PR',
Adam Williamson 01c1a4
            user='pingou',
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def tearDown(self):
Adam Williamson 01c1a4
        "Stop the patchers, as well as calling super."""
Adam Williamson 01c1a4
        super(StreamingServerTests, self).tearDown()
Adam Williamson 01c1a4
        self.mailpatcher.stop()
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def test_parse_path(self):
Adam Williamson 01c1a4
        """Tests for _parse_path."""
Adam Williamson 01c1a4
        # Result format is: (username, namespace, repo, objtype, objid)
Adam Williamson 01c1a4
        # Simple case: issue for non-namespaced, non-forked repo.
Pierre-Yves Chibon c13fca
        result = pagure.utils.parse_path('/pagure/issue/1')
Adam Williamson 01c1a4
        self.assertEqual(result, (None, None, 'pagure', 'issue', '1'))
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Pull request for namespaced repo.
Pierre-Yves Chibon c13fca
        result = pagure.utils.parse_path('/fedora-qa/fedfind/pull-request/2')
Adam Williamson 01c1a4
        self.assertEqual(result, (None, 'fedora-qa', 'fedfind', 'pull-request', '2'))
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Issue for forked repo.
Pierre-Yves Chibon c13fca
        result = pagure.utils.parse_path('/fork/adamwill/pagure/issue/3')
Adam Williamson 01c1a4
        self.assertEqual(result, ('adamwill', None, 'pagure', 'issue', '3'))
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Issue for forked, namespaced repo.
Pierre-Yves Chibon c13fca
        result = pagure.utils.parse_path('/fork/pingou/fedora-qa/fedfind/issue/4')
Adam Williamson 01c1a4
        self.assertEqual(result, ('pingou', 'fedora-qa', 'fedfind', 'issue', '4'))
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Issue for repo named 'pull-request' (yeah, now we're getting tricksy).
Pierre-Yves Chibon c13fca
        result = pagure.utils.parse_path('/pull-request/issue/5')
Adam Williamson 01c1a4
        self.assertEqual(result, (None, None, 'pull-request', 'issue', '5'))
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Unknown object type.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"No known object",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/pagure/unexpected/1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # No object ID.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"No project or object ID",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/pagure/issue'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # No repo name. Note: we cannot catch 'namespace but no repo name',
Pierre-Yves Chibon 930073
        # but that should fail later in pagure.lib.query.get_project
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"No project or object ID",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/issue/1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # /fork but no user name.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"no user found!",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/fork/pagure/issue/1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Too many path components before object type.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"More path components",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/fork/adamwill/fedora-qa/fedfind/unexpected/issue/1'
Adam Williamson 01c1a4
        )
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Pierre-Yves Chibon c13fca
            PagureException,
Adam Williamson 01c1a4
            r"More path components",
Pierre-Yves Chibon c13fca
            pagure.utils.parse_path,
Pierre-Yves Chibon c13fca
            '/fedora-qa/fedfind/unexpected/issue/1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def test_get_issue(self):
Adam Williamson 01c1a4
        """Tests for _get_issue."""
Adam Williamson 01c1a4
        # Simple case: get the existing issue from the existing repo.
Adam Williamson 01c1a4
        result = pss._get_issue(self.repo, '1')
Adam Williamson 01c1a4
        self.assertEqual(result.id, 1)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Issue that doesn't exist.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"Issue '3' not found",
Adam Williamson 01c1a4
            pss._get_issue, self.repo, '3'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Private issue (for now we don't handle auth).
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"issue is private",
Adam Williamson 01c1a4
            pss._get_issue, self.repo, '2'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Issue from a project with no issue tracker.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"No issue tracker found",
Adam Williamson 01c1a4
            pss._get_issue, self.repo2, '1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def test_get_pull_request(self):
Adam Williamson 01c1a4
        """Tests for _get_pull_request."""
Adam Williamson 01c1a4
        # Simple case: get the existing PR from the existing repo.
Adam Williamson 01c1a4
        result = pss._get_pull_request(self.repo, '3')
Adam Williamson 01c1a4
        self.assertEqual(result.id, 3)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # PR that doesn't exist.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"Pull-Request '2' not found",
Adam Williamson 01c1a4
            pss._get_pull_request, self.repo, '2'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # PR from a project with no PR tracker.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"No pull-request tracker found",
Adam Williamson 01c1a4
            pss._get_pull_request, self.repo2, '1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    def test_get_obj_from_path(self):
Adam Williamson 01c1a4
        """Tests for get_obj_from_path."""
Adam Williamson 01c1a4
        # Simple issue case.
Adam Williamson 01c1a4
        result = pss.get_obj_from_path('/test/issue/1')
Adam Williamson 01c1a4
        self.assertEqual(result.id, 1)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Simple PR case.
Adam Williamson 01c1a4
        result = pss.get_obj_from_path('/test/pull-request/3')
Adam Williamson 01c1a4
        self.assertEqual(result.id, 3)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # Non-existent repo.
Aurélien Bompard 626417
        six.assertRaisesRegex(
Aurélien Bompard 626417
            self,
Adam Williamson 01c1a4
            PagureEvException,
Adam Williamson 01c1a4
            r"Project 'foo' not found",
Adam Williamson 01c1a4
            pss.get_obj_from_path, '/foo/issue/1'
Adam Williamson 01c1a4
        )
Adam Williamson 01c1a4
Adam Williamson 01c1a4
        # NOTE: we cannot test the 'Invalid object provided' exception
Adam Williamson 01c1a4
        # as it's a backup (current code will never hit it)
Adam Williamson 01c1a4
Pierre-Yves Chibon 393f31
Adam Williamson 01c1a4
if __name__ == '__main__':
Pierre-Yves Chibon 393f31
    unittest.main(verbosity=2)