import unittest
import threading
from unittest.mock import MagicMock, patch
from olc.admission.admission_factory import AdmissionFactory
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, AdmissionType, AlgInstance
from olc.bean.tag_group import TagGroup
from olc.event.sny_event import Event
def _make_tag_group(name: str, domain: str = "test_domain") -> TagGroup:
return TagGroup(domain=domain, name=name, enabled=True, priority=500)
def _make_admission_policy(
name: str,
policy_type: str = "allow",
category: str = "test",
enabled: bool = True,
block_msg: str = "",
) -> AdmissionPolicy:
return AdmissionPolicy(
name=name,
category=category,
enabled=enabled,
block_msg=block_msg,
type=policy_type,
calculate_alg=None,
)
def _make_wrapper(tag_group: TagGroup, policy: AdmissionPolicy) -> MatchWrapper[AdmissionPolicy]:
return MatchWrapper(tag_group, policy)
class TestAdmissionFactory(unittest.TestCase):
def setUp(self):
AdmissionFactory._instance = None
def tearDown(self):
AdmissionFactory._instance = None
def test_singleton_pattern(self):
factory1 = AdmissionFactory.get_instance()
factory2 = AdmissionFactory.get_instance()
self.assertIs(factory1, factory2)
def test_create_allow_controller(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("allow_policy", "allow")
tag_group = _make_tag_group("allow_group")
wrapper = _make_wrapper(tag_group, policy)
controller = factory.get_controller(wrapper)
self.assertIsNotNone(controller)
self.assertIsInstance(controller, AllowController)
def test_create_deny_controller(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("deny_policy", "deny")
tag_group = _make_tag_group("deny_group")
wrapper = _make_wrapper(tag_group, policy)
controller = factory.get_controller(wrapper)
self.assertIsNotNone(controller)
self.assertIsInstance(controller, DenyController)
def test_create_dynamic_controller(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("dynamic_policy", "dynamic")
policy.calculate_alg = AlgInstance(alg_name="test_alg", alg_param={})
tag_group = _make_tag_group("dynamic_group")
wrapper = _make_wrapper(tag_group, policy)
with patch(
"olc.admission.admission_controller.OlcAlgManager.get_instance"
) as mock_mgr:
mock_alg = MagicMock()
mock_alg.every_request = False
mock_mgr.return_value.get_admission_alg.return_value = mock_alg
controller = factory.get_controller(wrapper)
self.assertIsNotNone(controller)
self.assertIsInstance(controller, DynamicController)
def test_unknown_type_defaults_to_allow(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("unknown_policy", "unknown_type")
tag_group = _make_tag_group("unknown_group")
wrapper = _make_wrapper(tag_group, policy)
controller = factory.get_controller(wrapper)
self.assertIsNotNone(controller)
self.assertIsInstance(controller, AllowController)
def test_case_insensitive_type(self):
factory = AdmissionFactory.get_instance()
for type_name in ["DENY", "Deny", "deny", "DeNy"]:
policy = _make_admission_policy(f"case_test_{type_name}", type_name)
tag_group = _make_tag_group(f"case_group_{type_name}")
wrapper = _make_wrapper(tag_group, policy)
controller = factory.get_controller(wrapper)
self.assertIsInstance(controller, DenyController)
def test_cache_reuse_same_tag_group(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("cached_policy", "allow")
tag_group = _make_tag_group("cached_group")
wrapper = _make_wrapper(tag_group, policy)
controller1 = factory.get_controller(wrapper)
controller2 = factory.get_controller(wrapper)
self.assertIs(controller1, controller2)
def test_different_tag_groups_create_different_controllers(self):
factory = AdmissionFactory.get_instance()
policy_allow = _make_admission_policy("allow_policy_diff", "allow")
policy_deny = _make_admission_policy("deny_policy_diff", "deny")
tag_group1 = _make_tag_group("group1")
tag_group2 = _make_tag_group("group2")
wrapper1 = _make_wrapper(tag_group1, policy_allow)
wrapper2 = _make_wrapper(tag_group2, policy_deny)
controller1 = factory.get_controller(wrapper1)
controller2 = factory.get_controller(wrapper2)
self.assertIsNot(controller1, controller2)
self.assertIsInstance(controller1, AllowController)
self.assertIsInstance(controller2, DenyController)
def test_different_domains_same_name_is_different(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("cross_domain_policy", "allow")
tag_group1 = _make_tag_group("same_name", "domain1")
tag_group2 = _make_tag_group("same_name", "domain2")
wrapper1 = _make_wrapper(tag_group1, policy)
wrapper2 = _make_wrapper(tag_group2, policy)
controller1 = factory.get_controller(wrapper1)
controller2 = factory.get_controller(wrapper2)
self.assertIsNot(controller1, controller2)
def test_get_all_controller_returns_list(self):
factory = AdmissionFactory.get_instance()
controllers = factory.get_all_controller()
self.assertIsInstance(controllers, list)
def test_subscribe_clean_clears_cache(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("clean_policy", "allow")
tag_group = _make_tag_group("clean_group")
wrapper = _make_wrapper(tag_group, policy)
controller1 = factory.get_controller(wrapper)
self.assertIsNotNone(controller1)
Event.get_instance().olc_clean_rule.send(None)
controller2 = factory.get_controller(wrapper)
self.assertIsNotNone(controller2)
self.assertIsNot(controller1, controller2)
def test_concurrent_get_controller_same_tag_group(self):
factory = AdmissionFactory.get_instance()
policy = _make_admission_policy("concurrent_policy", "allow")
tag_group = _make_tag_group("concurrent_group")
wrapper = _make_wrapper(tag_group, policy)
results = []
def worker():
controller = factory.get_controller(wrapper)
results.append(controller)
threads = [threading.Thread(target=worker) for _ in range(20)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(len(results), 20)
first = results[0]
for r in results:
self.assertIs(r, first)
def test_concurrent_get_controller_different_tag_groups(self):
factory = AdmissionFactory.get_instance()
results = {}
def worker(name: str):
policy = _make_admission_policy(f"policy_{name}", "allow")
tag_group = _make_tag_group(name)
wrapper = _make_wrapper(tag_group, policy)
controller = factory.get_controller(wrapper)
results[name] = controller
threads = [threading.Thread(target=worker, args=(f"group_{i}",)) for i in range(20)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(len(results), 20)
unique = set(id(v) for v in results.values())
self.assertEqual(len(unique), 20)
if __name__ == "__main__":
unittest.main()