import unittest
import threading
from olc.statistic.statistic import OlcStatistic, IStatistic, StatisticSnapshot
class TestOlcStatistic(unittest.TestCase):
"""测试 OlcStatistic 的核心功能"""
def setUp(self):
self.statistic = OlcStatistic()
def test_initial_thread_count(self):
"""初始并发数应为 0"""
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_initial_request_metrics(self):
self.assertEqual(self.statistic.total_request_num(), 0)
self.assertEqual(self.statistic.pass_request_num(), 0)
self.assertEqual(self.statistic.block_request_num(), 0)
self.assertEqual(self.statistic.pass_avg_latency_ms(), 0.0)
self.assertEqual(self.statistic.pass_max_latency_ms(), 0.0)
def test_request_counts(self):
self.statistic.inc_pass_request_num()
self.statistic.inc_pass_request_num()
self.statistic.inc_pass_request_num()
self.statistic.inc_block_request_num()
self.assertEqual(self.statistic.total_request_num(), 4)
self.assertEqual(self.statistic.pass_request_num(), 3)
self.assertEqual(self.statistic.block_request_num(), 1)
def test_pass_latency_metrics(self):
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(10.5)
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(20.5)
self.assertEqual(self.statistic.pass_avg_latency_ms(), 15.5)
self.assertEqual(self.statistic.pass_max_latency_ms(), 20.5)
def test_negative_pass_latency_is_ignored(self):
self.statistic.record_pass_latency_ms(-1)
self.assertEqual(self.statistic.pass_avg_latency_ms(), 0.0)
self.assertEqual(self.statistic.pass_max_latency_ms(), 0.0)
def test_snapshot_returns_current_metrics(self):
self.statistic.inc_thread_num()
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(10)
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(10)
self.statistic.inc_pass_request_num()
self.statistic.inc_block_request_num()
self.statistic.record_pass_latency_ms(40.0)
snapshot = self.statistic.snapshot()
self.assertIsInstance(snapshot, StatisticSnapshot)
self.assertEqual(snapshot.cur_thread_num, 1)
self.assertEqual(snapshot.total_request_num_pre_second, 4)
self.assertEqual(snapshot.pass_request_num_pre_second, 3)
self.assertEqual(snapshot.block_request_num_pre_second, 1)
self.assertEqual(snapshot.pass_avg_latency_ms_pre_second, 20.0)
self.assertEqual(snapshot.pass_max_latency_ms_pre_second, 40.0)
def test_snapshot_contains_dual_granularity(self):
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(5.0)
self.statistic.inc_pass_request_num()
self.statistic.record_pass_latency_ms(5.0)
self.statistic.inc_block_request_num()
snapshot = self.statistic.snapshot()
self.assertEqual(snapshot.second_snap.total_request_num, 3)
self.assertEqual(snapshot.second_snap.pass_request_num, 2)
self.assertEqual(snapshot.second_snap.block_request_num, 1)
self.assertEqual(snapshot.second_snap.pass_avg_latency_ms, 5.0)
self.assertEqual(snapshot.second_snap.pass_max_latency_ms, 5.0)
self.assertEqual(snapshot.custom_snap.total_request_num, 3)
self.assertEqual(snapshot.custom_snap.pass_request_num, 2)
self.assertEqual(snapshot.custom_snap.block_request_num, 1)
self.assertEqual(snapshot.custom_snap.pass_avg_latency_ms, 5.0)
self.assertEqual(snapshot.custom_snap.pass_max_latency_ms, 5.0)
def test_inc_thread_num_once(self):
"""增加一次并发数"""
self.statistic.inc_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 1)
def test_inc_thread_num_multiple(self):
"""多次增加并发数"""
for _ in range(10):
self.statistic.inc_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 10)
def test_dec_thread_num_after_inc(self):
"""先增加后减少，并发数应正确递减"""
for _ in range(5):
self.statistic.inc_thread_num()
for _ in range(3):
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 2)
def test_dec_thread_num_to_zero(self):
"""递减到刚好 0"""
self.statistic.inc_thread_num()
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_dec_thread_num_below_zero_should_clamp(self):
"""从 0 开始递减，应被限制在 0，不会变为负数"""
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_dec_thread_num_from_1_to_0_below_0(self):
"""从 1 递减 2 次，结果应为 0（被限制）"""
self.statistic.inc_thread_num()
self.statistic.dec_thread_num()
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_dec_thread_num_negative_accumulated(self):
"""多次从 0 递减，结果应始终为 0"""
for _ in range(100):
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_large_inc_and_dec(self):
"""大量递增和递减后，并发数应正确"""
for _ in range(1000):
self.statistic.inc_thread_num()
for _ in range(500):
self.statistic.dec_thread_num()
self.assertEqual(self.statistic.cur_thread_num(), 500)
def test_concurrent_inc_thread_safety(self):
"""多线程并发递增，结果应正确累计"""
def worker(stat, count):
for _ in range(count):
stat.inc_thread_num()
threads = []
thread_count = 10
ops_per_thread = 100
for _ in range(thread_count):
t = threading.Thread(target=worker, args=(self.statistic, ops_per_thread))
threads.append(t)
t.start()
for t in threads:
t.join()
self.assertEqual(self.statistic.cur_thread_num(), thread_count * ops_per_thread)
def test_concurrent_request_metrics_thread_safety(self):
def worker(stat, count):
for _ in range(count):
stat.inc_pass_request_num()
stat.record_pass_latency_ms(1.5)
threads = []
thread_count = 10
ops_per_thread = 100
for _ in range(thread_count):
t = threading.Thread(target=worker, args=(self.statistic, ops_per_thread))
threads.append(t)
t.start()
for t in threads:
t.join()
total = thread_count * ops_per_thread
self.assertEqual(self.statistic.total_request_num(), total)
self.assertEqual(self.statistic.pass_request_num(), total)
self.assertEqual(self.statistic.block_request_num(), 0)
self.assertEqual(self.statistic.pass_avg_latency_ms(), 1.5)
self.assertEqual(self.statistic.pass_max_latency_ms(), 1.5)
def test_concurrent_mixed_ops_thread_safety(self):
"""多线程混合递增和递减，并发数不应低于 0"""
results = []
errors = []
def worker_inc(stat, count):
for _ in range(count):
stat.inc_thread_num()
def worker_dec(stat, count):
for _ in range(count):
stat.dec_thread_num()
threads = []
for _ in range(5):
threads.append(threading.Thread(target=worker_inc, args=(self.statistic, 200)))
threads.append(threading.Thread(target=worker_dec, args=(self.statistic, 200)))
for t in threads:
t.start()
for t in threads:
t.join()
final_count = self.statistic.cur_thread_num()
self.assertGreaterEqual(final_count, 0, "并发数不应低于 0")
def test_concurrent_dec_thread_safety(self):
"""多线程并发递减（从 0 开始），验证不会变为负数"""
threads = []
def worker(stat, count):
for _ in range(count):
stat.dec_thread_num()
for _ in range(10):
t = threading.Thread(target=worker, args=(self.statistic, 100))
threads.append(t)
t.start()
for t in threads:
t.join()
self.assertEqual(self.statistic.cur_thread_num(), 0)
def test_multiple_instances_independence(self):
"""多个 OlcStatistic 实例应相互独立"""
stat1 = OlcStatistic()
stat2 = OlcStatistic()
stat1.inc_thread_num()
stat1.inc_thread_num()
stat1.inc_thread_num()
stat2.inc_thread_num()
stat2.inc_thread_num()
self.assertEqual(stat1.cur_thread_num(), 3)
self.assertEqual(stat2.cur_thread_num(), 2)
def test_interface_abstract_methods(self):
"""IStatistic 抽象方法不可直接实例化"""
with self.assertRaises(TypeError):
IStatistic()
def test_inc_thread_num_return_type(self):
"""inc_thread_num 不应返回值"""
result = self.statistic.inc_thread_num()
self.assertIsNone(result)
def test_dec_thread_num_return_type(self):
"""dec_thread_num 不应返回值"""
result = self.statistic.dec_thread_num()
self.assertIsNone(result)
def test_cur_thread_num_return_type(self):
"""cur_thread_num 返回 int 类型"""
result = self.statistic.cur_thread_num()
self.assertIsInstance(result, int)
if __name__ == "__main__":
unittest.main()