wehub-resource-sync
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""Regression tests for the per-engine SQLite bootstrap lock cache.
|
|
|
|
The cache (``deerflow.persistence.bootstrap._SQLITE_LOCKS``) maps an engine
|
|
to the ``asyncio.Lock`` that serialises its in-process bootstrap. It is keyed
|
|
by the engine object itself via ``WeakKeyDictionary`` -- not ``id(engine)`` --
|
|
to avoid two failure modes that are silent in production (one long-lived
|
|
engine) but real in pytest (one fresh engine per test):
|
|
|
|
1. **CPython id reuse.** After an engine is garbage-collected its memory
|
|
address can be reused by a new engine. An ``id``-keyed cache would hand
|
|
the new engine the dead engine's ``Lock``. That lock was bound to the
|
|
dead engine's event loop at first ``async with``; pytest gives each async
|
|
test its own loop, so reusing it raises ``RuntimeError: ... bound to a
|
|
different event loop``.
|
|
2. **Unbounded growth.** An ``id``-keyed cache never drops entries because
|
|
nothing notifies it when the engine dies. With ``WeakKeyDictionary`` the
|
|
entry disappears as soon as the engine is collected.
|
|
|
|
These tests do not open any DB connection -- they exercise the cache helper
|
|
directly so they can run without an event loop and without aiosqlite warnings
|
|
about unclosed engines.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import weakref
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from deerflow.persistence import bootstrap as bootstrap_mod
|
|
from deerflow.persistence.bootstrap import _get_sqlite_local_lock
|
|
|
|
|
|
def _make_engine():
|
|
return create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
|
|
|
|
def test_cache_is_weak_key_dictionary() -> None:
|
|
"""Pin the cache type so a refactor cannot silently revert to a plain
|
|
dict (which would reintroduce the id-reuse bug)."""
|
|
assert isinstance(bootstrap_mod._SQLITE_LOCKS, weakref.WeakKeyDictionary)
|
|
|
|
|
|
def test_same_engine_returns_same_lock() -> None:
|
|
engine = _make_engine()
|
|
assert _get_sqlite_local_lock(engine) is _get_sqlite_local_lock(engine)
|
|
|
|
|
|
def test_distinct_engines_get_distinct_locks() -> None:
|
|
"""Two live engines must not share a lock -- otherwise unrelated
|
|
bootstraps would serialise against each other."""
|
|
engine_a = _make_engine()
|
|
engine_b = _make_engine()
|
|
assert _get_sqlite_local_lock(engine_a) is not _get_sqlite_local_lock(engine_b)
|
|
|
|
|
|
def test_entry_drops_when_engine_is_garbage_collected() -> None:
|
|
"""The cache must not pin the engine alive.
|
|
|
|
This is the structural guarantee behind the id-reuse fix: when the engine
|
|
is collected, its lock entry goes with it, so a future engine landing on
|
|
the same address cannot inherit a stale, loop-bound lock.
|
|
"""
|
|
engine = _make_engine()
|
|
_get_sqlite_local_lock(engine)
|
|
assert engine in bootstrap_mod._SQLITE_LOCKS
|
|
|
|
engine_ref = weakref.ref(engine)
|
|
del engine
|
|
gc.collect()
|
|
|
|
assert engine_ref() is None, "engine should be collectible -- cache must not hold a strong ref"
|
|
# WeakKeyDictionary may defer removal until the next access; touch it.
|
|
assert all(ref() is not None for ref in bootstrap_mod._SQLITE_LOCKS.keyrefs())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fresh_engine_gets_lock_usable_on_current_loop() -> None:
|
|
"""End-to-end guard for the pytest pattern: a brand-new engine in a
|
|
brand-new event loop must receive a lock that ``async with`` accepts.
|
|
|
|
This is the behaviour an ``id``-keyed cache could break if the new engine
|
|
landed on a previously-used address -- it would return a lock bound to a
|
|
dead loop and raise ``RuntimeError: ... bound to a different event loop``.
|
|
"""
|
|
engine = _make_engine()
|
|
try:
|
|
lock = _get_sqlite_local_lock(engine)
|
|
async with lock:
|
|
pass
|
|
# Re-entrant acquire on the same loop must also succeed.
|
|
async with lock:
|
|
pass
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_does_not_grow_across_disposed_engines() -> None:
|
|
"""Create + dispose + drop many engines and assert the cache stays bounded.
|
|
|
|
Without ``WeakKeyDictionary`` this loop would leak one entry per engine.
|
|
"""
|
|
initial = len(bootstrap_mod._SQLITE_LOCKS)
|
|
for _ in range(20):
|
|
engine = _make_engine()
|
|
_get_sqlite_local_lock(engine)
|
|
await engine.dispose()
|
|
del engine
|
|
gc.collect()
|
|
# Touch the dict so WeakKeyDictionary clears any deferred removals.
|
|
_ = list(bootstrap_mod._SQLITE_LOCKS.items())
|
|
# Allow a small slack for any engine that is still pinned by a frame.
|
|
assert len(bootstrap_mod._SQLITE_LOCKS) - initial <= 1
|