68 lines
1.8 KiB
Python
68 lines
1.8 KiB
Python
"""New threads must not scan legacy SQLite sources on empty primary reads."""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
from deerflow.runtime.checkpointer.checkpoint_migration import DualReadCheckpointer
|
|
|
|
|
|
class _Primary:
|
|
def __init__(self):
|
|
self.tuple_calls = 0
|
|
self.list_calls = 0
|
|
|
|
async def aget_tuple(self, config):
|
|
self.tuple_calls += 1
|
|
return None
|
|
|
|
async def aget(self, config):
|
|
return None
|
|
|
|
async def alist(self, config, *, filter=None, before=None, limit=None):
|
|
self.list_calls += 1
|
|
if False:
|
|
yield None
|
|
|
|
async def aput(self, *args, **kwargs):
|
|
return None
|
|
|
|
|
|
class _Legacy:
|
|
def __init__(self):
|
|
self.tuple_calls = 0
|
|
self.list_calls = 0
|
|
self.closed = False
|
|
|
|
async def aget_tuple(self, config):
|
|
self.tuple_calls += 1
|
|
return {"legacy": True}
|
|
|
|
async def alist(self, config, *, filter=None, before=None, limit=None):
|
|
self.list_calls += 1
|
|
if False:
|
|
yield None
|
|
|
|
|
|
def test_remembered_thread_skips_empty_sqlite_fallback():
|
|
async def run():
|
|
primary = _Primary()
|
|
legacy = _Legacy()
|
|
saver = DualReadCheckpointer(primary, [SimpleNamespace(saver=legacy, closed=False)])
|
|
config = {"configurable": {"thread_id": "new-hello"}}
|
|
|
|
assert await saver.aget_tuple(config) == {"legacy": True}
|
|
assert legacy.tuple_calls == 1
|
|
|
|
saver.remember_primary_thread("new-hello")
|
|
assert await saver.aget_tuple(config) is None
|
|
assert legacy.tuple_calls == 1
|
|
|
|
chunks = [item async for item in saver.alist(config, limit=1)]
|
|
assert chunks == []
|
|
assert legacy.list_calls == 0
|
|
|
|
await saver.aput(config, {}, {}, {})
|
|
assert saver._skip_legacy_fallback(config) is True
|
|
|
|
asyncio.run(run())
|