# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
# OpenOLC is licensed under Mulan PSL v2.
# You can use this software according to the terms and conditions of the Mulan PSL v2.
# You may obtain a copy of Mulan PSL v2 at:
#         `http://license.coscl.org.cn/MulanPSL2`
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
# See the Mulan PSL v2 for more details.

import unittest
from unittest.mock import MagicMock, patch

from olc.bean.olc_control_request import OlcControlRequest
from olc.bean.tag_group import TagGroup
from olc.control.context.context import OlcContext
from olc.control.handler.statistic_handler import StatisticHandler
from olc.statistic.statistic import IStatistic, OlcStatistic


class TestStatisticHandler(unittest.TestCase):

    def setUp(self):
        self.handler = StatisticHandler()

    def test_in_coming_increments_thread_num(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]

        mock_statistic = MagicMock(spec=IStatistic)
        context.statistics[groups[0]] = mock_statistic

        with patch.object(self.handler, '_next_in_coming') as mock_next:
            self.handler.in_coming(context, groups)
            mock_statistic.inc_thread_num.assert_called_once()
            mock_next.assert_called_once_with(context, groups)

    def test_in_coming_records_start_time(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]

        mock_statistic = MagicMock(spec=IStatistic)
        context.statistics[groups[0]] = mock_statistic

        with patch("olc.control.handler.statistic_handler.time.perf_counter", return_value=10.5):
            with patch.object(self.handler, '_next_in_coming'):
                self.handler.in_coming(context, groups)

        self.assertEqual(context.start_time, 10.5)

    def test_out_coming_decrements_thread_num(self):
        context = OlcContext(OlcControlRequest())

        mock_statistic = MagicMock(spec=IStatistic)
        tag_group = TagGroup(domain="test", name="group1")
        context.statistics[tag_group] = mock_statistic

        with patch.object(self.handler, '_next_out_coming') as mock_next:
            self.handler.out_coming(context)
            mock_statistic.dec_thread_num.assert_called_once()
            mock_next.assert_called_once_with(context)

    def test_out_coming_records_pass_request_and_latency(self):
        context = OlcContext(OlcControlRequest())
        context.start_time = 10.0
        context.result.block = False

        mock_statistic = MagicMock(spec=IStatistic)
        tag_group = TagGroup(domain="test", name="group1")
        context.statistics[tag_group] = mock_statistic

        with patch("olc.control.handler.statistic_handler.time.perf_counter", return_value=10.025):
            with patch.object(self.handler, '_next_out_coming'):
                self.handler.out_coming(context)

        mock_statistic.inc_pass_request_num.assert_called_once()
        mock_statistic.inc_block_request_num.assert_not_called()
        mock_statistic.record_pass_latency_ms.assert_called_once()
        latency_ms = mock_statistic.record_pass_latency_ms.call_args.args[0]
        self.assertAlmostEqual(latency_ms, 25.0)

    def test_out_coming_records_block_request_and_no_latency(self):
        context = OlcContext(OlcControlRequest())
        context.start_time = 20.0
        context.result.block = True

        mock_statistic = MagicMock(spec=IStatistic)
        tag_group = TagGroup(domain="test", name="group1")
        context.statistics[tag_group] = mock_statistic

        with patch("olc.control.handler.statistic_handler.time.perf_counter", return_value=20.1):
            with patch.object(self.handler, '_next_out_coming'):
                self.handler.out_coming(context)

        mock_statistic.inc_pass_request_num.assert_not_called()
        mock_statistic.inc_block_request_num.assert_called_once()
        mock_statistic.record_pass_latency_ms.assert_not_called()

    def test_out_coming_skips_latency_without_start_time(self):
        context = OlcContext(OlcControlRequest())
        context.start_time = 0.0

        mock_statistic = MagicMock(spec=IStatistic)
        tag_group = TagGroup(domain="test", name="group1")
        context.statistics[tag_group] = mock_statistic

        with patch.object(self.handler, '_next_out_coming'):
            self.handler.out_coming(context)

        mock_statistic.record_pass_latency_ms.assert_not_called()

    def test_in_coming_with_multiple_statistics(self):
        context = OlcContext(OlcControlRequest())
        groups = [
            TagGroup(domain="test", name="group1"),
            TagGroup(domain="test", name="group2"),
        ]

        mock_statistic1 = MagicMock(spec=IStatistic)
        mock_statistic2 = MagicMock(spec=IStatistic)
        context.statistics[groups[0]] = mock_statistic1
        context.statistics[groups[1]] = mock_statistic2

        with patch.object(self.handler, '_next_in_coming'):
            self.handler.in_coming(context, groups)
            mock_statistic1.inc_thread_num.assert_called_once()
            mock_statistic2.inc_thread_num.assert_called_once()

    def test_out_coming_with_multiple_statistics(self):
        context = OlcContext(OlcControlRequest())

        mock_statistic1 = MagicMock(spec=IStatistic)
        mock_statistic2 = MagicMock(spec=IStatistic)
        tag_group1 = TagGroup(domain="test", name="group1")
        tag_group2 = TagGroup(domain="test", name="group2")
        context.statistics[tag_group1] = mock_statistic1
        context.statistics[tag_group2] = mock_statistic2

        with patch.object(self.handler, '_next_out_coming'):
            self.handler.out_coming(context)
            mock_statistic1.dec_thread_num.assert_called_once()
            mock_statistic2.dec_thread_num.assert_called_once()

    def test_in_coming_with_empty_statistics(self):
        context = OlcContext(OlcControlRequest())
        groups = []

        with patch.object(self.handler, '_next_in_coming') as mock_next:
            self.handler.in_coming(context, groups)
            mock_next.assert_called_once_with(context, groups)

    def test_out_coming_with_empty_statistics(self):
        context = OlcContext(OlcControlRequest())

        with patch.object(self.handler, '_next_out_coming') as mock_next:
            self.handler.out_coming(context)
            mock_next.assert_called_once_with(context)

    def test_in_coming_with_real_statistic(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]

        real_statistic = OlcStatistic()
        context.statistics[groups[0]] = real_statistic

        with patch.object(self.handler, '_next_in_coming'):
            self.handler.in_coming(context, groups)
            self.assertEqual(real_statistic.cur_thread_num(), 1)

    def test_out_coming_with_real_statistic(self):
        context = OlcContext(OlcControlRequest())

        real_statistic = OlcStatistic()
        real_statistic.inc_thread_num()
        tag_group = TagGroup(domain="test", name="group1")
        context.statistics[tag_group] = real_statistic

        with patch.object(self.handler, '_next_out_coming'):
            self.handler.out_coming(context)
            self.assertEqual(real_statistic.cur_thread_num(), 0)

    def test_in_out_coming_balance(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]

        real_statistic = OlcStatistic()
        context.statistics[groups[0]] = real_statistic

        with patch.object(self.handler, '_next_in_coming'):
            with patch.object(self.handler, '_next_out_coming'):
                self.handler.in_coming(context, groups)
                self.assertEqual(real_statistic.cur_thread_num(), 1)

                self.handler.out_coming(context)
                self.assertEqual(real_statistic.cur_thread_num(), 0)

    def test_repeated_out_coming_does_not_duplicate_completion_metrics(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]
        context.result.block = False

        real_statistic = OlcStatistic()
        context.statistics[groups[0]] = real_statistic

        with patch.object(self.handler, '_next_in_coming'), \
             patch.object(self.handler, '_next_out_coming'), \
             patch("olc.statistic.sliding_window.time.perf_counter") as mock_time:
            mock_time.return_value = 10.0
            self.handler.in_coming(context, groups)
            mock_time.return_value = 10.02
            self.handler.out_coming(context)
            mock_time.return_value = 10.03
            self.handler.out_coming(context)

            self.assertEqual(real_statistic.total_request_num(), 1)
            self.assertEqual(real_statistic.pass_request_num(), 1)
            self.assertEqual(real_statistic.block_request_num(), 0)
            self.assertEqual(real_statistic.cur_thread_num(), 0)

    def test_multiple_in_out_balance(self):
        context = OlcContext(OlcControlRequest())
        groups = [TagGroup(domain="test", name="group1")]

        real_statistic = OlcStatistic()
        context.statistics[groups[0]] = real_statistic

        with patch.object(self.handler, '_next_in_coming'):
            with patch.object(self.handler, '_next_out_coming'):
                for _ in range(5):
                    self.handler.in_coming(context, groups)
                self.assertEqual(real_statistic.cur_thread_num(), 5)

                for _ in range(3):
                    self.handler.out_coming(context)
                self.assertEqual(real_statistic.cur_thread_num(), 2)

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