Blob Blame Raw
# -*- coding: utf-8 -*-

"""
 (c) 2018 - Copyright Red Hat Inc

 Authors:
   Pierre-Yves Chibon <pingou@pingoured.fr>

"""

import unittest
import sys
import os

import json
from mock import patch, MagicMock

sys.path.insert(0, os.path.join(os.path.dirname(
    os.path.abspath(__file__)), '..'))

import pagure  # noqa
import pagure.lib  # noqa
import tests  # noqa


class PagureFlaskApiPRFlagtests(tests.Modeltests):
    """ Tests for the flask API of pagure for flagging pull-requests """

    maxDiff = None

    @patch('pagure.lib.notify.send_email', MagicMock(return_value=True))
    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(PagureFlaskApiPRFlagtests, self).setUp()

        pagure.config.config['REQUESTS_FOLDER'] = None

        tests.create_projects(self.session)
        tests.create_tokens(self.session)
        tests.create_tokens_acl(self.session)

        # Create a pull-request
        repo = pagure.lib.get_authorized_project(self.session, 'test')
        forked_repo = pagure.lib.get_authorized_project(self.session, 'test')
        req = pagure.lib.new_pull_request(
            session=self.session,
            repo_from=forked_repo,
            branch_from='master',
            repo_to=repo,
            branch_to='master',
            title='test pull-request',
            user='pingou',
            requestfolder=None,
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, 'test pull-request')

        # Check flags before
        # self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 0)

    def test_invalid_project(self):
        """ Test the flagging a PR on an invalid project. """

        headers = {'Authorization': 'token aaabbbcccddd'}

        # Invalid project
        output = self.app.post(
            '/api/0/foo/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 404)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Project not found",
                "error_code": "ENOPROJECT",
            }
        )

    def test_incorrect_project(self):
        """ Test the flagging a PR on the wrong project. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # Valid token, wrong project
        output = self.app.post(
            '/api/0/test2/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 401)
        data = json.loads(output.data)
        self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.name,
                         data['error_code'])
        self.assertEqual(pagure.api.APIERROR.EINVALIDTOK.value, data['error'])

    def test_no_pr(self):
        """ Test the flagging a PR when the PR doesn't exist. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # No PR
        output = self.app.post(
            '/api/0/test/pull-request/10/flag', headers=headers)
        self.assertEqual(output.status_code, 404)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Pull-Request not found",
                "error_code": "ENOREQ",
            }
        )

    def test_no_input(self):
        """ Test the flagging an existing PR but with no data. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # No input
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 400)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                u'error': u'Invalid or incomplete input submitted',
                u'error_code': u'EINVALIDREQ',
                u'errors': {
                    u'comment': [u'This field is required.'],
                    u'url': [u'This field is required.'],
                    u'username': [u'This field is required.']
                }
            }
        )

    def test_no_comment(self):
        """ Test the flagging an existing PR but with incomplete data. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'percent': 100,
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Incomplete request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 400)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Invalid or incomplete input submitted",
                "error_code": "EINVALIDREQ",
                "errors": {"comment": ["This field is required."]}
            }
        )

        # No change
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 0)

    @patch('pagure.lib.notify.send_email')
    def test_flagging_a_pul_request_with_notification(self, mock_email):
        """ Test the flagging a PR. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # Enable PR notifications
        repo = pagure.lib.get_authorized_project(self.session, 'test')
        settings = repo.settings
        settings['notify_on_pull-request_flag'] = True
        repo.settings = settings
        self.session.add(repo)
        self.session.commit()

        data = {
            'username': 'Jenkins',
            'comment': 'Tests running',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Valid request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        pr_uid = data['flag']['pull_request_uid']
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests running',
                    u'date_created': u'1510742565',
                    u'percent': None,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'pending',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests running')
        self.assertEqual(request.flags[0].percent, None)

        # Check the notification sent
        mock_email.assert_called_once_with(
            u'\nJenkins flagged the pull-request `test pull-request` '
            u'as pending: Tests running\n\n'
            u'https://pagure.org/test/pull-request/1\n',
            u'PR #1 - Jenkins: pending',
            u'bar@pingou.com',
            in_reply_to=u'test-pull-request-' + pr_uid,
            mail_id=u'test-pull-request-' + pr_uid + '-1',
            project_name=u'test',
            user_from=u'Jenkins'
        )

    def test_updating_flag(self):
        """ Test the updating the flag of a PR. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'comment': 'Tests running',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Valid request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests running',
                    u'date_created': u'1510742565',
                    u'percent': None,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'pending',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests running')
        self.assertEqual(request.flags[0].percent, None)

        # Update flag  -  w/o providing the status
        data = {
            'username': 'Jenkins',
            'percent': 100,
            'comment': 'Tests passed',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests passed',
                    u'date_created': u'1510742565',
                    u'percent': 100,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'success',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag updated',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests passed')
        self.assertEqual(request.flags[0].percent, 100)

    def test_adding_two_flags(self):
        """ Test the adding two flags to a PR. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'comment': 'Tests passed',
            'status': 'success',
            'percent': '100',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Valid request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests passed',
                    u'date_created': u'1510742565',
                    u'percent': 100,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'success',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests passed')
        self.assertEqual(request.flags[0].percent, 100)

        data = {
            'username': 'Jenkins',
            'comment': 'Tests running again',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
        }

        # Valid request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertNotEqual(
            data['uid'], 'jenkins_build_pagure_100+seed')
        data['uid'] = 'jenkins_build_pagure_100+seed'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests running again',
                    u'date_created': u'1510742565',
                    u'percent': None,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'pending',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # Two flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 2)
        self.assertEqual(request.flags[0].comment, 'Tests passed')
        self.assertEqual(request.flags[0].percent, 100)
        self.assertEqual(request.flags[1].comment, 'Tests running again')
        self.assertEqual(request.flags[1].percent, None)


class PagureFlaskApiPRFlagUserTokentests(tests.Modeltests):
    """ Tests for the flask API of pagure for flagging pull-requests using
    an user token (ie: not restricted to a specific project).
    """

    maxDiff = None

    @patch('pagure.lib.notify.send_email', MagicMock(return_value=True))
    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(PagureFlaskApiPRFlagUserTokentests, self).setUp()

        pagure.config.config['REQUESTS_FOLDER'] = None

        tests.create_projects(self.session)
        tests.create_tokens(self.session, project_id=None)
        tests.create_tokens_acl(self.session)

        # Create a pull-request
        repo = pagure.lib.get_authorized_project(self.session, 'test')
        forked_repo = pagure.lib.get_authorized_project(self.session, 'test')
        req = pagure.lib.new_pull_request(
            session=self.session,
            repo_from=forked_repo,
            branch_from='master',
            repo_to=repo,
            branch_to='master',
            title='test pull-request',
            user='pingou',
            requestfolder=None,
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, 'test pull-request')

        # Check flags before
        # self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 0)

    def test_no_pr(self):
        """ Test flagging a non-existing PR. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # Invalid project
        output = self.app.post(
            '/api/0/foo/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 404)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Project not found",
                "error_code": "ENOPROJECT",
            }
        )

    def test_no_pr_other_project(self):
        """ Test flagging a non-existing PR on a different project. """
        headers = {'Authorization': 'token aaabbbcccddd'}
        # Valid token, wrong project
        output = self.app.post(
            '/api/0/test2/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 404)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Pull-Request not found",
                "error_code": "ENOREQ",
            }
        )

    def test_no_input(self):
        """ Test flagging an existing PR but without submitting any data. """
        headers = {'Authorization': 'token aaabbbcccddd'}

        # No input
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', headers=headers)
        self.assertEqual(output.status_code, 400)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                u'error': u'Invalid or incomplete input submitted',
                u'error_code': u'EINVALIDREQ',
                u'errors': {
                    u'comment': [u'This field is required.'],
                    u'url': [u'This field is required.'],
                    u'username': [u'This field is required.']
                }
            }
        )

    def test_no_comment(self):
        """ Test flagging an existing PR but without all the required info.
        """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'percent': 100,
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Incomplete request
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 400)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Invalid or incomplete input submitted",
                "error_code": "EINVALIDREQ",
                "errors": {"comment": ["This field is required."]}
            }
        )

        # No change
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 0)

    def test_invalid_status(self):
        """ Test flagging an existing PR but with an invalid status.
        """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'status': 'failed',
            'comment': 'Failed to run the tests',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Invalid status submitted
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 400)
        data = json.loads(output.data)
        self.assertDictEqual(
            data,
            {
                "error": "Invalid or incomplete input submitted",
                "error_code": "EINVALIDREQ",
                "errors": {"status": ["Not a valid choice"]}
            }
        )

        # No change
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 0)

    @patch('pagure.lib.notify.send_email')
    def test_flag_pr_no_status(self, mock_email):
        """ Test flagging an existing PR without providing a status.

        Also check that no notifications have been sent.
        """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'percent': 0,
            'comment': 'Tests failed',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Valid request  -  w/o providing the status
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests failed',
                    u'date_created': u'1510742565',
                    u'percent': 0,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'failure',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests failed')
        self.assertEqual(request.flags[0].percent, 0)

        # no notifications sent
        mock_email.assert_not_called()

    def test_editing_flag(self):
        """ Test flagging an existing PR without providing a status.
        """
        headers = {'Authorization': 'token aaabbbcccddd'}

        data = {
            'username': 'Jenkins',
            'status': 'failure',
            'comment': 'Tests failed',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
        }

        # Valid request  -  w/o providing the status
        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests failed',
                    u'date_created': u'1510742565',
                    u'percent': None,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'failure',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag added',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # One flag added
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests failed')
        self.assertEqual(request.flags[0].percent, None)

        # Update flag
        data = {
            'username': 'Jenkins',
            'percent': 100,
            'comment': 'Tests passed',
            'url': 'http://jenkins.cloud.fedoraproject.org/',
            'uid': 'jenkins_build_pagure_100+seed',
            'status': 'success',
        }

        output = self.app.post(
            '/api/0/test/pull-request/1/flag', data=data, headers=headers)
        self.assertEqual(output.status_code, 200)
        data = json.loads(output.data)
        data['flag']['date_created'] = u'1510742565'
        data['flag']['pull_request_uid'] = u'62b49f00d489452994de5010565fab81'
        self.assertDictEqual(
            data,
            {
                u'flag': {
                    u'comment': u'Tests passed',
                    u'date_created': u'1510742565',
                    u'percent': 100,
                    u'pull_request_uid': u'62b49f00d489452994de5010565fab81',
                    u'status': u'success',
                    u'url': u'http://jenkins.cloud.fedoraproject.org/',
                    u'user': {
                        u'default_email': u'bar@pingou.com',
                        u'emails': [u'bar@pingou.com', u'foo@pingou.com'],
                        u'fullname': u'PY C',
                        u'name': u'pingou'
                    },
                    u'username': u'Jenkins'},
                u'message': u'Flag updated',
                u'uid': u'jenkins_build_pagure_100+seed'
            }
        )

        # Still only one flag
        self.session = pagure.lib.create_session(self.dbpath)
        request = pagure.lib.search_pull_requests(
            self.session, project_id=1, requestid=1)
        self.assertEqual(len(request.flags), 1)
        self.assertEqual(request.flags[0].comment, 'Tests passed')
        self.assertEqual(request.flags[0].percent, 100)


if __name__ == '__main__':
    unittest.main(verbosity=2)