import unittest
from unittest.mock import patch, MagicMock
from olc.bean.match_wrapper import MatchWrapper
from olc.bean.olc_config_rule import FlowPolicy
from olc.bean.olc_control_request import OlcControlRequest
from olc.bean.tag_group import TagGroup, SubGroup
from olc.control.matcher.policy_matcher import (
PolicyMatcher,
AbstractPolicyMatcher,
FlowMatcher,
)
from olc.rule.olc_rule_manager import OlcRuleManager
class ConcretePolicyMatcher(AbstractPolicyMatcher):
def __init__(self, policy_return_value=None):
super().__init__()
self._policy_return_value = policy_return_value
def get_policy(self, name: str):
return self._policy_return_value
class TestPolicyMatcher(unittest.TestCase):
def test_policy_matcher_is_abstract(self):
with self.assertRaises(TypeError):
PolicyMatcher()
class TestAbstractPolicyMatcher(unittest.TestCase):
def test_match_empty_groups_returns_empty_list(self):
matcher = ConcretePolicyMatcher()
request = OlcControlRequest(tags={"env": "prod"})
result = matcher.match(request, [])
self.assertEqual(result, [])
def test_match_none_groups_returns_empty_list(self):
matcher = ConcretePolicyMatcher()
request = OlcControlRequest(tags={"env": "prod"})
result = matcher.match(request, None)
self.assertEqual(result, [])
def test_match_cache_hit_recomputes_result(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
wrapper = MatchWrapper(TagGroup(domain="test", name="group1"), policy)
matcher = ConcretePolicyMatcher(policy_return_value=policy)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
matcher._cache.get.return_value = [wrapper]
result = matcher.match(request, groups)
self.assertEqual(len(result), 1)
self.assertIsInstance(result[0], MatchWrapper)
self.assertEqual(result[0].tag_group.name, "group1")
self.assertEqual(result[0].policy.name, "flow_policy")
def test_match_cache_miss_returns_empty_list(self):
matcher = ConcretePolicyMatcher(policy_return_value=None)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
matcher._cache.get.return_value = None
result = matcher.match(request, groups)
self.assertEqual(len(result), 0)
def test_match_cache_hit_with_policy_returns_new_wrapper(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
matcher = ConcretePolicyMatcher(policy_return_value=policy)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
cached_wrapper = MatchWrapper(TagGroup(domain="test", name="group1"), policy)
matcher._cache.get.return_value = [cached_wrapper]
result = matcher.match(request, groups)
self.assertEqual(len(result), 1)
self.assertIsNot(result[0], cached_wrapper)
def test_match_with_policy_none_skips_group(self):
matcher = ConcretePolicyMatcher(policy_return_value=None)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock()]
result = matcher.match(request, groups)
self.assertEqual(len(result), 0)
def test_match_with_sub_group_uses_parent_name(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
parent_group = TagGroup(domain="test", name="parent_group")
sub_group = SubGroup.__new__(SubGroup)
sub_group.parent = parent_group
sub_group.name = "parent_group$$env@prod"
sub_group.domain = "test"
sub_group.priority = 0
sub_group.enabled = True
sub_group.tags = []
call_tracker = {"name": None}
class TrackingMatcher(AbstractPolicyMatcher):
def get_policy(self, name: str):
call_tracker["name"] = name
return policy
matcher = TrackingMatcher()
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock()]
request = OlcControlRequest(tags={"env": "prod"})
result = matcher.match(request, [sub_group])
self.assertEqual(call_tracker["name"], "parent_group")
self.assertEqual(len(result), 1)
def test_match_with_multiple_groups(self):
policy1 = FlowPolicy(
name="flow_policy1",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
policy2 = FlowPolicy(
name="flow_policy2",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=200,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
policies = {"group1": policy1, "group2": policy2}
class MultiPolicyMatcher(AbstractPolicyMatcher):
def get_policy(self, name: str):
return policies.get(name)
matcher = MultiPolicyMatcher()
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock(), MagicMock()]
request = OlcControlRequest(tags={"env": "prod"})
groups = [
TagGroup(domain="test", name="group1"),
TagGroup(domain="test", name="group2"),
]
result = matcher.match(request, groups)
self.assertEqual(len(result), 2)
self.assertEqual(result[0].policy.name, "flow_policy1")
self.assertEqual(result[1].policy.name, "flow_policy2")
def test_match_caches_non_empty_result(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
matcher = ConcretePolicyMatcher(policy_return_value=policy)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock()]
result = matcher.match(request, groups)
matcher._cache.put.assert_called_once()
def test_match_does_not_cache_empty_result(self):
matcher = ConcretePolicyMatcher(policy_return_value=None)
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock()]
result = matcher.match(request, groups)
matcher._cache.put.assert_called_once()
def test_match_mixed_policies_some_none(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
policies = {"group1": policy, "group2": None}
class MixedPolicyMatcher(AbstractPolicyMatcher):
def get_policy(self, name: str):
return policies.get(name)
matcher = MixedPolicyMatcher()
matcher._cache = MagicMock()
matcher._cache.get.return_value = [MagicMock(), MagicMock()]
request = OlcControlRequest(tags={"env": "prod"})
groups = [
TagGroup(domain="test", name="group1"),
TagGroup(domain="test", name="group2"),
]
result = matcher.match(request, groups)
self.assertEqual(len(result), 1)
self.assertEqual(result[0].tag_group.name, "group1")
def test_get_policy_is_abstract(self):
self.assertTrue(hasattr(AbstractPolicyMatcher, "get_policy"))
def test_init_creates_safe_ttl_cache(self):
from olc.cache.safe__ttl_cache import SafeTTLCache
matcher = ConcretePolicyMatcher()
self.assertIsInstance(matcher._cache, SafeTTLCache)
class TestFlowMatcher(unittest.TestCase):
def test_get_policy_calls_rule_manager(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
with patch.object(OlcRuleManager, "get_instance") as mock_mgr:
mock_mgr.return_value.get_flow_policy_by_group_name.return_value = (
policy
)
matcher = FlowMatcher()
result = matcher.get_policy("group1")
self.assertEqual(result, policy)
mock_mgr.return_value.get_flow_policy_by_group_name.assert_called_once_with(
"group1"
)
def test_get_policy_returns_none_for_nonexistent_group(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
with patch.object(OlcRuleManager, "get_instance") as mock_mgr:
mock_mgr.return_value.get_flow_policy_by_group_name.return_value = None
matcher = FlowMatcher()
result = matcher.get_policy("nonexistent")
self.assertIsNone(result)
def test_get_policy_returns_none_for_empty_name(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
with patch.object(OlcRuleManager, "get_instance") as mock_mgr:
mock_mgr.return_value.get_flow_policy_by_group_name.return_value = None
matcher = FlowMatcher()
result = matcher.get_policy("")
self.assertIsNone(result)
def test_clean_cache(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
matcher = FlowMatcher()
matcher._cache = MagicMock()
matcher.clean_cache()
matcher._cache.clear.assert_called_once()
def test_subscribe_clears_cache(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
matcher = FlowMatcher()
matcher._cache = MagicMock()
matcher.subscribe_clean(sender=None)
matcher._cache.clear.assert_called_once()
def test_flow_matcher_is_abstract_policy_matcher(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
matcher = FlowMatcher()
self.assertIsInstance(matcher, AbstractPolicyMatcher)
def test_flow_matcher_init_subscribes_to_clean_rule_event(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_signal = MagicMock()
mock_event.get_instance.return_value.olc_clean_rule = mock_signal
matcher = FlowMatcher()
mock_signal.connect.assert_called_once_with(matcher.subscribe_clean)
def test_flow_matcher_match_with_cached_result(self):
policy = FlowPolicy(
name="flow_policy",
category="flow",
enabled=True,
block_msg="",
policy_type="NODE",
time_unit="second",
time_interval=1,
rate_limit=100,
burst_limit=0,
flow_control_mode="qps",
max_wait_time_ms=0,
calculate_alg=None,
assign_alg=None,
)
wrapper = MatchWrapper(TagGroup(domain="test", name="group1"), policy)
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
with patch.object(OlcRuleManager, "get_instance") as mock_mgr:
mock_mgr.return_value.get_flow_policy_by_group_name.return_value = (
policy
)
matcher = FlowMatcher()
matcher._cache = MagicMock()
matcher._cache.get.return_value = [wrapper]
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
result = matcher.match(request, groups)
self.assertEqual(len(result), 1)
self.assertEqual(result[0].policy.name, "flow_policy")
def test_flow_matcher_match_cache_miss_returns_empty(self):
with patch("olc.control.matcher.policy_matcher.Event") as mock_event:
mock_event.get_instance.return_value.olc_clean_rule = MagicMock()
matcher = FlowMatcher()
matcher._cache = MagicMock()
matcher._cache.get.return_value = None
request = OlcControlRequest(tags={"env": "prod"})
groups = [TagGroup(domain="test", name="group1")]
result = matcher.match(request, groups)
self.assertEqual(len(result), 0)
if __name__ == "__main__":
unittest.main()