# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
# MindIE 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, call

from olc.admission.admission_controller import (
    AbsAdmissionController,
    AllowController,
    DenyController,
    DynamicController,
)
from olc.bean.match_wrapper import MatchWrapper
from olc.bean.olc_config_rule import AdmissionPolicy, AlgInstance
from olc.bean.tag_group import TagGroup
from olc.exception.olc_dynamic_exception import OLCDynamicException


def _make_dynamic_wrapper(alg_name: str = "test_alg", alg_param: dict = None) -> MatchWrapper[AdmissionPolicy]:
    tag_group = TagGroup(domain="test", name="test_group", enabled=True, priority=500)
    policy = AdmissionPolicy(
        name="test_policy",
        category="test",
        enabled=True,
        block_msg="",
        type="dynamic",
        calculate_alg=AlgInstance(alg_name=alg_name, alg_param=alg_param or {}),
    )
    return MatchWrapper(tag_group, policy)


class TestAllowController(unittest.TestCase):

    def test_refresh_status_does_not_raise(self):
        controller = AllowController()
        controller.refresh_status()

    def test_get_result_always_false(self):
        controller = AllowController()
        self.assertFalse(controller.get_result())

    def test_multiple_calls_always_false(self):
        controller = AllowController()
        for _ in range(100):
            self.assertFalse(controller.get_result())


class TestDenyController(unittest.TestCase):

    def test_refresh_status_does_not_raise(self):
        controller = DenyController()
        controller.refresh_status()

    def test_get_result_always_true(self):
        controller = DenyController()
        self.assertTrue(controller.get_result())

    def test_multiple_calls_always_true(self):
        controller = DenyController()
        for _ in range(100):
            self.assertTrue(controller.get_result())


class TestDynamicController(unittest.TestCase):

    def test_refresh_status_does_not_raise_when_no_alg(self):
        wrapper = _make_dynamic_wrapper()
        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = None
            controller = DynamicController(wrapper)
            controller.refresh_status()

    def test_get_result_returns_bool(self):
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = False
        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            result = controller.get_result()
            self.assertIsInstance(result, bool)

    def test_get_result_consistent(self):
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = False
        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            for _ in range(100):
                self.assertIsInstance(controller.get_result(), bool)

    def test_run_alg_with_no_alg_returns_false(self):
        """测试算法不存在时 run_alg 返回 False"""
        wrapper = _make_dynamic_wrapper("non_existent_alg")
        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = None
            controller = DynamicController(wrapper)
            result = controller.run_alg()
            self.assertFalse(result)

    def test_run_alg_executes_make_decision(self):
        """测试 run_alg 调用算法的 make_decision"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {}
        mock_alg.make_decision.return_value = True

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            result = controller.run_alg()

        self.assertTrue(result)
        mock_alg.make_decision.assert_called_once()

    def test_run_alg_with_static_params(self):
        """测试 run_alg 处理静态参数"""
        wrapper = _make_dynamic_wrapper(alg_param={"threshold": 10})
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {"threshold": MagicMock(dynamic=False)}
        mock_alg.make_decision.return_value = False

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            result = controller.run_alg()

        self.assertFalse(result)

    def test_run_alg_with_dynamic_params(self):
        """测试 run_alg 处理动态参数"""
        wrapper = _make_dynamic_wrapper(alg_param={"threshold": "$dynamic_key"})
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {"threshold": MagicMock(dynamic=True, key="threshold")}
        mock_alg.make_decision.return_value = True

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr, patch(
            "olc.admission.admission_controller.OlcCollectorManager.get_instance"
        ) as mock_collector:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            mock_collector.return_value.get_dyn_key_true_value.return_value = 20
            controller = DynamicController(wrapper)
            result = controller.run_alg()

        self.assertTrue(result)
        mock_collector.return_value.get_dyn_key_true_value.assert_called_once_with("$dynamic_key")

    def test_run_alg_with_dynamic_params_failure(self):
        """测试 run_alg 在动态参数获取失败时返回 False"""
        wrapper = _make_dynamic_wrapper(alg_param={"threshold": "$dynamic_key"})
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {"threshold": MagicMock(dynamic=True, key="threshold")}

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr, patch(
            "olc.admission.admission_controller.OlcCollectorManager.get_instance"
        ) as mock_collector:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            mock_collector.return_value.get_dyn_key_true_value.return_value = None
            controller = DynamicController(wrapper)
            result = controller.run_alg()

        self.assertFalse(result)

    def test_refresh_status_when_every_request_false(self):
        """测试 refresh_status 在 every_request=False 时调用 run_alg 并更新状态"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = False
        mock_alg.para_names = {}
        mock_alg.make_decision.return_value = True

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            controller.refresh_status()

        mock_alg.make_decision.assert_called_once()
        self.assertTrue(controller._status)

    def test_refresh_status_when_every_request_true(self):
        """测试 refresh_status 在 every_request=True 时不调用 run_alg"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {}

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            controller.refresh_status()

        mock_alg.make_decision.assert_not_called()

    def test_get_result_when_every_request_true(self):
        """测试 get_result 在 every_request=True 时调用 run_alg"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {}
        mock_alg.make_decision.return_value = True

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            result = controller.get_result()

        self.assertTrue(result)
        mock_alg.make_decision.assert_called_once()

    def test_get_result_when_every_request_false(self):
        """测试 get_result 在 every_request=False 时返回缓存状态"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = False
        mock_alg.para_names = {}

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            controller._status = True
            result = controller.get_result()

        self.assertTrue(result)
        mock_alg.make_decision.assert_not_called()

    def test_update_controller_updates_status(self):
        """测试 update_controller 更新状态"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = False
        mock_alg.para_names = {}

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)

            controller._status = False
            controller.update_controller(True)

            self.assertTrue(controller._status)
            self.assertFalse(controller._pre_status)

            controller.update_controller(False)

            self.assertFalse(controller._status)
            self.assertTrue(controller._pre_status)

    def test_run_alg_exception_returns_false(self):
        """测试 run_alg 异常时返回 False"""
        wrapper = _make_dynamic_wrapper()
        mock_alg = MagicMock()
        mock_alg.every_request = True
        mock_alg.para_names = {}
        mock_alg.make_decision.side_effect = Exception("test exception")

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = mock_alg
            controller = DynamicController(wrapper)
            result = controller.run_alg()

        self.assertFalse(result)

    def test_repr_contains_info(self):
        """测试 __repr__ 方法包含关键信息"""
        tag_group = TagGroup(domain="test_domain", name="test_group")
        policy = AdmissionPolicy(
            name="test_policy",
            category="test",
            enabled=True,
            block_msg="blocked",
            type="dynamic",
            calculate_alg=AlgInstance(alg_name="test_alg", alg_param={}),
        )
        wrapper = MatchWrapper(tag_group, policy)

        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = None
            controller = DynamicController(wrapper)

        repr_str = repr(controller)
        self.assertIn("test_group", repr_str)
        self.assertIn("test_policy", repr_str)


class TestAbsAdmissionControlllerInterface(unittest.TestCase):

    def test_allow_controller_is_instance(self):
        self.assertIsInstance(AllowController(), AbsAdmissionController)

    def test_deny_controller_is_instance(self):
        self.assertIsInstance(DenyController(), AbsAdmissionController)

    def test_dynamic_controller_is_instance(self):
        wrapper = _make_dynamic_wrapper()
        with patch(
            "olc.admission.admission_controller.OlcAlgManager.get_instance"
        ) as mock_mgr:
            mock_mgr.return_value.get_admission_alg.return_value = None
            self.assertIsInstance(DynamicController(wrapper), AbsAdmissionController)

    def test_controllers_do_not_share_state(self):
        allow = AllowController()
        deny = DenyController()
        self.assertFalse(allow.get_result())
        self.assertTrue(deny.get_result())


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