wehub-resource-sync
76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
"""Regression: ``bootstrap_schema`` offloads ``alembic.command.stamp`` /
|
|
``alembic.command.upgrade`` via ``asyncio.to_thread``.
|
|
|
|
The alembic commands are synchronous: they open their own engine and execute
|
|
DDL. Calling them directly on the FastAPI lifespan event loop would block --
|
|
exactly the failure mode of the issue chain that motivated the hybrid
|
|
bootstrap (sync IO on the loop = silent stalls / timeouts).
|
|
|
|
Anchor strategy
|
|
---------------
|
|
|
|
We can't run a real ``init_engine(backend="sqlite", ...)`` under the strict
|
|
Blockbuster gate without tripping on ``create_async_engine``'s own
|
|
``os.path.abspath`` (which is a pre-existing concern, not the bootstrap's).
|
|
The companion ``test_persistence_engine_sqlite.py`` covers the ``init_engine``
|
|
makedirs offload by mocking ``create_async_engine`` away entirely. That same
|
|
mocking approach would defeat the point here, because the alembic stamp /
|
|
upgrade calls in ``bootstrap_schema`` need a *real* on-disk SQLite DB to
|
|
exercise.
|
|
|
|
So this test installs a spy on ``asyncio.to_thread`` and confirms that the
|
|
two alembic entry points -- ``_stamp`` and ``_upgrade`` from
|
|
``bootstrap_schema`` -- are dispatched through it, not invoked inline. If a
|
|
future refactor inlines either call, the spy records zero invocations for
|
|
that function and the assertion fails.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
import deerflow.persistence.models # noqa: F401
|
|
from deerflow.persistence import bootstrap as bootstrap_mod
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
@pytest.mark.allow_blocking_io
|
|
async def test_bootstrap_offloads_alembic_stamp_and_upgrade(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
"""Stamp + upgrade must go through ``asyncio.to_thread``.
|
|
|
|
Marked ``allow_blocking_io`` so the strict Blockbuster gate does not flag
|
|
incidental blocking IO in test-fixture setup (engine creation paths,
|
|
SQLite path resolution). The point of this test is the
|
|
``asyncio.to_thread`` wrapping invariant, which the spy below checks
|
|
deterministically.
|
|
"""
|
|
seen: list[str] = []
|
|
|
|
original_to_thread = asyncio.to_thread
|
|
|
|
async def spy_to_thread(func, *args, **kwargs):
|
|
seen.append(getattr(func, "__name__", repr(func)))
|
|
return await original_to_thread(func, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(bootstrap_mod.asyncio, "to_thread", spy_to_thread)
|
|
|
|
# Use a real SQLite DB so alembic actually runs stamp + upgrade.
|
|
db_path = tmp_path / "spy.db"
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{db_path.as_posix()}")
|
|
try:
|
|
# Empty branch -> create_all + stamp head. ``_stamp`` must be offloaded.
|
|
await bootstrap_mod.bootstrap_schema(engine, backend="sqlite")
|
|
assert "_stamp" in seen, f"_stamp not offloaded; saw: {seen}"
|
|
|
|
# Re-run -> versioned branch -> upgrade head (no-op at head). ``_upgrade`` must be offloaded.
|
|
seen.clear()
|
|
await bootstrap_mod.bootstrap_schema(engine, backend="sqlite")
|
|
assert "_upgrade" in seen, f"_upgrade not offloaded; saw: {seen}"
|
|
finally:
|
|
await engine.dispose()
|