# -*- encoding: utf-8 -*-
#   Copyright 2008 Agile42 GmbH, Berlin (Germany)
#   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.
#   
#   Author: 
#       - Jonas von Poser <jonas.vonposer_at_agile42.com>
#       - Felix Schwarz <felix.schwarz_at_agile42.com>

import unittest
from datetime import datetime, timedelta

import agilo.utils.filterwarnings

from trac.util.datefmt import get_timezone, localtz, to_datetime, utc

from agilo.charts.chart_generator import ChartGenerator
from agilo.scrum.backlog import RemainingTime
from agilo.scrum.backlog.charts import BurndownChartGenerator, \
    calculate_trend_series, _calculate_moving_average
from agilo.scrum.charts import ChartType
from agilo.scrum.team import TeamController
from agilo.utils import Key, Type
from agilo.utils.days_time import midnight

from agilo.test import TestEnvHelper
from agilo.scrum.backlog.tests.backlog_test import enable_backlog_filter, \
    add_attribute_as_task_field


class TrendCalculationTest(unittest.TestCase):
    def test_calculate_moving_average(self):
        avg = _calculate_moving_average
        
        self.assertRaises(ValueError, avg, [], 2)
        # matplotlib.mlab.movavg has also this strange behavior which we did not
        # implement: Does not look to useful.
        # self.assertEqual([1, 1], avg([2], 2))
        self.assertEqual([1.5], avg([1, 2], 2))
        self.assertEqual([1.5, 2.5, 4.0, 4.5], avg([1, 2, 3, 5, 4], 2))
        self.assertEqual([2, 3, 4], avg([1, 2, 3, 4, 5], 3))
        self.assertEqual([2], avg([1, 2, 3], 3))
    
    def test_trend_data_is_extrapolated(self):
        trend = calculate_trend_series([20, 30, 10], range(7))
        self.assertEqual(7, len(trend))
        self.assertEqual([25, 20, 15, 10, 5, 0], trend[1:])
    
    def test_can_extrapolate_trend_data_for_longer_sprints(self):
        trend = calculate_trend_series([20, 30, 10], range(20))
        self.assertEqual(20, len(trend))
        self.assertEqual([25, 20, 15, 10, 5] + [0] * 14, trend[1:])
    
    def test_trend_data_is_moving_average_of_last_two_items(self):
        trend = calculate_trend_series([20, 30, 10], range(4))
        self.assertEqual([25, 20], trend[1:3])
    
    def test_trend_series_starts_from_the_same_point_as_given_input_data(self):
        trend = calculate_trend_series([20, 30, 10], range(4))
        self.assertEqual(20, trend[0])


class BurndownChartTestCase(unittest.TestCase):
    # Class to share the setup code for the burndown chart between the 
    # tests for the 'component burndown' and the generic burndown chart. I 
    # wanted to keep the 'component burndown' functionality as separated as 
    # possible.
    
    def setUp(self, *args, **kwargs):
        super(BurndownChartTestCase, self).setUp(*args, **kwargs)
        self.teh = TestEnvHelper()
        self.teh.disable_sprint_date_normalization()
        self.env = self.teh.get_env()
        self.env.compmgr.enabled[BurndownChartGenerator] = True
    
    def tearDown(self):
        self.teh.cleanup()
    
    def _create_ticket(self, sprint, t_type=Type.TASK, **kwargs):
        attributes = {Key.SUMMARY: 'Task', Key.SPRINT: sprint.name}
        attributes.update(kwargs)
        ticket = self.teh.create_ticket(t_type, props=attributes)
        return ticket
    
    def _create_tasks_for_backlog(self):
        member = self.teh.create_member('FooBarMember', team='MyTeam')
        self.team = member.team
        self.now = to_datetime(None).replace(microsecond=0)
        sprint_start = self.now - timedelta(days=4)
        sprint = self.teh.create_sprint("FilteredChartSprint", start=sprint_start, 
                                        team=self.team)
        self.tasks = [self._create_ticket(sprint, **{Key.COMPONENT: 'foo', Key.REMAINING_TIME: '1'}),
                      self._create_ticket(sprint, **{Key.COMPONENT: 'bar', Key.REMAINING_TIME: '2'}), 
                      self._create_ticket(sprint, **{Key.REMAINING_TIME: '3'}), ]
        RemainingTime(self.env, self.tasks[0]).set_remaining_time(3, sprint_start)
        RemainingTime(self.env, self.tasks[1]).set_remaining_time(6, sprint_start)
        RemainingTime(self.env, self.tasks[2]).set_remaining_time(9, sprint_start)
        self.sprint = sprint
    
    def _store_commitment(self, commitment):
        cmd_class = TeamController.StoreTeamMetricCommand
        cmd = cmd_class(self.env, sprint=self.sprint, team=self.sprint.team,
                        metric=Key.COMMITMENT, value=commitment)
        TeamController(self.env).process_command(cmd)
    
    def get_widget(self, sprint_name, use_cache=False, filter_by=None, tz=localtz):
        get_widget = ChartGenerator(self.env).get_chartwidget
        widget = get_widget(ChartType.BURNDOWN, sprint_name=sprint_name,
                            use_cache=use_cache, filter_by=filter_by)
        widget.prepare_rendering(self.teh.mock_request(tz=tz))
        return widget
    
    def get_chart_data(self, param_name, filter_by=None, use_cache=False, tz=localtz):
        new_widget = self.get_widget(self.sprint.name, use_cache, filter_by, tz)
        return new_widget.data[param_name]
    
    def get_values_from_chart(self, param_name='remaining_times', 
                              filter_by=None, use_cache=False):
        data = self.get_chart_data(param_name, filter_by, use_cache)
        parameters = [rt for (t, rt) in data]
        return parameters
    
    def get_times_from_chart(self, param_name='remaining_times', tz=localtz):
        data = self.get_chart_data(param_name, None, False, tz=tz)
        parameters = [t for (t, rt) in data]
        return parameters


class BurndownChartTest(BurndownChartTestCase):
    def setUp(self):
        super(BurndownChartTest, self).setUp()
        self._create_tasks_for_backlog()
    
    def test_ideal_burndown_with_capacity_and_commitment(self):
        self._store_commitment(25)
        ideal_data = self.get_values_from_chart(param_name='ideal_data')
        self.assertNotEqual(0, len(ideal_data))
        self.assertAlmostEqual(25, ideal_data[0], 2)
        self.assertEqual(0, ideal_data[-1])
    
    def test_ideal_burndown_visible_even_if_no_team_assigned_to_sprint(self):
        self.sprint.team = None
        self.sprint.save()
        ideal_data = self.get_values_from_chart(param_name='ideal_data')
        self.assertNotEqual(0, len(ideal_data))
        self.assertEqual(3+6+9, ideal_data[0])
    
    def test_ideal_burndown_visible_even_if_no_commitment(self):
        self.assertNotEqual(None, self.sprint.team)
        self.assertNotEqual(0, len(self.sprint.team.members))
        ideal_data = self.get_values_from_chart(param_name='ideal_data')
        self.assertNotEqual(0, len(ideal_data))
        self.assertAlmostEqual(3+6+9, ideal_data[0], 2)
    
    def test_display_linear_ideal_burndown_if_no_capacity_available(self):
        for member in self.team.members:
            member.team = None
            member.save()
        self.assertEqual(0, len(self.team.members))
        initial_commitment = 3 + 6 + 9
        self._store_commitment(initial_commitment)
        
        ideal_data = self.get_values_from_chart(param_name='ideal_data')
        self.assertNotEqual(0, len(ideal_data))
        self.assertEqual(initial_commitment, ideal_data[0])
        
        nr_days = len(ideal_data)
        decrement_per_day = float(initial_commitment) / nr_days
        self.assertAlmostEqual(decrement_per_day * (nr_days-1), ideal_data[1], 1)
        self.assertAlmostEqual(0, ideal_data[-1], 2)
    
    def _js_to_datetime(self, value):
        timestamp = value / 1000
        # AT: we can replace the second to 0, in this case, the issues would be
        # that the chart and widget recalculate the "now" and may happen a sec.
        # later than the now of the test.
        return datetime.utcfromtimestamp(timestamp).replace(tzinfo=utc, second=0)
    
    def _as_flot_fake_utc(self, value):
        # AT: we can replace the second to 0, in this case, the issues would be
        # that the chart and widget recalculate the "now" and may happen a sec.
        # later than the now of the test.
        return value.replace(microsecond=0, second=0, tzinfo=utc)
    
    def test_last_item_in_ideal_burndown_is_end_of_sprint_time(self):
        times = self.get_times_from_chart('ideal_data')
        local_end = self.sprint.end.astimezone(localtz)
        fake_end = self._as_flot_fake_utc(local_end)
        self.assertEqual(fake_end, self._js_to_datetime(times[-1]))
    
    def test_can_deal_with_localtz_timezone_for_dates(self):
        times = self.get_times_from_chart()
        local_start = self.sprint.start.astimezone(localtz)
        fake_start = self._as_flot_fake_utc(local_start)
        self.assertEqual(fake_start, self._js_to_datetime(times[0]))
        fake_now = self._as_flot_fake_utc(self.now)
        self.assertEqual(fake_now, self._js_to_datetime(times[-1]))
    
    def test_can_deal_with_different_timezones_for_dates(self):
        # Actually Bangkok has no summer/winter time so that's ideal to test 
        # even with trac's own timezones which don't know anything about dst.
        bangkok_tz = get_timezone('GMT +7:00')
        times = self.get_times_from_chart(tz=bangkok_tz)
        start_in_bangkok = self.sprint.start.astimezone(bangkok_tz)
        fake_start = self._as_flot_fake_utc(start_in_bangkok)
        self.assertEqual(fake_start, self._js_to_datetime(times[0]))
        
        now = self.now.astimezone(bangkok_tz)
        fake_now = self._as_flot_fake_utc(now)
        self.assertEqual(fake_now, self._js_to_datetime(times[-1]))
    
    def test_weekends_are_relative_to_the_users_timezone(self):
        self.sprint.start = datetime(2009, 6, 23, 7, 0, tzinfo=utc)
        self.sprint.end = datetime(2009, 7, 6, 19, 0, tzinfo=utc)
        self.sprint.save()
        bangkok_tz = get_timezone('GMT +7:00')
        times = self.get_times_from_chart('utc_weekend_data', tz=bangkok_tz)
        
        self.assertEqual(4, len(times))
        first_weekend = [datetime(2009, 6, 27, 0, 0, tzinfo=bangkok_tz),
                         datetime(2009, 6, 28, 0, 0, tzinfo=bangkok_tz),]
        self.assertEqual(first_weekend, times[0:2])
        
        second_weekend = [datetime(2009, 7, 4, 0, 0, tzinfo=bangkok_tz),
                          datetime(2009, 7, 5, 0, 0, tzinfo=bangkok_tz),]
        self.assertEqual(second_weekend, times[2:])
    
    def test_today_is_relative_to_the_users_timezone(self):
        self.assertTrue(self.sprint.is_currently_running)
        bangkok_tz = get_timezone('GMT +7:00')
        times = self.get_chart_data('utc_today_data', tz=bangkok_tz)[0]
        self.assertEqual(2, len(times))
        # Now we test that midnight in bankgkok is normalized as UTC data with
        # offset
        now_in_bangkok = self.now.astimezone(bangkok_tz)
        today_start_in_bangkok = midnight(now_in_bangkok)
        self.assertEqual(today_start_in_bangkok, times[0])
        self.assertEqual(today_start_in_bangkok + timedelta(days=1), times[1])
    
    def test_guard_against_no_sprint(self):
        widget = self.get_widget('invalid_sprint')
        self.assertTrue('error_message' in widget.data)
    
    def _delete_commitment(self):
        cmd_class = TeamController.DeleteTeamMetricCommand
        cmd = cmd_class(self.env, sprint=self.sprint, metric=Key.COMMITMENT)
        TeamController(self.env).process_command(cmd)
    
    def set_remaining_time(self, ticket, value, since):
        self.teh.purge_ticket_history(ticket)
        rm = RemainingTime(self.env, ticket)
        rm.set_remaining_time(value, since)
        ticket[Key.REMAINING_TIME] = str(value)
        ticket.save_changes(None, None)
    
    def test_change_today_color_if_above_tolerance(self):
        # If the sprint is running since several days it might be that the team
        # is above the ideal burndown even in the first case so we need to set
        # a high commitment so that ideal burndown is above the actual one!
        self._store_commitment(150)
        self.set_remaining_time(self.tasks[0], 50, self.sprint.start)
        self.set_remaining_time(self.tasks[1], 20, self.sprint.start)
        
        widget = self.get_widget(self.sprint.name, use_cache=False)
        color_doing_well = widget.data['today_color']
        
        # Now the team should have more remaining hours left than in their 
        # ideal burndown
        self._delete_commitment()
        self.set_remaining_time(self.tasks[0], 1, self.sprint.start)
        self.set_remaining_time(self.tasks[1], 2, self.sprint.start)
        self.set_remaining_time(self.tasks[2], 3, self.sprint.start)
        
        new_widget = self.get_widget(self.sprint.name)
        color_for_problems = new_widget.data['today_color']
        self.assertNotEqual(color_doing_well, color_for_problems)
    
    def test_only_one_point_for_today(self):
        # this is the crazy burndown chart regression test
        now = to_datetime(None)
        three_days_ago = now - timedelta(days=3)
        # AT: why the time is set to 22:00 in UTC when now is in
        # localtz? A bit too many assumption may be?
        sprint_start = three_days_ago.replace(hour=22, minute=0, tzinfo=utc)
        self.sprint.start = sprint_start
        self.sprint.save()
        
        bangkok = get_timezone('GMT +7:00')
        times = self.get_times_from_chart(tz=bangkok)
        self.assertTrue(times[-2] < times[-1],
                        "%s is > of %s" % (times[-2], times[-1]))
        # We have to remove the difference of offset from the shifted UTC time
        fake_utc = self._as_flot_fake_utc(now) + \
            (bangkok.utcoffset(now) - localtz.utcoffset(now))
        self.assertEquals(fake_utc, self._js_to_datetime(times[-1]))
    
    def test_trend_lines_until_end_of_sprint(self):
        three_days_ago = to_datetime(None) - timedelta(days=3)
        sprint_start = three_days_ago.replace(hour=22, minute=0, tzinfo=utc)
        self.sprint.start = sprint_start
        self.sprint.save()
        
        berlin = get_timezone('GMT +2:00')
        trend_line = self.get_times_from_chart('trend_data', tz=berlin)
        self.assertNotEqual(0, len(trend_line))
        
        local_end = self.sprint.end.astimezone(berlin)
        fake_end = self._as_flot_fake_utc(local_end)
        self.assertEqual(fake_end, self._js_to_datetime(trend_line[-1]))


class BurndownChartCanBeFilteredByAdditionalAttributeTest(BurndownChartTestCase):
    
    def setUp(self):
        super(BurndownChartCanBeFilteredByAdditionalAttributeTest, self).setUp()
        add_attribute_as_task_field(self.env, Key.COMPONENT)
        enable_backlog_filter(self.env, Key.COMPONENT)
        self._create_tasks_for_backlog()
    
    def test_chart_can_show_all_items(self):
        remaining_times = self.get_values_from_chart()
        self.assertEqual(3 + 6 + 9, remaining_times[0])
        self.assertEqual(1 + 2 + 3, remaining_times[-1])
    
    def test_chart_shows_only_filtered_tasks(self):
        remaining_times = self.get_values_from_chart(filter_by='foo')
        self.assertEqual(3 + 9, remaining_times[0])
        self.assertEqual(1 + 3, remaining_times[-1])
    
    def test_no_capacity_line_if_filtered(self):
        capacity_data = self.get_values_from_chart(param_name='capacity_data')
        self.assertNotEqual(0, len(capacity_data))
        
        capacity_data = self.get_values_from_chart(param_name='capacity_data',
                                                   filter_by='foo')
        self.assertEqual(0, len(capacity_data))
    
    def get_commitment(self, team, sprint):
        cmd = TeamController.GetTeamCommitmentCommand(self.env, sprint=sprint, team=team)
        return TeamController(self.env).process_command(cmd)
    
    def test_filtered_chart_does_not_use_commitment_for_first_point(self):
        self._store_commitment(20)
        self.assertEqual(20, self.get_commitment(self.team, self.sprint))
        
        remaining_times = self.get_values_from_chart()
        self.assertEqual(20, remaining_times[0])
        
        remaining_times = self.get_values_from_chart(filter_by='foo')
        self.assertEqual(3 + 9, remaining_times[0])
    
    def test_cache_handles_filtered_chart_gracefully(self):
        capacity_data = self.get_values_from_chart(param_name='capacity_data', 
                                                   use_cache=True)
        self.assertNotEqual(0, len(capacity_data))
        
        capacity_data = self.get_values_from_chart(param_name='capacity_data',
                                                   filter_by='foo', 
                                                   use_cache=True)
        self.assertEqual(0, len(capacity_data))


if __name__ == '__main__':
    from agilo.test.testfinder import run_unit_tests
    run_unit_tests(__file__)


