import unittest
import threading
from olc.bean.tag_group import TagGroup
from olc.statistic.statistic import IStatistic, OlcStatistic, StatisticSnapshot
from olc.statistic.statistic_factory import OlcStatisticFactory
class TestOlcStatisticFactory(unittest.TestCase):
"""测试 OlcStatisticFactory 的核心功能"""
def setUp(self):
OlcStatisticFactory._instance = None
self.factory = OlcStatisticFactory.get_instance()
def tearDown(self):
OlcStatisticFactory._instance = None
def test_singleton_pattern(self):
"""单例模式：多次调用 get_instance 应返回同一实例"""
factory1 = OlcStatisticFactory.get_instance()
factory2 = OlcStatisticFactory.get_instance()
self.assertIs(factory1, factory2)
def test_get_statistic_returns_none_for_uncreated(self):
"""对于未创建的 tag_group，get_statistic 返回 None"""
tag_group = TagGroup(domain="test", name="nonexistent")
result = self.factory.get_statistic(tag_group)
self.assertIsNone(result)
def test_get_statistic_with_none_tag_group(self):
"""传入 None 的 tag_group，get_statistic 返回 None"""
result = self.factory.get_statistic(None)
self.assertIsNone(result)
def test_get_and_create_returns_statistic(self):
"""get_and_create 应返回 IStatistic 实例"""
tag_group = TagGroup(domain="test", name="create1")
result = self.factory.get_and_create(tag_group)
self.assertIsInstance(result, IStatistic)
self.assertIsInstance(result, OlcStatistic)
def test_get_and_create_creates_new_instance_per_tag_group(self):
"""不同的 tag_group 应创建不同的实例"""
tg1 = TagGroup(domain="a", name="1")
tg2 = TagGroup(domain="a", name="2")
stat1 = self.factory.get_and_create(tg1)
stat2 = self.factory.get_and_create(tg2)
self.assertIsNot(stat1, stat2)
def test_get_and_create_returns_same_instance_for_same_tag_group(self):
"""相同的 tag_group 应返回缓存的同一实例"""
tag_group = TagGroup(domain="test", name="same")
stat1 = self.factory.get_and_create(tag_group)
stat2 = self.factory.get_and_create(tag_group)
self.assertIs(stat1, stat2)
def test_get_statistic_returns_cached_instance(self):
"""get_and_create 后 get_statistic 应返回同一实例"""
tag_group = TagGroup(domain="test", name="cached")
created = self.factory.get_and_create(tag_group)
retrieved = self.factory.get_statistic(tag_group)
self.assertIs(created, retrieved)
def test_get_and_create_with_none_tag_group_raises(self):
"""get_and_create 传入 None 应抛出 ValueError"""
with self.assertRaises(ValueError) as ctx:
self.factory.get_and_create(None)
self.assertIn("tag_group", str(ctx.exception))
def test_get_and_create_idempotent(self):
"""多次 get_and_create 同一 tag_group 应返回同一实例"""
tag_group = TagGroup(domain="test", name="idempotent")
results = [self.factory.get_and_create(tag_group) for _ in range(10)]
for i in range(1, len(results)):
self.assertIs(results[0], results[i])
def test_tag_group_equality_as_cache_key(self):
"""相同的 domain + name 的 TagGroup 应命中同一缓存"""
tg1 = TagGroup(domain="test", name="eq")
tg2 = TagGroup(domain="test", name="eq")
stat1 = self.factory.get_and_create(tg1)
stat2 = self.factory.get_and_create(tg2)
self.assertIs(stat1, stat2)
def test_tag_group_different_domain_same_name(self):
"""不同 domain 但相同 name 的 TagGroup 应视为不同的缓存项"""
tg1 = TagGroup(domain="a", name="same_name")
tg2 = TagGroup(domain="b", name="same_name")
stat1 = self.factory.get_and_create(tg1)
stat2 = self.factory.get_and_create(tg2)
self.assertIsNot(stat1, stat2)
def test_subscribe_clean_clears_cache(self):
"""subscribe_clean 应清空所有缓存"""
tg1 = TagGroup(domain="test", name="clean1")
tg2 = TagGroup(domain="test", name="clean2")
self.factory.get_and_create(tg1)
self.factory.get_and_create(tg2)
self.factory.subscribe_clean(None)
self.assertIsNone(self.factory.get_statistic(tg1))
self.assertIsNone(self.factory.get_statistic(tg2))
def test_after_clean_get_and_create_works(self):
"""清空后再次 get_and_create 应创建新实例"""
tag_group = TagGroup(domain="test", name="recreate")
stat1 = self.factory.get_and_create(tag_group)
self.factory.subscribe_clean(None)
stat2 = self.factory.get_and_create(tag_group)
self.assertIsNot(stat1, stat2)
self.assertIsInstance(stat2, OlcStatistic)
def test_statistic_operations_after_factory_retrieval(self):
"""通过工厂获取的统计实例应能正常操作"""
tag_group = TagGroup(domain="test", name="ops")
stat = self.factory.get_and_create(tag_group)
stat.inc_thread_num()
stat.inc_thread_num()
stat.dec_thread_num()
self.assertEqual(stat.cur_thread_num(), 1)
stat.dec_thread_num()
stat.dec_thread_num()
self.assertEqual(stat.cur_thread_num(), 0)
def test_concurrent_get_and_create_same_tag_group(self):
"""多线程并发 get_and_create 同一 tag_group，应只创建一个实例"""
tag_group = TagGroup(domain="test", name="concurrent_create")
results = []
def worker():
for _ in range(50):
stat = self.factory.get_and_create(tag_group)
results.append(stat)
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
first = results[0]
for r in results:
self.assertIs(r, first)
def test_concurrent_get_and_create_different_tag_groups(self):
"""多线程并发创建不同 tag_group，所有实例应不同"""
results = {}
def worker(domain_suffix):
tg = TagGroup(domain="test", name=f"concurrent_{domain_suffix}")
stat = self.factory.get_and_create(tg)
results[domain_suffix] = stat
threads = [threading.Thread(target=worker, args=(i,)) for i in range(50)]
for t in threads:
t.start()
for t in threads:
t.join()
unique_instances = set(id(v) for v in results.values())
self.assertEqual(len(unique_instances), 50)
def test_cache_ttl_eviction(self):
"""等待 TTL 过期后，缓存的统计实例应被清理"""
tag_group = TagGroup(domain="test", name="ttl")
old_factory = OlcStatisticFactory.get_instance()
stat = old_factory.get_and_create(tag_group)
retrieved = old_factory.get_statistic(tag_group)
self.assertIs(stat, retrieved)
def test_get_statistic_does_not_create(self):
"""get_statistic 不应创建新的统计实例"""
tag_group = TagGroup(domain="test", name="nope")
result = self.factory.get_statistic(tag_group)
self.assertIsNone(result)
def test_get_all_snapshots_returns_tag_group_snapshot_dict(self):
"""get_all_snapshots 应返回 TagGroup 到 StatisticSnapshot 的映射"""
tag_group1 = TagGroup(domain="test", name="snap1")
tag_group2 = TagGroup(domain="test", name="snap2")
statistic1 = self.factory.get_and_create(tag_group1)
statistic2 = self.factory.get_and_create(tag_group2)
statistic1.inc_thread_num()
statistic1.inc_pass_request_num()
statistic1.record_pass_latency_ms(12.0)
statistic2.inc_block_request_num()
snapshots = self.factory.get_all_snapshots()
self.assertIsInstance(snapshots, dict)
self.assertIn(tag_group1, snapshots)
self.assertIn(tag_group2, snapshots)
self.assertIsInstance(snapshots[tag_group1], StatisticSnapshot)
self.assertIsInstance(snapshots[tag_group2], StatisticSnapshot)
self.assertEqual(snapshots[tag_group1].cur_thread_num, 1)
self.assertEqual(snapshots[tag_group1].total_request_num_pre_second, 1)
self.assertEqual(snapshots[tag_group1].pass_request_num_pre_second, 1)
self.assertEqual(snapshots[tag_group1].block_request_num_pre_second, 0)
self.assertEqual(snapshots[tag_group2].total_request_num_pre_second, 1)
self.assertEqual(snapshots[tag_group2].pass_request_num_pre_second, 0)
self.assertEqual(snapshots[tag_group2].block_request_num_pre_second, 1)
if __name__ == "__main__":
unittest.main()