#!/usr/bin/env python
# -*- 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:
#     - Andrea Tomasini <andrea.tomasini__at__agile42.com>
#     - Felix Schwarz <felix.schwarz__at__agile42.com>
#
import unittest

from trac.ticket.model import Milestone
from trac.util.datefmt import to_datetime, to_timestamp

from agilo.ticket.api import AgiloTicketSystem
from agilo.ticket.model import AgiloTicket, AgiloTicketModelManager
from agilo.scrum.sprint import Sprint
from agilo.utils import Key, Type
from agilo.utils.config import AgiloConfig
from agilo.test import TestEnvHelper


class TestAgiloTicket(unittest.TestCase):
    
    def setUp(self):
        # Creates the environment and a couple of tickets
        self.teh = TestEnvHelper()
        self.env = self.teh.get_env()
        
    def tearDown(self):
        self.teh.delete_created_tickets()
    
    def test_creation_of_agilo_ticket(self):
        t = AgiloTicket(self.env)
        t[Key.TYPE] = "story"
        t[Key.SUMMARY] = "This is an AgiloTicket"
        t[Key.DESCRIPTION] = "This is the description of the ticket..."
        t_id = t.insert()
        t2 = AgiloTicket(self.env, t_id)
        for k, v in t.values.items():
            if k in t.fields_for_type:
                # Compare only the fields which makes sense
                self.assertEqual(v, t2[k])
    
    def test_correct_handling_of_ticket_types(self):
        def matching_properties(agilo_ticket):
            # returns true if the ticket properties match the defined
            # ticket type proprties
            match = True
            config_fields = AgiloConfig(self.env).TYPES.get(agilo_ticket.get_type(), [])
            #print "Config Fields: %s" % config_fields
            for f in agilo_ticket.fields:
                if f.has_key(Key.SKIP):
                    #print "Field: %s, Skip: %s Match: %s" % (f[Key.NAME], f[Key.SKIP], match)
                    match = match and (f[Key.SKIP] != (f[Key.NAME] in config_fields))
            return match
            
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        story[Key.SUMMARY] = "This is a story..."
        story[Key.DESCRIPTION] = "A very nice one..."
        self.assertTrue(matching_properties(story))
        story.insert()
        self.assertTrue(matching_properties(story))
        
        # Not existing type should show all the properties
        nonex = AgiloTicket(self.env, t_type="nonex")
        nonex[Key.SUMMARY] = "This is a story..."
        nonex[Key.DESCRIPTION] = "A very nice one..."
        nonex.insert()
        self.assertTrue(matching_properties(nonex))
        
        # Setting type on the fly should work
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task[Key.SUMMARY] = "This should be a task"
        self.assertTrue(matching_properties(task))
        task.insert()
        self.assertTrue(matching_properties(task))
    
    def test_new_agilo_tickets_dont_have_a_resolution_by_default(self):
        # This bug was triggered by declaring the default resolution option in
        # trac's TicketModule so we make sure that the option is always known
        # when running this test
        from trac.env import Option
        Option('ticket', 'default_resolution', 'fixed', '')
        ticket = AgiloTicket(self.env, t_type=Type.TASK)
        self.assertEqual('', ticket[Key.RESOLUTION])
    
    def test_calculated_field_names(self):
        '''Test calculated field names'''
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        calculated_names = story.get_calculated_fields_names()
        self.assertEqual([Key.TOTAL_REMAINING_TIME, Key.ESTIMATED_REMAINING_TIME], 
                         calculated_names)
    
    def test_sync_milestone_attribute_with_sprint_automatically(self):
        m = Milestone(self.env)
        m.name = 'Test Milestone'
        m.insert()
        s = Sprint(self.env, name="Test", start=to_datetime(None), duration=20, milestone=m.name)
        s.save()
        # Now creates a ticket and set the Sprint to check whether the milestone is set too
        task = AgiloTicket(self.env)
        task[Key.SUMMARY] = "Test Task for Sprint Test"
        task[Key.TYPE] = Type.TASK
        self.assertEqual(Type.TASK, task[Key.TYPE])
        task[Key.SPRINT] = s.name
        task.insert()
        self.assertEqual(task[Key.MILESTONE], m.name)
        self.assertEqual(task[Key.MILESTONE], s.milestone)
        # Now change the milestone
        task[Key.MILESTONE] = 'Another One'
        task.save_changes('tester', 'changed milestone')
        self.assertEqual(task[Key.MILESTONE], 'Another One')
        self.assertEqual(task[Key.SPRINT], None)
        self.teh.move_changetime_to_the_past([task])
        # Now reset it to the Sprint1
        task[Key.SPRINT] = s.name
        task.save_changes('tester', 'changed sprint back')
        self.assertEqual(task[Key.MILESTONE], m.name)
        self.assertEqual(task[Key.MILESTONE], s.milestone)
        
    def test_owner_team_member_with_sprint_related_tickets(self):
        """
        Tests that the owner of a sprint related ticket, in case of restrict_owner
        option, is limited to the team members assigned to the sprint, in case the
        ticket has the Remaining Time property
        """
        # Set the restrict_owner option in the config
        teh = TestEnvHelper(strict=True)
        env = teh.get_env()
        self.assertTrue(AgiloTicketSystem(env).restrict_owner)
        t = teh.create_team('A-Team')
        self.assertTrue(t.save())
        s = teh.create_sprint('Test Me', team=t)
        self.assertEqual('Test Me', s.name)
        self.assertEqual(s.team, t)
        tm1 = teh.create_member('TM1', t)
        self.assertTrue(tm1.save())
        
        self.assertTrue(tm1.name in [m.name for m in t.members])
        tm2 = teh.create_member('TM2', t)
        self.assertTrue(tm2.save())
        self.assertTrue(tm2.name in [m.name for m in t.members])
        tm3 = teh.create_member('TM3')
        self.assertTrue(tm3.save())
        
        # Now make a ticket that is dependent from the sprint
        task = teh.create_ticket(Type.TASK, 
                                 props={Key.REMAINING_TIME: '12.5',
                                        Key.OWNER: tm1.name,
                                        Key.SPRINT: s.name})
        self.assertEqual(s.name, task[Key.SPRINT])
        f = task.get_field(Key.OWNER)
        self.assertTrue(tm1.name in f[Key.OPTIONS])
        self.assertTrue(tm2.name in f[Key.OPTIONS])
        self.assertFalse(tm3.name in f[Key.OPTIONS])

    def test_resource_list_when_no_resource_field_present(self):
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        self.assertEqual(None, story[Key.RESOURCES])
        self.assertEqual(0, len(story.get_resource_list()))
    
    def test_resource_list_can_include_owner(self):
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        story[Key.OWNER] = 'Foo'
        self.assertEqual(0, len(story.get_resource_list()))
        self.assertEqual(['Foo'], story.get_resource_list(include_owner=True))
        story[Key.OWNER] = ''
        self.assertEqual(0, len(story.get_resource_list(include_owner=True)))
    
    def test_resource_list_with_owner_and_resources(self):
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        story[Key.OWNER] = 'Foo'
        story[Key.RESOURCES] = 'Bar, Baz'
        self.assertEqual(['Foo', 'Bar', 'Baz'], 
                         story.get_resource_list(include_owner=True))
    
    def test_owner_is_only_listed_once_in_resource_list(self):
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        story[Key.OWNER] = 'Foo Bar'
        story[Key.RESOURCES] = 'Foo Bar'
        self.assertEqual(['Foo Bar'], story.get_resource_list(include_owner=True))
    
    def test_split_resource_string_for_resource_list(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task[Key.RESOURCES] = 'foo,bar,baz'
        self.assertEqual(['foo', 'bar', 'baz'], task.get_resource_list())
    
    def test_strip_usernames_for_resource_list(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task[Key.RESOURCES] = '  foo, bar,baz  '
        self.assertEqual(['foo', 'bar', 'baz'], task.get_resource_list())
    
    def test_empty_resource_list_for_empty_resource_string(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task[Key.RESOURCES] = ''
        self.assertEqual([], task.get_resource_list())
        
        task[Key.RESOURCES] = '   '
        self.assertEqual([], task.get_resource_list())
        
        task[Key.RESOURCES] = ' ,  , ,  '
        self.assertEqual([], task.get_resource_list())
        
    def test_get_field_with_field_name(self):
        """Tests the method get_field of a ticket, that returns the field dictionary"""
        task = AgiloTicket(self.env, t_type=Type.TASK)
        f = task.get_field(Key.REMAINING_TIME)
        self.assertNotEqual(None, f)
        self.assertEqual(Key.REMAINING_TIME, f[Key.NAME])
        
        us = AgiloTicket(self.env, t_type=Type.USER_STORY)
        f = us.get_field(Key.STORY_POINTS)
        self.assertNotEqual(None, f)
        self.assertEqual(Key.STORY_POINTS, f[Key.NAME])
        
        # Now calculated fields
        c = us.get_field(Key.TOTAL_REMAINING_TIME)
        self.assertNotEqual(None, c)
        self.assertEqual(Key.TOTAL_REMAINING_TIME, c[Key.NAME])
    
    def _set_default_type(self, new_value):
        agilo_config = AgiloConfig(self.env)
        old_value = agilo_config.get('default_type', 'ticket')
        agilo_config.change_option('default_type', new_value, 
                                   section='ticket', save=True)
        return old_value
    
    def test_new_ticket_uses_default_type_if_none_given(self):
        # This is the unit test covering bug #611 (see also regression test
        # TestOnlyTasksFieldsAreShownInPreview)
        def field_names(ticket):
            return [f['name'] for f in ticket.fields]
        
        old_default_type = self._set_default_type(Type.TASK)
        try:
            story = AgiloTicket(self.env, t_type=Type.USER_STORY)
            self.assertEqual(Type.USER_STORY, story[Key.TYPE])
            self.assertTrue(Key.STORY_POINTS in field_names(story))
            self.assertFalse(Key.REMAINING_TIME in field_names(story))
            self.assertFalse(Key.BUSINESS_VALUE in field_names(story))
            
            ticket = AgiloTicket(self.env)
            self.assertEqual(Type.TASK, ticket[Key.TYPE])
            self.assertTrue(Key.REMAINING_TIME in field_names(ticket))
            self.assertFalse(Key.STORY_POINTS in field_names(ticket))
            self.assertFalse(Key.BUSINESS_VALUE in field_names(ticket))
        finally:
            self._set_default_type(old_default_type)
    
    def test_can_serialize_ticket_task_to_dict(self):
        ticket = AgiloTicket(self.env, t_type=Type.TASK)
        self.assertNotEqual('fixed', ticket[Key.RESOLUTION])
        ticket[Key.SUMMARY] = 'My Summary'
        ticket.insert()
        expected = {
            # required
            Key.ID: ticket.id,
            Key.TYPE: Type.TASK,
            Key.SUMMARY: 'My Summary',
            Key.DESCRIPTION: '',
            Key.STATUS: '',
            Key.RESOLUTION: '',
            Key.REPORTER: '',
            Key.OWNER: '',
            # type specific
            Key.SPRINT: None,
            Key.REMAINING_TIME: None,
            Key.RESOURCES: None,
            
            'outgoing_links': [],
            'incoming_links': [],
            'time_of_last_change': to_timestamp(ticket.time_changed),
        }
        self.assertEquals(expected, ticket.as_dict())
    
    def test_can_serialize_calculated_fields(self):
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task[Key.REMAINING_TIME] = '5'
        story.link_to(task)
        story.insert()
        self.assertEquals(5, story.as_dict()[Key.TOTAL_REMAINING_TIME])
    
    def test_can_serialize_links(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task.insert()
        story = AgiloTicket(self.env, t_type=Type.USER_STORY)
        story.link_to(task)
        story.insert()
        self.assertEquals([task.id], story.as_dict()['outgoing_links'])
        self.assertEquals([story.id], task.as_dict()['incoming_links'])
    
    def test_throws_exception_if_not_persisted_ticket_should_be_serialized(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        self.assertRaises(ValueError, task.as_dict)
    
    def test_always_serialize_id_as_int(self):
        task = AgiloTicket(self.env, t_type=Type.TASK)
        task.insert()
        
        task_id = task.id
        task = AgiloTicket(self.env, str(task_id))
        self.assertEqual(task_id, task.as_dict()['id'])


class TestTicketModelManager(unittest.TestCase):
    """Tests AgiloTicket ModelManager, in particular the select method"""
    def setUp(self):
        self.teh = TestEnvHelper()
        self.env = self.teh.get_env()
        self.manager = AgiloTicketModelManager(self.env)
        
    def test_create_ticket(self):
        """Tests the creation of a ticket using the ModelManager"""
        t = self.manager.create(summary="This is a ticket")
        self.assertTrue(t.exists, "Ticket not existing...")
        self.assertEqual("This is a ticket", t[Key.SUMMARY])
        # Create without saving
        t2 = self.manager.create(summary="Not saved", save=False)
        self.assertFalse(t2.exists, "The ticket has been saved!")
        self.assertEqual("Not saved", t2[Key.SUMMARY])
        # Now add something and change the summary
        t2[Key.DESCRIPTION] = "changed"
        t2[Key.SUMMARY] = "Now saved"
        self.manager.save(t2)
        self.assertTrue(t2.exists)
        self.assertEqual("changed", t2[Key.DESCRIPTION])
        self.assertEqual("Now saved", t2[Key.SUMMARY])
    
    def test_ticket_caching(self):
        """Tests the ticket caching"""
        t1 = self.manager.create(summary="Ticket #1", 
                                 t_type=Type.USER_STORY)
        t1_dupe = self.manager.get(tkt_id=t1.id)
        self.assertEqual(t1, t1_dupe)
        
    def test_select_tickets(self):
        """Tests the select method to get tickets"""
        sprint = self.teh.create_sprint('Test', milestone='Test')
        t1 = self.manager.create(summary="Ticket #1", 
                                 t_type=Type.USER_STORY,
                                 sprint=sprint.name)
        t2 = self.manager.create(summary="Ticket #2",
                                 t_type=Type.TASK)
        # Now the plan select should return both tickets
        tickets = self.manager.select()
        self.assertTrue(t1 in tickets, "T1 is not in tickets!?")
        self.assertTrue(t2 in tickets, "T2 is not in tickets!?")
        # Now selects all tickets planned for sprint Test
        self.assertEqual(sprint.name, t1[Key.SPRINT])
        tickets = self.manager.select(criteria={Key.SPRINT: 'Test'})
        self.assertTrue(t1 in tickets, "T1 is not in ticket!?")
        self.assertFalse(t2 in tickets, "T2 is in tickets and should not?!")
        # Now selects all tickets planned for milestone Test
        self.assertEqual('Test', t1[Key.MILESTONE])
        tickets = self.manager.select(criteria={Key.MILESTONE: 'Test'})
        self.assertTrue(t1 in tickets, "T1 is not in ticket!?")
        self.assertFalse(t2 in tickets, "T2 is in tickets and should not?!")
        # Now tests the select with a limit to 1
        tickets = self.manager.select(limit=1)
        self.assertEqual(1, len(tickets))
        tickets = self.manager.select(limit=2)
        self.assertEqual(2, len(tickets))
        # Now select all the tickets that have been created before now
        tickets = self.manager.select(criteria={'changetime': '<=%s' % \
                                                t1.time_created})
        self.assertEqual(2, len(tickets))
        # Now try out the order by
        tickets = self.manager.select(order_by=['-sprint'])
        self.assertEqual(tickets[0], t1)
        self.assertEqual(tickets[1], t2)


if __name__ == '__main__':
#    suite = unittest.TestLoader().loadTestsFromTestCase(TestAgiloTicket)
#    #suite = unittest.TestSuite()
#    #suite.addTest(TestAgiloTicket('testTicketType'))
#    unittest.TextTestRunner(verbosity=0).run(suite)
    from agilo.ticket.web_ui import AgiloTicketModule
    teh = TestEnvHelper()
    env = teh.get_env()
    ticket = AgiloTicket(env, t_type=Type.TASK)
    assert 'fixed' != ticket[Key.RESOLUTION]
