50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
|
"""cache_service 缓存单元测试（内存后端，无需 Redis）。"""
|
||
|
|
import time
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from app.service.cache_service import CacheService, InMemoryBackend
|
||
|
|
|
||
|
|
|
||
|
|
class TestCacheService(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.svc = CacheService(backend=InMemoryBackend())
|
||
|
|
|
||
|
|
def test_sql_hash_deterministic(self):
|
||
|
|
self.assertEqual(self.svc.sql_hash("SELECT 1"), self.svc.sql_hash("SELECT 1"))
|
||
|
|
self.assertNotEqual(self.svc.sql_hash("SELECT 1"), self.svc.sql_hash("SELECT 2"))
|
||
|
|
|
||
|
|
def test_permission_fingerprint_differs_by_scope(self):
|
||
|
|
a = self.svc.permission_fingerprint("S1", "assigned", ["CUST-1"])
|
||
|
|
b = self.svc.permission_fingerprint("S1", "assigned", ["CUST-2"])
|
||
|
|
self.assertNotEqual(a, b)
|
||
|
|
|
||
|
|
def test_ttl_layering(self):
|
||
|
|
self.assertEqual(self.svc.ttl_for(["core_trade"]), 5 * 60)
|
||
|
|
self.assertEqual(self.svc.ttl_for(["risk_alert"]), 60 * 60)
|
||
|
|
self.assertEqual(self.svc.ttl_for(["core_customer"]), 24 * 60 * 60)
|
||
|
|
|
||
|
|
def test_roundtrip(self):
|
||
|
|
fp = self.svc.permission_fingerprint("S1", "full")
|
||
|
|
sql = "SELECT COUNT(*) FROM core_customer"
|
||
|
|
self.assertIsNone(self.svc.get_result(fp, sql))
|
||
|
|
self.svc.set_result(fp, sql, {"cnt": 33}, ["core_customer"])
|
||
|
|
self.assertEqual(self.svc.get_result(fp, sql), {"cnt": 33})
|
||
|
|
|
||
|
|
def test_invalidate(self):
|
||
|
|
fp = self.svc.permission_fingerprint("S1", "full")
|
||
|
|
sql = "SELECT 1"
|
||
|
|
self.svc.set_result(fp, sql, {"x": 1}, ["core_customer"])
|
||
|
|
self.svc.invalidate_by_sql(fp, sql)
|
||
|
|
self.assertIsNone(self.svc.get_result(fp, sql))
|
||
|
|
|
||
|
|
def test_inmemory_expiry(self):
|
||
|
|
b = InMemoryBackend()
|
||
|
|
b.set("k", "v", ttl=1)
|
||
|
|
self.assertEqual(b.get("k"), "v")
|
||
|
|
time.sleep(1.2)
|
||
|
|
self.assertIsNone(b.get("k"))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|