# -*- coding: utf-8 -*-

"""

 (c) 2015-2018 - Copyright Red Hat Inc

 Authors:

   Pierre-Yves Chibon <pingou@pingoured.fr>

"""

from __future__ import unicode_literals

__requires__ = ['SQLAlchemy >= 0.8']

import pkg_resources

import datetime

import unittest

import shutil

import sys

import os

import markdown

from mock import patch, MagicMock

sys.path.insert(0, os.path.join(os.path.dirname(

    os.path.abspath(__file__)), '..'))

import pagure

import pagure.lib

import pagure.lib.model

import tests

class PagureLibtests_search_user(tests.Modeltests):

    """

    Test the pagure.lib search_user() method

    """

    maxDiff = None

    def test_search_user_all(self):

        """

        Test the method returns all the users for the given session

        """

        # Retrieve all users

        items = pagure.lib.search_user(self.session)

        self.assertEqual(2, len(items))

        self.assertEqual(2, items[0].id)

        self.assertEqual('foo', items[0].user)

        self.assertEqual('foo', items[0].username)

        self.assertEqual([], items[1].groups)

        self.assertEqual(1, items[1].id)

        self.assertEqual('pingou', items[1].user)

        self.assertEqual('pingou', items[1].username)

        self.assertEqual([], items[1].groups)

    def test_search_user_username(self):

        """

        Test the method returns the user for a given username

        """

        # Retrieve user by username

        item = pagure.lib.search_user(self.session, username='foo')

        self.assertEqual('foo', item.user)

        self.assertEqual('foo', item.username)

        self.assertEqual([], item.groups)

        item = pagure.lib.search_user(self.session, username='bar')

        self.assertEqual(None, item)

    def test_search_user_email(self):

        """

        Test the method returns a user for a given email address

        """

        # Retrieve user by email

        item = pagure.lib.search_user(self.session, email='foo@foo.com')

        self.assertEqual(None, item)

        item = pagure.lib.search_user(self.session, email='foo@bar.com')

        self.assertEqual('foo', item.user)

        self.assertEqual('foo', item.username)

        self.assertEqual([], item.groups)

        self.assertEqual(

            ['foo@bar.com'], [email.email for email in item.emails])

        item = pagure.lib.search_user(self.session, email='foo@pingou.com')

        self.assertEqual('pingou', item.user)

        self.assertEqual(

            sorted(['bar@pingou.com', 'foo@pingou.com']),

            sorted([email.email for email in item.emails]))

    def test_search_user_token(self):

        """

        Test the method returns a user for a given token

        """

        # Retrieve user by token

        item = pagure.lib.search_user(self.session, token='aaa')

        self.assertEqual(None, item)

        item = pagure.lib.model.User(

            user='pingou2',

            fullname='PY C',

            token='aaabbb',

            default_email='bar@pingou.com',

        )

        self.session.add(item)

        self.session.commit()

        item = pagure.lib.search_user(self.session, token='aaabbb')

        self.assertEqual('pingou2', item.user)

        self.assertEqual('PY C', item.fullname)

    def test_search_user_pattern(self):

        """

        Test the method returns a user for a given pattern

        """

        # Retrieve user by pattern

        item = pagure.lib.search_user(self.session, pattern='a*')

        self.assertEqual([], item)

        item = pagure.lib.model.User(

            user='pingou2',

            fullname='PY C',

            token='aaabbb',

            default_email='bar@pingou.com',

        )

        self.session.add(item)

        self.session.commit()

        items = pagure.lib.search_user(self.session, pattern='p*')

        self.assertEqual(2, len(items))

        self.assertEqual(1, items[0].id)

        self.assertEqual('pingou', items[0].user)

        self.assertEqual('pingou', items[0].username)

        self.assertEqual([], items[0].groups)

        self.assertEqual(

            sorted(['bar@pingou.com', 'foo@pingou.com']),

            sorted([email.email for email in items[0].emails]))

        self.assertEqual(3, items[1].id)

        self.assertEqual('pingou2', items[1].user)

        self.assertEqual('pingou2', items[1].username)

        self.assertEqual([], items[1].groups)

        self.assertEqual(

            [], [email.email for email in items[1].emails])

class PagureLibtests_search_projects(tests.Modeltests):

    """

    Test the pagure.lib search_projects() method

    """

    def setUp(self):

        super(PagureLibtests_search_projects, self).setUp()

        tests.create_projects(self.session)

    def test_search_projects_all(self):

        """

        Test the method returns all the projects for the given session

        """

        projects = pagure.lib.search_projects(self.session)

        self.assertEqual(len(projects), 3)

        self.assertEqual(projects[0].id, 1)

        self.assertEqual(projects[1].id, 2)

    def test_search_projects_username(self):

        """

        Test the method returns all the projects for the given username

        """

        projects = pagure.lib.search_projects(self.session, username='foo')

        self.assertEqual(len(projects), 0)

        projects = pagure.lib.search_projects(self.session, username='pingou')

        self.assertEqual(len(projects), 3)

        self.assertEqual(projects[0].id, 1)

        self.assertEqual(projects[1].id, 2)

    def test_search_projects_start(self):

        """

        Test the method returns all the projects for the given start

        """

        projects = pagure.lib.search_projects(self.session, start=1)

        self.assertEqual(len(projects), 2)

        self.assertEqual(projects[0].id, 2)

    def test_search_projects_limit(self):

        """

        Test the method returns all the projects for the given limit

        """

        projects = pagure.lib.search_projects(self.session, limit=1)

        self.assertEqual(len(projects), 1)

        self.assertEqual(projects[0].id, 1)

    def test_search_projects_count(self):

        """

        Test the method returns the count of the projects

        """

        projects = pagure.lib.search_projects(self.session, count=True)

        self.assertEqual(projects, 3)

    def test_search_projects_commit_access(self):

        """

        Test the method returns the project of user with only commit access

        """

        # Also check if the project shows up if a user doesn't

        # have admin access in the project

        # Check with commit access first

        project = pagure.lib.get_authorized_project(self.session, project_name='test')

        msg = pagure.lib.add_user_to_project(

            self.session,

            project=project,

            new_user='foo',

            user='pingou',

            access='commit'

        )

        self.assertEqual(msg, 'User added')

        self.session.commit()

        projects = pagure.lib.search_projects(self.session, username='foo')

        self.assertEqual(len(projects), 1)

    def test_search_projects_ticket_access(self):

        """

        Test the method does not returns the project of user with only ticket access

        """

        # Now check with only ticket access

        project = pagure.lib.get_authorized_project(self.session, project_name='test')

        msg = pagure.lib.add_user_to_project(

            self.session,

            project=project,

            new_user='foo',

            user='pingou',

            access='ticket'

        )

        self.assertEqual(msg, 'User added')

        self.session.commit()

        projects = pagure.lib.search_projects(self.session, username='foo')

        self.assertEqual(len(projects), 0)

    def test_search_project_forked(self):

        """

        Test the search_project for forked projects in pagure.lib.

        """

        # Create two forked repo

        item = pagure.lib.model.Project(

            user_id=2,  # foo

            name='test',

            description='test project #1',

            is_fork=True,

            parent_id=1,

            hook_token='aaabbbttt',

        )

        self.session.add(item)

        item = pagure.lib.model.Project(

            user_id=2,  # foo

            name='test2',

            description='test project #2',

            is_fork=True,

            parent_id=2,

            hook_token='aaabbbuuu',

        )

        self.session.add(item)

        # Since we have two forks, let's search them

        projects = pagure.lib.search_projects(self.session, fork=True)

        self.assertEqual(len(projects), 2)

        projects = pagure.lib.search_projects(self.session, fork=False)

        self.assertEqual(len(projects), 3)

    def test_search_projects_private(self):

        """

        Test the method for private projects

        """

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name='private_test',

            description='Private test project #1',

            hook_token='aaabbbcccpp',

            private=True,

        )

        self.session.add(item)

        self.session.commit()

        # non authenticated

        projects = pagure.lib.search_projects(self.session)

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['test.git', 'test2.git',

             'somenamespace/test3.git']

        )

        # non authenticated

        projects = pagure.lib.search_projects(

            self.session, username='pingou')

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['test.git', 'test2.git',

             'somenamespace/test3.git']

        )

        # authenticated as pingou

        projects = pagure.lib.search_projects(

            self.session, username='pingou', private='pingou')

        self.assertEqual(len(projects), 4)

        self.assertEqual(

            [p.path for p in projects],

            ['private_test.git', 'test.git', 'test2.git',

             'somenamespace/test3.git']

        )

        # authenticated as foo

        projects = pagure.lib.search_projects(

            self.session, username='pingou', private='foo')

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['test.git', 'test2.git',

             'somenamespace/test3.git']

        )

    def test_search_projects_tags(self):

        """

        Test the method returns all the projects for the given tags

        """

        # Add tags to the project

        project = pagure.lib._get_project(self.session, 'test')

        tag = pagure.lib.model.Tag(

            tag='fedora'

        )

        self.session.add(tag)

        self.session.commit()

        tp = pagure.lib.model.TagProject(

            project_id=project.id,

            tag='fedora'

        )

        self.session.add(tp)

        self.session.commit()

        projects = pagure.lib.search_projects(

            self.session, tags='fedora')

        self.assertEqual(len(projects), 1)

        self.assertEqual(projects[0].path, 'test.git')

    def test_search_projects_pattern(self):

        """

        Test the method returns all the projects for the given pattern

        """

        projects = pagure.lib.search_projects(

            self.session, pattern='test*')

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['test.git', 'test2.git', 'somenamespace/test3.git']

        )

    def test_search_projects_sort(self):

        """

        Test the method returns all the projects sorted by lastest and oldest

        """

        projects = pagure.lib.search_projects(

            self.session, pattern='*', sort='latest')

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['somenamespace/test3.git', 'test2.git', 'test.git']

        )

        projects = pagure.lib.search_projects(

            self.session, pattern='*', sort='oldest')

        self.assertEqual(len(projects), 3)

        self.assertEqual(

            [p.path for p in projects],

            ['test.git', 'test2.git', 'somenamespace/test3.git']

        )

class PagureLibtests(tests.Modeltests):

    """ Tests for pagure.lib """

    maxDiff = None

    def test_get_next_id(self):

        """ Test the get_next_id function of pagure.lib. """

        tests.create_projects(self.session)

        self.assertEqual(1, pagure.lib.get_next_id(self.session, 1))

    @patch('pagure.lib.git.update_git')

    @patch('pagure.lib.notify.send_email')

    def test_new_issue(self, p_send_email, p_ugt):

        """ Test the new_issue of pagure.lib. """

        p_send_email.return_value = True

        p_ugt.return_value = True

        tests.create_projects(self.session)

        repo = pagure.lib._get_project(self.session, 'test')

        # Set some priorities to the project

        repo.priorities = {'1': 'High', '2': 'Normal'}

        self.session.add(repo)

        self.session.commit()

        # Before

        issues = pagure.lib.search_issues(self.session, repo)

        self.assertEqual(len(issues), 0)

        self.assertEqual(repo.open_tickets, 0)

        self.assertEqual(repo.open_tickets_public, 0)

        # See where it fails

        self.assertRaises(

            pagure.exceptions.PagureException,

            pagure.lib.new_issue,

            session=self.session,

            repo=repo,

            title='Test issue',

            content='We should work on this',

            user='blah',

            ticketfolder=None

        )

        # Fails since we're trying to give a non-existant priority

        self.assertRaises(

            pagure.exceptions.PagureException,

            pagure.lib.new_issue,

            session=self.session,

            repo=repo,

            title='Test issue',

            content='We should work on this',

            user='pingou',

            ticketfolder=None,

            priority=0,

        )

        # Add an extra user to project `foo`

        repo = pagure.lib._get_project(self.session, 'test')

        msg = pagure.lib.add_user_to_project(

            session=self.session,

            project=repo,

            new_user='foo',

            user='pingou'

        )

        self.session.commit()

        self.assertEqual(msg, 'User added')

        # Try adding again this extra user to project `foo`

        self.assertRaises(

            pagure.exceptions.PagureException,

            pagure.lib.add_user_to_project,

            session=self.session,

            project=repo,

            new_user='foo',

            user='pingou'

        )

        self.session.commit()

        self.assertEqual(msg, 'User added')

        # Create issues to play with

        msg = pagure.lib.new_issue(

            session=self.session,

            repo=repo,

            title='Test issue',

            content='We should work on this',

            user='pingou',

            ticketfolder=None

        )

        self.session.commit()

        self.assertEqual(msg.title, 'Test issue')

        self.assertEqual(repo.open_tickets, 1)

        self.assertEqual(repo.open_tickets_public, 1)

        msg = pagure.lib.new_issue(

            session=self.session,

            repo=repo,

            title='Test issue #2',

            content='We should work on this for the second time',

            user='foo',

            status='Open',

            ticketfolder=None

        )

        self.session.commit()

        self.assertEqual(msg.title, 'Test issue #2')

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

        # After

        issues = pagure.lib.search_issues(self.session, repo)

        self.assertEqual(len(issues), 2)

    @patch('pagure.lib.git.update_git')

    @patch('pagure.lib.notify.send_email')

    def test_edit_issue(self, p_send_email, p_ugt):

        """ Test the edit_issue of pagure.lib. """

        p_send_email.return_value = True

        p_ugt.return_value = True

        self.test_new_issue()

        repo = pagure.lib._get_project(self.session, 'test')

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

        # Edit the issue

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None)

        self.session.commit()

        self.assertEqual(msg, None)

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            title='Test issue #2',

            content='We should work on this for the second time',

            status='Open',

        )

        self.session.commit()

        self.assertEqual(msg, None)

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            title='Foo issue #2',

            content='We should work on this period',

            status='Closed',

            close_status='Invalid',

            private=True,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue status updated to: Closed (was: Open)',

                'Issue close_status updated to: Invalid',

                'Issue private status set to: True'

            ]

        )

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            title='Foo issue #2',

            content='Fixed!',

            status='Closed',

            close_status='Fixed',

            private=False,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue close_status updated to: Fixed (was: Invalid)',

                'Issue private status set to: False (was: True)'

            ]

        )

        repo = pagure.lib._get_project(self.session, 'test')

        self.assertEqual(repo.open_tickets, 1)

        self.assertEqual(repo.open_tickets_public, 1)

        self.assertEqual(repo.issues[1].status, 'Closed')

        self.assertEqual(repo.issues[1].close_status, 'Fixed')

        # Edit the status: re-open the ticket

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            status='Open',

            ticketfolder=None,

            private=True,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue status updated to: Open (was: Closed)',

                'Issue private status set to: True'

            ]

        )

        repo = pagure.lib._get_project(self.session, 'test')

        for issue in repo.issues:

            self.assertEqual(issue.status, 'Open')

            self.assertEqual(issue.close_status, None)

        # 2 open but one of them is private

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 1)

        # Edit the status: re-close the ticket

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            status='Closed',

            close_status='Invalid',

            ticketfolder=None,

            private=True,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue status updated to: Closed (was: Open)',

                'Issue close_status updated to: Invalid'

            ]

        )

        repo = pagure.lib._get_project(self.session, 'test')

        self.assertEqual(repo.open_tickets, 1)

        self.assertEqual(repo.open_tickets_public, 1)

        self.assertEqual(repo.issues[1].status, 'Closed')

        self.assertEqual(repo.issues[1].close_status, 'Invalid')

    @patch('pagure.lib.git.update_git')

    @patch('pagure.lib.notify.send_email')

    def test_edit_issue_close_status(self, p_send_email, p_ugt):

        """ Test the edit_issue of pagure.lib. """

        p_send_email.return_value = True

        p_ugt.return_value = True

        self.test_new_issue()

        repo = pagure.lib._get_project(self.session, 'test')

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        self.assertEqual(issue.status, 'Open')

        self.assertEqual(issue.close_status, None)

        repo = pagure.lib._get_project(self.session, 'test')

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

        # Edit the issue, providing just a close_status should also close

        # the ticket

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            close_status='Fixed',

            ticketfolder=None)

        self.session.commit()

        self.assertEqual(msg, ['Issue close_status updated to: Fixed'])

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        self.assertEqual(issue.status, 'Closed')

        self.assertEqual(issue.close_status, 'Fixed')

        repo = pagure.lib._get_project(self.session, 'test')

        self.assertEqual(repo.open_tickets, 1)

        self.assertEqual(repo.open_tickets_public, 1)

        # Edit the issue, editing the status to open, should reset the

        # close_status

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            status='Open',

        )

        self.session.commit()

        self.assertEqual(

            msg, ['Issue status updated to: Open (was: Closed)'])

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        self.assertEqual(issue.status, 'Open')

        self.assertEqual(issue.close_status, None)

        repo = pagure.lib._get_project(self.session, 'test')

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

    @patch('pagure.lib.git.update_git')

    @patch('pagure.lib.notify.send_email')

    def test_edit_issue_priority(self, p_send_email, p_ugt):

        """ Test the edit_issue of pagure.lib when changing the priority.

        """

        p_send_email.return_value = True

        p_ugt.return_value = True

        self.test_new_issue()

        repo = pagure.lib.get_authorized_project(self.session, 'test')

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        # Set some priorities to the repo

        repo = pagure.lib.get_authorized_project(self.session, 'test')

        repo.priorities = {'1': 'High', '2': 'Normal'}

        self.session.add(repo)

        self.session.commit()

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

        # Edit the issue -- Wrong priority value: No changes

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            priority=3,

        )

        self.session.commit()

        self.assertEqual(msg, None)

        # Edit the issue -- Good priority

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            priority=2,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue priority set to: Normal'

            ]

        )

        # Edit the issue -- Update priority

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            priority=1,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue priority set to: High (was: Normal)'

            ]

        )

    @patch('pagure.lib.git.update_git')

    @patch('pagure.lib.notify.send_email')

    def test_edit_issue_depending(self, p_send_email, p_ugt):

        """ Test the edit_issue of pagure.lib when the issue depends on

        another.

        """

        p_send_email.return_value = True

        p_ugt.return_value = True

        tests.create_projects(self.session)

        repo = pagure.lib.get_authorized_project(self.session, 'test')

        # Create 3 issues

        msg = pagure.lib.new_issue(

            session=self.session,

            repo=repo,

            title='Test issue #1',

            content='We should work on this for the second time',

            user='foo',

            status='Open',

            ticketfolder=None

        )

        self.session.commit()

        self.assertEqual(msg.title, 'Test issue #1')

        msg = pagure.lib.new_issue(

            session=self.session,

            repo=repo,

            title='Test issue #2',

            content='We should work on this for the second time',

            user='foo',

            status='Open',

            ticketfolder=None

        )

        self.session.commit()

        self.assertEqual(msg.title, 'Test issue #2')

        msg = pagure.lib.new_issue(

            session=self.session,

            repo=repo,

            title='Test issue #3',

            content='We should work on this for the second time',

            user='foo',

            status='Open',

            ticketfolder=None

        )

        self.session.commit()

        self.assertEqual(msg.title, 'Test issue #3')

        issue = pagure.lib.search_issues(self.session, repo, issueid=2)

        self.assertEqual(repo.open_tickets, 3)

        self.assertEqual(repo.open_tickets_public, 3)

        # Make issue #2 blocking on issue #1

        msgs = pagure.lib.update_blocked_issue(

            self.session,

            repo,

            issue,

            blocks=['1'],

            username='pingou',

            ticketfolder=None,

        )

        self.assertEqual(msgs, ['Issue marked as blocking: #1'])

        # Make issue #2 depend on issue #3

        msgs = pagure.lib.update_dependency_issue(

            self.session,

            repo,

            issue,

            depends=['3'],

            username='pingou',

            ticketfolder=None,

        )

        self.assertEqual(msgs, ['Issue marked as depending on: #3'])

        # Edit the issue #3

        issue = pagure.lib.search_issues(self.session, repo, issueid=3)

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None)

        self.session.commit()

        self.assertEqual(msg, None)

        msg = pagure.lib.edit_issue(

            session=self.session,

            issue=issue,

            user='pingou',

            ticketfolder=None,

            title='Foo issue #2',

            content='We should work on this period',

            status='Closed',

            close_status='Invalid',

            private=True,

        )

        self.session.commit()

        self.assertEqual(

            msg,

            [

                'Issue status updated to: Closed (was: Open)',

                'Issue close_status updated to: Invalid',

                'Issue private status set to: True'

            ]

        )

        self.assertEqual(repo.open_tickets, 2)

        self.assertEqual(repo.open_tickets_public, 2)

    @patch('pagure.lib.REDIS', MagicMock(return_value=True))

    @patch('pagure.lib.git.update_git', MagicMock(return_value=True))

    @patch('pagure.lib.notify.send_email', MagicMock(return_value=True))

    def test_add_issue_dependency(self):

        """ Test the add_issue_dependency of pagure.lib. """

        self.test_new_issue()

        repo = pagure.lib._get_project(self.session, 'test')

        issue = pagure.lib.search_issues(self.session, repo, issueid=1)

        issue_blocked = pagure.lib.search_issues(

            self.session, repo, issueid=2)

        # Before

        self.assertEqual(issue.parents, [])

        self.assertEqual(issue.children, [])