12a4d9584e
Stabilize cross-protocol ask-mode/streaming behavior and reduce session-reuse branch collisions, then add focused docs/tests for multimodal normalization and pool/stats/config paths. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from app.session_cache import SessionCache, hash_branch_context, hash_user_context
|
|
|
|
|
|
class SessionCacheToolingTests(unittest.IsolatedAsyncioTestCase):
|
|
def test_hash_user_context_ignores_assistant_and_tool(self) -> None:
|
|
base = [
|
|
{"role": "system", "content": "S"},
|
|
{"role": "user", "content": "U"},
|
|
]
|
|
with_extra = base + [
|
|
{"role": "assistant", "content": "A1"},
|
|
{"role": "tool", "content": "T1"},
|
|
]
|
|
self.assertEqual(hash_user_context(base), hash_user_context(with_extra))
|
|
|
|
def test_hash_branch_context_distinguishes_assistant_tool_branch(self) -> None:
|
|
base = [
|
|
{"role": "system", "content": "S"},
|
|
{"role": "user", "content": "U"},
|
|
{"role": "assistant", "content": "A1"},
|
|
{"role": "tool", "content": "T1", "tool_call_id": "call-1"},
|
|
]
|
|
changed = [
|
|
{"role": "system", "content": "S"},
|
|
{"role": "user", "content": "U"},
|
|
{"role": "assistant", "content": "A2"},
|
|
{"role": "tool", "content": "T1", "tool_call_id": "call-1"},
|
|
]
|
|
self.assertNotEqual(hash_branch_context(base), hash_branch_context(changed))
|
|
|
|
def test_build_key_changes_with_tool_config(self) -> None:
|
|
cache = SessionCache(max_entries=8, ttl_sec=60)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
key1 = cache.build_key("k", msgs, tool_config={"a": 1, "b": 2})
|
|
key2 = cache.build_key("k", msgs, tool_config={"b": 2, "a": 1})
|
|
key3 = cache.build_key("k", msgs, tool_config={"a": 1})
|
|
self.assertEqual(key1, key2)
|
|
self.assertNotEqual(key1, key3)
|
|
|
|
def test_build_key_keeps_legacy_shape_without_branch_context(self) -> None:
|
|
cache = SessionCache(max_entries=8, ttl_sec=60)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
legacy = cache.build_key("k", msgs)
|
|
with_branch = cache.build_key("k", msgs, branch_context="abc")
|
|
self.assertEqual(legacy.count(":"), 2)
|
|
self.assertEqual(with_branch.count(":"), 3)
|
|
|
|
async def test_lru_evicts_oldest(self) -> None:
|
|
cache = SessionCache(max_entries=2, ttl_sec=600)
|
|
await cache.put("k1", "s1")
|
|
await cache.put("k2", "s2")
|
|
await cache.put("k3", "s3")
|
|
self.assertIsNone(await cache.get("k1"))
|
|
self.assertEqual(cache.evict_total, 1)
|
|
|
|
async def test_ttl_expiry_increments_expire_counter(self) -> None:
|
|
cache = SessionCache(max_entries=4, ttl_sec=0.001)
|
|
await cache.put("k1", "s1")
|
|
await __import__("asyncio").sleep(0.01)
|
|
self.assertIsNone(await cache.get("k1"))
|
|
self.assertEqual(cache.expire_total, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|