mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-21 10:15:47 +00:00
* feat(persistence): wire alembic migrations + bootstrap schema on startup Closes #3682. Pre-#3658 DBs lack the `runs.token_usage_by_model` column because alembic was never wired up — startup only ran `create_all`, which never ALTERs existing tables. Adds a hybrid bootstrap in FastAPI lifespan (replaces bare `create_all`): - empty DB → create_all + stamp head - legacy DB → stamp 0001_baseline + upgrade head - versioned DB → upgrade head Concurrency: Postgres `pg_advisory_lock` (cross-process); SQLite per-engine `asyncio.Lock` + 30s `PRAGMA busy_timeout` on both prod and alembic engines. Column revisions use `safe_add_column` / `safe_drop_column` idempotent helpers as fallback. Other bits: - 0001 baseline (chain root) + 0002 add `runs.token_usage_by_model` - `include_object` filter so alembic ignores LangGraph checkpointer tables - `make migrate-rev MSG="..."` for authoring new revisions (no migrate/stamp targets — startup is the only execution path) - Tests: three-branch decision, concurrency, #3682 regression, env filter, blocking-IO gate anchor - CLAUDE.md: new "Schema migrations" section * fix(style): fix lint error * perf(persistence): address review feedback on alembic bootstrap Behavioural fixes - _SQLITE_LOCKS now keyed via WeakKeyDictionary so id-reuse after GC cannot return a stale, loop-bound lock and the cache cannot leak one entry per disposed engine. - safe_add_column compares nullable / server_default against the desired column when the name already exists and emits a warning on drift, surfacing manual-ALTER workarounds instead of silently no-op'ing. - _postgres_lock issues SET LOCAL idle_in_transaction_session_timeout=0 before pg_advisory_lock, so managed Postgres cannot kill the idle lock-holding session mid-upgrade and silently release the advisory lock. - legacy branch now backfills missing baseline tables via a restricted create_all (Base.metadata.create_all scoped to _BASELINE_TABLE_NAMES). Restores pre-#1930 upgraders whose channel_* tables were never provisioned, without pre-empting future create_table revisions for newly-added models. Schema parity - runs.token_usage_by_model gains server_default=text("'{}'") in both the ORM model and the 0001_baseline create_table, matching what 0002 adds via ALTER. create_all and alembic-upgrade paths now produce identical column definitions. - New parity test compares Base.metadata.create_all output against a pure alembic upgrade base->head, asserting column-set, nullable, and server_default agree across all tables (normalized through the same helper safe_add_column's drift check uses). Guards - test_baseline_table_names_constant_matches_0001 pins _BASELINE_TABLE_NAMES to 0001_baseline.upgrade()'s actual output -- the constant cannot drift silently when someone edits 0001. - test_legacy_backfill_skips_non_baseline_tables verifies the restricted backfill does not create a phantom table on Base.metadata, modelling a future revision that would otherwise collide on op.create_table. Doc residuals - Three-branch decision table is now consistent across bootstrap.py top docstring, engine.py comment, test module docstring, and CLAUDE.md. - Stale test anchor in blocking_io/test_persistence_engine_sqlite.py docstring now points at the real file. * fix(style): fix lint error * fix(persistence): close drift detection holes - _check_column_drift compares column type via a family equivalence allowlist ({JSON, JSONB}). Catches the wrong-type workaround `TEXT NOT NULL DEFAULT '{}'` that previously slipped through silently, while keeping Postgres JSON/JSONB dialect reflection quiet. Reflected and desired type are also echoed in every drift warning's payload for operator triage. - Extract _escape_url_for_alembic so bootstrap._alembic_safe_url and scripts/_autogen_revision share the ConfigParser % escape rule instead of duplicating it. - backend/README.md: add `make migrate-rev MSG=...` to Commands and a Schema Migrations section per the repo's README/CLAUDE.md sync policy. - test_base_to_dict.py: scope the test ORM class to an isolated MetaData so the create_all-vs-alembic parity test (added in the previous commit) is not polluted by the phantom table on the full pytest session.
637 lines
29 KiB
Python
637 lines
29 KiB
Python
"""Tests for ``deerflow.persistence.bootstrap.bootstrap_schema``.
|
||
|
||
Covers the three-branch decision table:
|
||
|
||
| DB state | Action |
|
||
|---------------------------------------|-----------------------------------------|
|
||
| empty | create_all + stamp head |
|
||
| legacy (DeerFlow tables, no alembic_version) | create_all (baseline tables only, backfill) + stamp baseline + upgrade head |
|
||
| versioned | upgrade head |
|
||
|
||
Each test seeds a temp SQLite to the relevant pre-state, runs
|
||
``bootstrap_schema``, and asserts both the resulting schema and the
|
||
``alembic_version`` row.
|
||
|
||
The legacy branch is exercised across three scenarios: token-usage column
|
||
missing, token-usage column already present, and a baseline-era table
|
||
missing entirely (the ``channel_*`` backfill case). The first two prove the
|
||
column-level idempotent helpers handle both sub-cases; the third proves the
|
||
table-level backfill works.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
import sqlalchemy as sa
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
||
# Pre-import models so Base.metadata is populated before bootstrap reads it.
|
||
import deerflow.persistence.models # noqa: F401
|
||
from deerflow.persistence.base import Base
|
||
from deerflow.persistence.bootstrap import (
|
||
_BASELINE_TABLE_NAMES,
|
||
_decide_state,
|
||
_get_alembic_config,
|
||
_get_head_revision,
|
||
_run_baseline_create_all_sync,
|
||
_upgrade,
|
||
bootstrap_schema,
|
||
)
|
||
from deerflow.persistence.migrations._helpers import _normalize_default
|
||
|
||
# Mark only async tests via the decorator below; module-level pytestmark would
|
||
# spuriously warn for the sync ``TestDecideState`` cases.
|
||
asyncio_test = pytest.mark.asyncio
|
||
|
||
|
||
HEAD = "0002_runs_token_usage"
|
||
BASELINE = "0001_baseline"
|
||
|
||
|
||
def _url(tmp_path: Path, name: str = "test.db") -> str:
|
||
return f"sqlite+aiosqlite:///{(tmp_path / name).as_posix()}"
|
||
|
||
|
||
async def _table_names(engine) -> set[str]:
|
||
async with engine.connect() as conn:
|
||
return await conn.run_sync(lambda c: set(sa.inspect(c).get_table_names()))
|
||
|
||
|
||
async def _runs_columns(engine) -> set[str]:
|
||
async with engine.connect() as conn:
|
||
return await conn.run_sync(lambda c: {col["name"] for col in sa.inspect(c).get_columns("runs")})
|
||
|
||
|
||
async def _runs_column_meta(engine, column_name: str) -> dict:
|
||
async with engine.connect() as conn:
|
||
cols = await conn.run_sync(lambda c: sa.inspect(c).get_columns("runs"))
|
||
for c in cols:
|
||
if c["name"] == column_name:
|
||
return c
|
||
raise AssertionError(f"column {column_name!r} not found in runs")
|
||
|
||
|
||
async def _alembic_version(engine) -> str | None:
|
||
async with engine.connect() as conn:
|
||
row = await conn.execute(sa.text("SELECT version_num FROM alembic_version"))
|
||
return row.scalar()
|
||
|
||
|
||
async def _seed_legacy_without_column(engine) -> None:
|
||
"""Build the pre-#3658 schema: create_all, then drop the new column."""
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
async with engine.begin() as conn:
|
||
# SQLite supports DROP COLUMN from 3.35.0; the test runner pins recent
|
||
# Python which bundles a 3.40+ sqlite, so this is safe.
|
||
await conn.execute(sa.text("ALTER TABLE runs DROP COLUMN token_usage_by_model"))
|
||
|
||
|
||
async def _seed_legacy_with_column(engine) -> None:
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
|
||
async def _seed_legacy_missing_channel_tables(engine) -> None:
|
||
"""Build a pre-#1930 schema: baseline tables exist but ``channel_*`` do not.
|
||
|
||
Models the worst-case legacy DB the bootstrap layer has to repair -- a
|
||
user who upgraded across multiple releases and never had the channel_*
|
||
tables provisioned in the first place. We achieve it by running the full
|
||
``create_all`` and then dropping the channel_* tables in FK-dependency
|
||
order (credentials/conversations reference channel_connections).
|
||
"""
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
async with engine.begin() as conn:
|
||
for table in (
|
||
"channel_credentials",
|
||
"channel_conversations",
|
||
"channel_oauth_states",
|
||
"channel_connections",
|
||
):
|
||
await conn.execute(sa.text(f"DROP TABLE IF EXISTS {table}"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Branch 1: empty DB
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_empty_branch_creates_all_and_stamps_head(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
tables = await _table_names(engine)
|
||
for required in {
|
||
"runs",
|
||
"threads_meta",
|
||
"feedback",
|
||
"users",
|
||
"run_events",
|
||
"channel_connections",
|
||
"channel_credentials",
|
||
"channel_conversations",
|
||
"channel_oauth_states",
|
||
"alembic_version",
|
||
}:
|
||
assert required in tables, f"missing table: {required}"
|
||
assert "token_usage_by_model" in await _runs_columns(engine)
|
||
assert await _alembic_version(engine) == HEAD
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Branch 2: legacy DB without token_usage_by_model
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_without_column_branch_upgrades(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
await _seed_legacy_without_column(engine)
|
||
assert "token_usage_by_model" not in await _runs_columns(engine)
|
||
assert "alembic_version" not in await _table_names(engine)
|
||
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
|
||
assert "token_usage_by_model" in await _runs_columns(engine)
|
||
assert await _alembic_version(engine) == HEAD
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Legacy backfill: a DB that pre-dates a later-added baseline table (e.g. the
|
||
# ``channel_*`` tables from PR #1930) must end up with all baseline tables
|
||
# after bootstrap, otherwise the channels API 500s with ``no such table``.
|
||
# The fix runs ``create_all`` (idempotent) before ``stamp 0001_baseline`` so
|
||
# missing baseline tables are backfilled with their current ORM schema.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_missing_channel_tables_get_backfilled(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
await _seed_legacy_missing_channel_tables(engine)
|
||
tables = await _table_names(engine)
|
||
# Sanity-check the seeded pre-state: ``runs`` triggers the legacy
|
||
# branch (has_deerflow_tables=True, no alembic_version) while the
|
||
# channel_* tables are absent.
|
||
assert "runs" in tables
|
||
assert "alembic_version" not in tables
|
||
for missing in {
|
||
"channel_connections",
|
||
"channel_credentials",
|
||
"channel_conversations",
|
||
"channel_oauth_states",
|
||
}:
|
||
assert missing not in tables, f"seed should not have {missing}"
|
||
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
|
||
tables = await _table_names(engine)
|
||
for required in {
|
||
"channel_connections",
|
||
"channel_credentials",
|
||
"channel_conversations",
|
||
"channel_oauth_states",
|
||
}:
|
||
assert required in tables, f"legacy backfill missed: {required}"
|
||
assert await _alembic_version(engine) == HEAD
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Branch 3: legacy DB that ALREADY has the column (post-#3658 create_all,
|
||
# or user-applied manual ALTER). The branch is the same as the
|
||
# legacy-without-column case -- bootstrap stamps baseline and tries to
|
||
# upgrade. The idempotent revision helper (``safe_add_column``) silently
|
||
# skips when the column is present, so the schema does not change.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_with_column_branch_upgrade_is_idempotent(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
await _seed_legacy_with_column(engine)
|
||
assert "token_usage_by_model" in await _runs_columns(engine)
|
||
assert "alembic_version" not in await _table_names(engine)
|
||
cols_before = await _runs_columns(engine)
|
||
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
|
||
cols_after = await _runs_columns(engine)
|
||
assert cols_after == cols_before, "idempotent upgrade should not alter schema"
|
||
assert await _alembic_version(engine) == HEAD
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Drift-warning guard: a column re-added by a manual ALTER (e.g. the #3682
|
||
# workaround) survives the legacy branch because ``safe_add_column`` is
|
||
# name-keyed, but the helper must ``logger.warning`` so the operator notices
|
||
# the residual nullability / server_default / type drift from the model.
|
||
# Two scenarios are pinned: (1) nullable JSON workaround -- nullability +
|
||
# server_default drift fire, type matches; (2) ``TEXT NOT NULL DEFAULT
|
||
# '{}'`` workaround -- only type drifts, must STILL fire thanks to the
|
||
# JSON/TEXT family check in ``_type_equivalent``.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_with_manual_workaround_column_warns_on_drift(
|
||
tmp_path: Path,
|
||
caplog: pytest.LogCaptureFixture,
|
||
) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
# Pre-#3658 schema with a workaround-style re-add: nullable JSON,
|
||
# no server default -- diverges from the model's NOT NULL DEFAULT '{}'.
|
||
# Type matches (JSON vs JSON) so the type-equivalence check stays quiet
|
||
# and the warning fires purely on nullability + server_default.
|
||
await _seed_legacy_without_column(engine)
|
||
async with engine.begin() as conn:
|
||
await conn.execute(sa.text("ALTER TABLE runs ADD COLUMN token_usage_by_model JSON"))
|
||
|
||
with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"):
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
|
||
# Bootstrap still completes -- the helper does not block on drift.
|
||
assert await _alembic_version(engine) == HEAD
|
||
# And the manually-added column survives untouched (no auto-repair).
|
||
col = await _runs_column_meta(engine, "token_usage_by_model")
|
||
assert col["nullable"] is True
|
||
|
||
drift_warnings = [r for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers" and "safe_add_column" in r.getMessage() and "token_usage_by_model" in r.getMessage()]
|
||
assert drift_warnings, "expected safe_add_column to warn about the drifted column"
|
||
msg = drift_warnings[0].getMessage()
|
||
assert "nullable" in msg
|
||
assert "server_default" in msg
|
||
# Type info is always echoed in the payload for triage context.
|
||
assert "actual_type=" in msg and "desired_type=" in msg, f"warning missing type info: {msg!r}"
|
||
# JSON ≈ JSON, so the equivalence check must NOT produce a "type" diff
|
||
# entry here -- that would be a false positive on the matching-type case.
|
||
assert "type actual=" not in msg, f"unexpected type drift on matching JSON column: {msg!r}"
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_with_wrong_type_workaround_warns_on_type_drift(
|
||
tmp_path: Path,
|
||
caplog: pytest.LogCaptureFixture,
|
||
) -> None:
|
||
"""The precise reviewer scenario: nullability + server_default match the
|
||
model, only the type is wrong. Pre-family-check, this returned zero
|
||
warning (silent JSON-vs-TEXT drift). The family check in
|
||
``_type_equivalent`` must catch this while leaving JSON/JSONB pairs
|
||
equivalent so Postgres dialect synonyms don't false-positive."""
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
# Reviewer's exact workaround: right nullability/default, wrong type.
|
||
await _seed_legacy_without_column(engine)
|
||
async with engine.begin() as conn:
|
||
await conn.execute(sa.text("ALTER TABLE runs ADD COLUMN token_usage_by_model TEXT NOT NULL DEFAULT '{}'"))
|
||
|
||
with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"):
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
|
||
assert await _alembic_version(engine) == HEAD
|
||
# No auto-repair: the TEXT column survives unchanged so the operator
|
||
# can decide whether to ALTER it themselves.
|
||
col = await _runs_column_meta(engine, "token_usage_by_model")
|
||
assert col["nullable"] is False
|
||
|
||
drift_warnings = [r for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers" and "safe_add_column" in r.getMessage() and "token_usage_by_model" in r.getMessage()]
|
||
assert drift_warnings, "expected safe_add_column to warn about pure type drift (was silent before the family check)"
|
||
msg = drift_warnings[0].getMessage()
|
||
# The drift entry must explicitly name the type mismatch -- this is
|
||
# what was missing before the family check existed.
|
||
assert "type actual=" in msg and "desired=" in msg, f"warning missing type drift entry: {msg!r}"
|
||
assert "TEXT" in msg and "JSON" in msg, f"warning missing TEXT/JSON in payload: {msg!r}"
|
||
# Nullability + server_default match the model -- no other diffs.
|
||
assert "nullable" not in msg, f"unexpected nullability drift on matching column: {msg!r}"
|
||
assert "server_default" not in msg, f"unexpected server_default drift on matching column: {msg!r}"
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _type_equivalent unit tests: pin the JSON/JSONB equivalence so Postgres
|
||
# dialect synonyms stay quiet, and pin the TEXT/JSON divergence so the
|
||
# reviewer's wrong-type scenario keeps firing.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_type_equivalent_matches_known_dialect_synonyms() -> None:
|
||
from deerflow.persistence.migrations._helpers import _type_equivalent
|
||
|
||
# JSON ↔ JSONB (Postgres dialect difference, operationally interchangeable
|
||
# for our schema). Both directions, and via raw strings.
|
||
assert _type_equivalent(sa.JSON(), "JSONB()") is True
|
||
assert _type_equivalent("JSON", "JSONB") is True
|
||
assert _type_equivalent("JSONB", "JSON") is True
|
||
|
||
|
||
def test_type_equivalent_catches_wholesale_type_mismatch() -> None:
|
||
from deerflow.persistence.migrations._helpers import _type_equivalent
|
||
|
||
# The reviewer scenario: TEXT NOT NULL DEFAULT '{}' workaround.
|
||
assert _type_equivalent("TEXT", "JSON") is False
|
||
assert _type_equivalent("TEXT", "JSONB") is False
|
||
# Unrelated families also don't accidentally pair up.
|
||
assert _type_equivalent("INTEGER", "JSON") is False
|
||
|
||
|
||
def test_type_equivalent_ignores_type_parameters() -> None:
|
||
"""Length / precision differences are out of scope for this helper --
|
||
the goal is wholesale-type drift, not dialect-rendered size defaults."""
|
||
from deerflow.persistence.migrations._helpers import _type_equivalent
|
||
|
||
assert _type_equivalent("VARCHAR(255)", "VARCHAR(500)") is True
|
||
assert _type_equivalent("NUMERIC(10,2)", "NUMERIC(20,4)") is True
|
||
|
||
|
||
def test_type_equivalent_returns_true_on_missing_info() -> None:
|
||
"""Missing reflected info must not false-positive into a noisy warning."""
|
||
from deerflow.persistence.migrations._helpers import _type_equivalent
|
||
|
||
assert _type_equivalent(None, sa.JSON()) is True
|
||
assert _type_equivalent(sa.JSON(), None) is True
|
||
assert _type_equivalent("", "JSON") is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Branch 4: versioned DB
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_versioned_branch_is_noop_at_head(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
# First bootstrap takes us through the empty branch.
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
cols_before = await _runs_columns(engine)
|
||
# Second call hits the versioned branch.
|
||
await bootstrap_schema(engine, backend="sqlite")
|
||
cols_after = await _runs_columns(engine)
|
||
assert cols_after == cols_before
|
||
assert await _alembic_version(engine) == HEAD
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Schema-parity guard: legacy-upgraded DB must end up structurally identical
|
||
# to a fresh DB on the columns the migration touches. This is the property
|
||
# that catches drift between ``Base.metadata`` and ``0002``'s DDL -- exactly
|
||
# the failure mode of the original #3682 bug, just at a different layer.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_token_usage_column_parity_between_fresh_and_upgraded(tmp_path: Path) -> None:
|
||
fresh = create_async_engine(_url(tmp_path, "fresh.db"))
|
||
upgraded = create_async_engine(_url(tmp_path, "upgraded.db"))
|
||
try:
|
||
# Fresh DB -> empty branch -> create_all
|
||
await bootstrap_schema(fresh, backend="sqlite")
|
||
fresh_col = await _runs_column_meta(fresh, "token_usage_by_model")
|
||
|
||
# Legacy DB -> stamp baseline + 0002 upgrade
|
||
await _seed_legacy_without_column(upgraded)
|
||
await bootstrap_schema(upgraded, backend="sqlite")
|
||
upgraded_col = await _runs_column_meta(upgraded, "token_usage_by_model")
|
||
|
||
# Pin the contract: the column must have the same nullability AND
|
||
# server_default after either bootstrap path. If 0002 ever drifts
|
||
# from the model's ``Mapped[dict] = mapped_column(JSON, default=dict,
|
||
# server_default=text("'{}'"))`` (i.e. ``nullable=False`` plus the
|
||
# ``'{}'`` DB-side default), this fires.
|
||
assert fresh_col["nullable"] == upgraded_col["nullable"], f"nullability drift: fresh={fresh_col['nullable']} upgraded={upgraded_col['nullable']}"
|
||
# The model declares Mapped[dict] (non-optional) -> NOT NULL.
|
||
assert fresh_col["nullable"] is False
|
||
assert upgraded_col["nullable"] is False
|
||
# Normalize through the same helper the drift warning uses so dialect
|
||
# quirks (outer parens, ``::cast``) do not cause false negatives.
|
||
assert _normalize_default(fresh_col.get("default")) == _normalize_default(upgraded_col.get("default")), f"server_default drift: fresh={fresh_col.get('default')!r} upgraded={upgraded_col.get('default')!r}"
|
||
finally:
|
||
await fresh.dispose()
|
||
await upgraded.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Full schema parity: ``Base.metadata.create_all`` and ``alembic upgrade
|
||
# base->head`` MUST produce structurally identical schemas. Both are
|
||
# independent sources of the same schema in this codebase -- fresh DBs are
|
||
# provisioned by the former (empty branch), historical/upgraded DBs by the
|
||
# latter (versioned branch and the alembic tail of the legacy branch). If
|
||
# they diverge, two users running the same app version end up with different
|
||
# DB structures: exactly the cross-deployment drift this PR exists to kill.
|
||
#
|
||
# The check is intentionally scoped to columns × (nullable, server_default)
|
||
# instead of full type/index/FK reflection. Those are the two highest-signal
|
||
# attributes for the drift modes seen so far (#3682 was a nullability
|
||
# mismatch; review todo #6 was a server_default mismatch). Type, index, and
|
||
# FK reflection differ enough across dialects to require careful
|
||
# normalization helpers that aren't worth introducing for this PR's scope;
|
||
# see review todo #7 for the wider plan.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _reflect_columns_sync(sync_conn) -> dict[str, dict[str, dict]]:
|
||
insp = sa.inspect(sync_conn)
|
||
out: dict[str, dict[str, dict]] = {}
|
||
for table in insp.get_table_names():
|
||
# ``alembic_version`` is alembic's own bookkeeping table, not part of
|
||
# our schema -- one path creates it (upgrade) and the other doesn't
|
||
# (create_all), so comparing it would produce a guaranteed false
|
||
# positive every run.
|
||
if table == "alembic_version":
|
||
continue
|
||
out[table] = {c["name"]: c for c in insp.get_columns(table)}
|
||
return out
|
||
|
||
|
||
async def _reflect_columns(engine) -> dict[str, dict[str, dict]]:
|
||
async with engine.connect() as conn:
|
||
return await conn.run_sync(_reflect_columns_sync)
|
||
|
||
|
||
@asyncio_test
|
||
async def test_create_all_and_alembic_upgrade_produce_same_schema(tmp_path: Path) -> None:
|
||
fresh = create_async_engine(_url(tmp_path, "fresh.db"))
|
||
upgraded = create_async_engine(_url(tmp_path, "upgraded.db"))
|
||
try:
|
||
# Path A: ``Base.metadata.create_all`` -- the empty-branch code path.
|
||
async with fresh.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
# Path B: pure alembic ``upgrade base->head``. Note we deliberately
|
||
# bypass ``bootstrap_schema`` on this side -- its empty branch uses
|
||
# ``create_all``, not the alembic chain -- to exercise the path a
|
||
# versioned-DB upgrade actually takes.
|
||
cfg = _get_alembic_config(upgraded)
|
||
await asyncio.to_thread(_upgrade, cfg, "head")
|
||
|
||
fresh_tables = await _reflect_columns(fresh)
|
||
upgraded_tables = await _reflect_columns(upgraded)
|
||
|
||
# Same set of tables. A mismatch here means either ``Base.metadata``
|
||
# has gained/lost a table without a matching revision, or a revision
|
||
# creates/drops a table without a matching model change.
|
||
assert set(fresh_tables) == set(upgraded_tables), f"table-set drift between create_all and alembic upgrade: only-in-create_all={set(fresh_tables) - set(upgraded_tables)} only-in-alembic={set(upgraded_tables) - set(fresh_tables)}"
|
||
|
||
for table in sorted(fresh_tables):
|
||
fresh_cols = fresh_tables[table]
|
||
upgraded_cols = upgraded_tables[table]
|
||
assert set(fresh_cols) == set(upgraded_cols), f"{table}: column-set drift only-in-create_all={set(fresh_cols) - set(upgraded_cols)} only-in-alembic={set(upgraded_cols) - set(fresh_cols)}"
|
||
for col_name in sorted(fresh_cols):
|
||
f_col = fresh_cols[col_name]
|
||
u_col = upgraded_cols[col_name]
|
||
assert f_col["nullable"] == u_col["nullable"], f"{table}.{col_name}: nullable drift create_all={f_col['nullable']} alembic={u_col['nullable']}"
|
||
# Normalize through ``_normalize_default`` to absorb the
|
||
# dialect-rendering quirks (outer parens, ``::cast``) that
|
||
# would otherwise cause false positives.
|
||
f_default = _normalize_default(f_col.get("default"))
|
||
u_default = _normalize_default(u_col.get("default"))
|
||
assert f_default == u_default, f"{table}.{col_name}: server_default drift create_all={f_col.get('default')!r} alembic={u_col.get('default')!r}"
|
||
finally:
|
||
await fresh.dispose()
|
||
await upgraded.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Baseline-table-restriction guards. The legacy branch's backfill must
|
||
# create *only* the baseline-era tables, not the full ``Base.metadata``.
|
||
# Otherwise it would pre-empt a future ``op.create_table`` revision for a
|
||
# newly-added model (the revision would crash with ``relation already
|
||
# exists``). Two tests cover this:
|
||
#
|
||
# 1. ``_BASELINE_TABLE_NAMES`` is pinned against what ``0001_baseline``
|
||
# actually creates -- editing 0001 without updating the constant fires
|
||
# here, forcing the developer to keep the two in sync.
|
||
# 2. Regression for the leak itself: a phantom table outside the constant
|
||
# must NOT be created by the backfill helper.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@asyncio_test
|
||
async def test_baseline_table_names_constant_matches_0001(tmp_path: Path) -> None:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
cfg = _get_alembic_config(engine)
|
||
# Run only up to baseline (not head) and reflect what it produced.
|
||
await asyncio.to_thread(_upgrade, cfg, BASELINE)
|
||
|
||
async with engine.connect() as conn:
|
||
reflected = await conn.run_sync(lambda c: set(sa.inspect(c).get_table_names()))
|
||
# ``alembic_version`` is alembic's bookkeeping table, not part of
|
||
# our schema -- the constant is about DeerFlow-owned baseline tables.
|
||
reflected.discard("alembic_version")
|
||
|
||
assert reflected == _BASELINE_TABLE_NAMES, f"_BASELINE_TABLE_NAMES drifted from 0001_baseline.upgrade()'s output: only-in-0001={sorted(reflected - _BASELINE_TABLE_NAMES)} only-in-constant={sorted(_BASELINE_TABLE_NAMES - reflected)}"
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@asyncio_test
|
||
async def test_legacy_backfill_skips_non_baseline_tables(tmp_path: Path) -> None:
|
||
"""Regression: legacy backfill must not create tables outside the baseline
|
||
set, because a later ``op.create_table`` revision for the same name would
|
||
fail. We synthesise a phantom table on ``Base.metadata`` (modelling a
|
||
future model addition), run the backfill helper, and assert the phantom
|
||
is absent from the resulting DB.
|
||
"""
|
||
phantom_name = "phantom_future_table_for_test"
|
||
phantom = sa.Table(
|
||
phantom_name,
|
||
Base.metadata,
|
||
sa.Column("id", sa.Integer, primary_key=True),
|
||
)
|
||
try:
|
||
engine = create_async_engine(_url(tmp_path))
|
||
try:
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(_run_baseline_create_all_sync)
|
||
|
||
async with engine.connect() as conn:
|
||
tables = await conn.run_sync(lambda c: set(sa.inspect(c).get_table_names()))
|
||
|
||
assert phantom_name not in tables, f"legacy backfill leaked {phantom_name!r}; a future ``op.create_table({phantom_name!r})`` revision would now collide"
|
||
# Sanity: baseline tables ARE created by the backfill helper.
|
||
assert "runs" in tables
|
||
assert "channel_connections" in tables
|
||
finally:
|
||
await engine.dispose()
|
||
finally:
|
||
Base.metadata.remove(phantom)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _decide_state unit tests (pure function, no DB needed)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDecideState:
|
||
def test_empty(self):
|
||
assert _decide_state({"has_alembic_version": False, "has_deerflow_tables": False}) == "empty"
|
||
|
||
def test_empty_with_unrelated_tables(self):
|
||
# LangGraph checkpointer tables present but DeerFlow has nothing yet.
|
||
# ``has_deerflow_tables`` is derived from the metadata intersection in
|
||
# production, so the only thing the decision function needs is the
|
||
# bool itself.
|
||
assert _decide_state({"has_alembic_version": False, "has_deerflow_tables": False}) == "empty"
|
||
|
||
def test_legacy(self):
|
||
assert _decide_state({"has_alembic_version": False, "has_deerflow_tables": True}) == "legacy"
|
||
|
||
def test_versioned(self):
|
||
assert _decide_state({"has_alembic_version": True, "has_deerflow_tables": True}) == "versioned"
|
||
|
||
def test_versioned_takes_precedence_over_empty(self):
|
||
# Pathological: alembic_version row exists but no managed tables yet
|
||
# (e.g. someone restored only the alembic_version table from backup).
|
||
# We still go versioned -> upgrade head, which is the right thing:
|
||
# alembic will run every revision from base.
|
||
assert _decide_state({"has_alembic_version": True, "has_deerflow_tables": False}) == "versioned"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Sanity: head revision is the one this module expects
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_head_revision_is_token_usage_revision() -> None:
|
||
assert _get_head_revision() == HEAD
|
||
|
||
|
||
def test_baseline_revision_id_is_known() -> None:
|
||
"""Detect a baseline rename: the bootstrap code hardcodes ``0001_baseline``
|
||
as the stamp target for the legacy branch, so a rename would silently
|
||
break that branch unless caught here."""
|
||
from pathlib import Path # noqa: PLC0415
|
||
|
||
from alembic.config import Config # noqa: PLC0415
|
||
from alembic.script import ScriptDirectory # noqa: PLC0415
|
||
|
||
migrations_dir = Path(__file__).resolve().parents[1] / "packages/harness/deerflow/persistence/migrations"
|
||
cfg = Config()
|
||
cfg.set_main_option("script_location", str(migrations_dir))
|
||
script = ScriptDirectory.from_config(cfg)
|
||
all_ids = {rev.revision for rev in script.walk_revisions()}
|
||
assert BASELINE in all_ids, f"baseline revision id {BASELINE!r} not found in {all_ids}"
|