zhanghongyu_0626
bcd6175d4e
- Introduced `analyst_auth_adapter.py` for managing authentication context and access control for the data analysis agent. - Added new API endpoints in `analyst.py` for chat, dashboard, asset management, and metrics, utilizing the new authentication context. - Created Pydantic models in `analyst_schemas.py` for request and response structures, ensuring consistent data handling. - Updated SQL guard logic in `sql_guard.py` to enforce access restrictions based on user roles and contexts. - Implemented migration scripts for new database tables related to the data analysis agent, enhancing data management capabilities. - Removed legacy authentication code from `auth.py`, streamlining the authentication process. This update significantly enhances the data analysis capabilities, providing a robust framework for querying and managing data securely.
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()
|