# -*- coding: utf-8 -*-
#   Copyright 2008 Agile42 GmbH - Andrea Tomasini 
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
# 
# Authors:
#    - Jonas von Poser <jonas.vonposer@agile42.com>
#    - Felix Schwarz <felix.schwarz__at__agile42.com>


from datetime import timedelta
import unittest

from trac.perm import PermissionCache, PermissionError, PermissionSystem
from trac.resource import Resource
from trac.ticket.model import Type as TicketType
from trac.util.datefmt import to_datetime
from trac.web.api import Request

from agilo.ticket.web_ui import AgiloTicketModule
from agilo.utils import Action, Type, Key, Role, Realm
from agilo.utils.config import AgiloConfig
from agilo.utils.permissions import AgiloPolicy
from agilo.test import TestEnvHelper


name_team_member = 'tester'
name_product_owner = 'product owner'

class TestAgiloPermissions(unittest.TestCase):
    
    def setUp(self):
        """Creates the environment and a couple of tickets"""
        self.teh = TestEnvHelper()
        self.env = self.teh.get_env()
        self.env.config.set('trac', 'permission_policies', 'AgiloPolicy, DefaultPermissionPolicy, LegacyAttachmentPolicy')
        self.perm = PermissionSystem(self.env)
    
    def tearDown(self):
        self.teh.delete_created_tickets()
    
    def test_roles(self):
        self.teh.grant_permission('master', 'SCRUM_MASTER')
        self.teh.grant_permission('owner', 'PRODUCT_OWNER')
        
        # test if contains main and sub permission
        # for scrum master
        permissions = self.perm.get_user_permissions('master')
        self.failUnless('SCRUM_MASTER' in permissions)
        self.failUnless(Action.SAVE_REMAINING_TIME in permissions)
        
        # for product owner
        permissions = self.perm.get_user_permissions('owner')
        self.failUnless('PRODUCT_OWNER' in permissions)
        self.failUnless(Action.CREATE_STORY in permissions)
        self.failUnless(Action.CREATE_REQUIREMENT in permissions)
    
    def test_ticket_permissions(self):
        # create user with the necessary permission
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        perm_cache = PermissionCache(self.env, name_team_member)
        
        ticket = self.teh.create_ticket(Type.TASK)
        perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.TICKET_EDIT)
        # a team member should be able to save the remaining time for unassigned
        # tickets
        perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.SAVE_REMAINING_TIME)
        
        # Don't change the remaining time for tickets which belong to other 
        # team members
        ticket[Key.OWNER] = "Just another team member"
        ticket.save_changes("foo", "bar")
        
        new_perm_cache = PermissionCache(self.env, name_team_member)
        self.assertRaises(PermissionError,
                          new_perm_cache(Realm.TICKET, ticket.id).assert_permission, 
                          Action.SAVE_REMAINING_TIME)
    
    def test_edit_description_action_is_scoped_as_well(self):
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        perm = PermissionCache(self.env, name_team_member)
        requirement = self.teh.create_ticket(Type.REQUIREMENT)
        requirement_resource = requirement.resource
        self.assertEqual(False, perm.has_permission(Action.TICKET_EDIT_DESCRIPTION, requirement_resource))
    
    def test_reporters_can_only_edit_unassigned_tickets(self):
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        ticket = self.teh.create_ticket(Type.TASK)
        ticket[Key.REPORTER] = name_team_member
        ticket[Key.OWNER] = None
        some_minutes_ago = to_datetime(None) - timedelta(minutes=2)
        ticket.save_changes("foo", "bar", when=some_minutes_ago)
        self.assertEqual(name_team_member, ticket[Key.REPORTER])
        self.assertNotEqual(name_team_member, ticket[Key.OWNER])
        
        perm_cache = PermissionCache(self.env, name_team_member)
        perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.TICKET_EDIT)
        perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.TICKET_EDIT_PAGE_ACCESS)
    
    def test_ticket_owner_or_resource_can_save_time(self):
        another_team_member = 'another team member'
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        self.teh.grant_permission(another_team_member, Role.TEAM_MEMBER)
        ticket = self.teh.create_ticket(Type.TASK, props={Key.OWNER: name_team_member,
                                                          Key.REMAINING_TIME: '12'})
        # Check that name_team_member can change remaining time
        new_perm_cache = PermissionCache(self.env, name_team_member)
        new_perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.SAVE_REMAINING_TIME)
        
        # Check that another_team_member can't change remaining time
        self.assertTrue(another_team_member not in ticket.get_resource_list(include_owner=True))
        new_perm_cache = PermissionCache(self.env, another_team_member)
        self.assertRaises(PermissionError,
                          new_perm_cache(Realm.TICKET, ticket.id).assert_permission, 
                          Action.SAVE_REMAINING_TIME)
    
    def test_all_users_in_resources_can_edit_ticket(self):
        ticket = self.teh.create_ticket(Type.TASK)
        ticket[Key.OWNER] = "Just another team member"
        ticket[Key.RESOURCES] = " foo, %s, bar " % name_team_member
        ticket.save_changes("foo", "bar")
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        perm_cache = PermissionCache(self.env, name_team_member)
        
        self.assertNotEqual(ticket[Key.OWNER], name_team_member)
        perm_cache(Realm.TICKET, ticket.id).assert_permission(Action.TICKET_EDIT)
    
    def test_attachement_permissions(self):
        # create user with the necessary permission
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        self.teh.grant_permission(name_product_owner, Role.PRODUCT_OWNER)
        
        ticket = self.teh.create_ticket(Type.REQUIREMENT)
        # None of the two users we use for testing must be the ticket's owner,
        # else they could always edit the ticket.
        ticket[Key.OWNER] = 'someone'
        ticket.save_changes('foo', 'we have to change the owner')
        
        # a product owner should be able to attach files to the ticket, but a 
        # team member must not
        po_perm_cache = PermissionCache(self.env, name_product_owner)
        self.assertEqual(True, Action.ATTACHMENT_CREATE in po_perm_cache(Realm.TICKET, ticket.id))
        
        tm_perm_cache = PermissionCache(self.env, name_team_member)
        self.assertRaises(PermissionError,
            tm_perm_cache(Realm.TICKET, ticket.id).assert_permission, Action.ATTACHMENT_CREATE)
    
    def test_can_access_ticket_edit_page_if_he_can_create_referenced_tickets(self):
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        perm_cache = PermissionCache(self.env, name_team_member)
        
        story = self.teh.create_ticket(Type.USER_STORY)
        self.assertEqual(True, perm_cache.has_permission(Action.TICKET_EDIT_PAGE_ACCESS, story.resource))
        self.assertEqual(False, perm_cache.has_permission(Action.TICKET_EDIT, story.resource))
    
    # TODO: self.perm() as replacement for perm_cache...
    
    def _create_custom_ticket_type(self, type_name, field_names):
        custom_type = TicketType(self.env)
        custom_type.name = type_name
        custom_type.insert()
        
        config = AgiloConfig(self.env)
        config.change_option(type_name, field_names, section=AgiloConfig.AGILO_TYPES)
        config.reload()
        self.assertTrue(type_name in config.get_available_types())
    
    def test_can_access_ticket_edit_page_for_custom_types(self):
        custom_type_name = 'MaintenanceTask'
        self._create_custom_ticket_type(custom_type_name, [Key.COMPONENT])
        permission_name = 'CREATE_' + custom_type_name.upper()
        self.teh.grant_permission(name_team_member, permission_name)
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        
        perm_cache = PermissionCache(self.env, name_team_member)
        ticket = self.teh.create_ticket(custom_type_name)
        self.assertEqual(True, perm_cache.has_permission(Action.TICKET_EDIT, ticket.resource))
        self.assertEqual(True, perm_cache.has_permission(Action.TICKET_EDIT_PAGE_ACCESS, ticket.resource))
    
    def test_team_members_can_edit_bugs(self):
        bug = self.teh.create_ticket(Type.BUG, {Key.SUMMARY: 'A bug'})
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        
        perm = PermissionCache(self.env, name_team_member)
        self.assertEqual(True, perm.has_permission(Action.TICKET_EDIT, bug.resource))


class TestAgiloTicketModulePermissions(unittest.TestCase):
    """
    Tests the permissions using a Mock HTTP request to mock the behavior of
    the ticket module
    """
    class MockRequest(Request):
        """
        Represent a Mock HTTP Request object, to embed permission and
        environment
        """
        def __init__(self, env, *args, **kwargs):
            def get(*args):
                return 'http://localhost/'
            env.get = get
            super(TestAgiloTicketModulePermissions.MockRequest, self).__init__(env, None)
        @property
        def method(self):
            return u'MOCK'
    
    def setUp(self):
        """Creates the environment and a couple of tickets"""
        self.teh = TestEnvHelper()
        self.env = self.teh.get_env()
        # add agilo policy to policies
        self.env.config.set('trac', 'permission_policies', 'AgiloPolicy, DefaultPermissionPolicy, LegacyAttachmentPolicy')
        # create user with the necessary permission
        self.teh.grant_permission(name_team_member, Role.TEAM_MEMBER)
        self.teh.grant_permission(name_product_owner, Role.PRODUCT_OWNER)
        
    def tearDown(self):
        self.teh.delete_created_tickets()
    
    def test_can_edit_ticket_from_ticket_module(self):
        """
        Test the can_edit_ticket method directly on the TicketModule
        using a Mocked Request Object
        """
        task = self.teh.create_ticket(Type.TASK, props={Key.OWNER: name_team_member})
        # Build a fake HTTPRequest
        env = self.teh.get_env()
        req = TestAgiloTicketModulePermissions.MockRequest(env)
        req.perm = PermissionCache(env, name_team_member)
        req.authname = name_team_member
        self.assertTrue(AgiloTicketModule(env).can_edit_ticket(req, task),
                        "No permission to edit: %s as: %s!!!" % (task, req.authname))
        # Now make sure that the product owner can't edit task
        req.perm = PermissionCache(env, name_product_owner)
        req.authname = name_product_owner
        self.assertFalse(AgiloTicketModule(env).can_edit_ticket(req, task),
                         "Permission to edit: %s as: %s!!!" % (task, req.authname))
        # Now create a Story, both role should be able to edit it
        story = self.teh.create_ticket(Type.USER_STORY, 
                                       props={Key.OWNER: name_product_owner,
                                              Key.STORY_PRIORITY: 'Mandatory'})
        self.assertTrue(AgiloTicketModule(env).can_edit_ticket(req, story),
                        "No permission to edit: %s as: %s!!!" % (story, req.authname))
        # Now the team member too
        req.perm = PermissionCache(env, name_team_member)
        req.authname = name_team_member
        self.assertEqual(False, AgiloTicketModule(env).can_edit_ticket(req, story))
        # Now a Requirement that should only be touched by the Product Owner
        requirement = self.teh.create_ticket(Type.REQUIREMENT, 
                                             props={Key.OWNER: name_product_owner,
                                                    Key.BUSINESS_VALUE: '2000'})
        self.assertFalse(AgiloTicketModule(env).can_edit_ticket(req, requirement),
                         "Permission to edit: %s as: %s!!!" % (requirement, req.authname))
        # Now the team member too
        req.perm = PermissionCache(env, name_team_member)
        req.authname = name_product_owner
        self.assertTrue(AgiloTicketModule(env).can_edit_ticket(req, requirement),
                        "No permission to edit: %s as: %s!!!" % (requirement, req.authname))
    
    def test_create_related_tickets(self):
        """
        Test the list of possible related ticket to create given a type and
        a login permissions.
        """
        env = self.teh.get_env()
        req = TestAgiloTicketModulePermissions.MockRequest(env)
        req.perm = PermissionCache(env, name_product_owner)
        req.authname = name_product_owner
        story = self.teh.create_ticket(Type.USER_STORY, 
                                       props={Key.OWNER: name_product_owner,
                                              Key.STORY_PRIORITY: 'Mandatory'})
        # Build a fake data dictionary containing the ticket
        data = {Key.TICKET: story}
        AgiloTicketModule(env)._prepare_create_referenced(req, data)
        # Now check that being a Product Owner there is no link to create a task
        self.assertFalse(Type.TASK in data['create_referenced'])
        
        # Now login as a team member and the link should be there
        req.perm = PermissionCache(env, name_team_member)
        req.authname = name_team_member
        AgiloTicketModule(env)._prepare_create_referenced(req, data)
        allowed_links = data['create_referenced']
        allowed_destination_types = [l.dest_type for l in allowed_links]
        self.assertTrue(Type.TASK in allowed_destination_types)



class BacklogEditPermissionTest(unittest.TestCase):
    
    def setUp(self):
        self.teh = TestEnvHelper()
        env = self.teh.get_env()
        self.policy = AgiloPolicy(env)
        self.env = env
    
    def policy_decision(self, resource=None, username='foo'):
        perm = PermissionCache(self.env, username)
        return self.policy.check_permission(Action.BACKLOG_EDIT, username, resource, perm)
    
    def test_backlog_edit_without_resource_falls_back_to_trac_permissions(self):
        self.assertEqual(None, self.policy_decision(resource=None))
        
        self.teh.grant_permission('foo', Action.BACKLOG_EDIT)
        self.assertEqual(None, self.policy_decision(resource=None))
    
    def test_product_owner_has_backlog_edit_without_resource_because_they_can_potentially_edit_the_product_backlog(self):
        self.teh.grant_permission('foo', Role.PRODUCT_OWNER)
        self.assertEqual(True, self.policy_decision(resource=None))
    
    def test_scrum_master_has_backlog_edit_without_resource_because_they_can_potentially_edit_the_sprint_backlog(self):
        self.teh.grant_permission('foo', Role.SCRUM_MASTER)
        self.assertEqual(True, self.policy_decision(resource=None))
    
    def product_backlog_resource(self):
        return Resource(Realm.BACKLOG, Key.PRODUCT_BACKLOG)
    
    def sprint_backlog_resource(self):
        return Resource(Realm.BACKLOG, Key.SPRINT_BACKLOG)
    
    def other_backlog_resource(self):
        return Resource(Realm.BACKLOG, 'My Own Backlog')
    
    def test_product_owner_can_edit_the_product_backlog(self):
        self.teh.grant_permission('foo', Role.PRODUCT_OWNER)
        
        self.assertEqual(None, self.policy_decision(resource=self.sprint_backlog_resource()))
        self.assertEqual(True, self.policy_decision(resource=self.product_backlog_resource()))
    
    def test_scrum_master_can_edit_the_sprint_backlog(self):
        self.teh.grant_permission('foo', Role.SCRUM_MASTER)
        
        self.assertEqual(None, self.policy_decision(resource=self.product_backlog_resource()))
        self.assertEqual(True, self.policy_decision(resource=self.sprint_backlog_resource()))
    
    def test_all_authenticated_users_can_unknown_backlogs(self):
        self.teh.grant_permission('foo', Role.SCRUM_MASTER)
        
        other_backlog = self.other_backlog_resource()
        self.assertEqual(True, self.policy_decision(resource=other_backlog))
        other_backlog = self.policy_decision(resource=self.other_backlog_resource(), username='anonymous')
        self.assertEqual(None, other_backlog)
    
    def test_no_endless_loop_if_permission_is_checked_with_string_instead_of_resource(self):
        perm = PermissionCache(self.env, 'foo')
        perm.has_permission('AGILO_BACKLOG_EDIT', '%s:Sprint Backlog' % Realm.BACKLOG) 


if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(TestAgiloPermissions)
    unittest.TextTestRunner(verbosity=0).run(suite)
