deer-flow/backend/tests/test_persistence_bootstrap.py
Xinmin Zeng 4fc08b4f15
feat: add scheduled tasks MVP (#3898)
* feat: add scheduled tasks MVP

* fix: harden scheduled task execution semantics

* feat(scheduled-tasks): preset-driven schedule form with timezone and live preview

Replace the raw cron input with a preset Select (hourly/daily/weekly/monthly/custom)
plus structured inputs (time picker, weekday toggles, day-of-month), datetime-local
for one-time tasks, a timezone selector defaulting to the browser timezone, and a
live human-readable preview. Reuses one ScheduledTaskScheduleInput for create and
edit; backend contract unchanged; zero new deps (pure Intl + DST-safe offset helpers).

* feat(scheduled-tasks): full-page i18n + recipe templates + E2E locale pin

Localize the rest of the scheduled-tasks page (filters, detail pane, actions,
edit form, run list, enum values) via t.scheduledTasks.* in en/zh. Add four
built-in recipe templates (GitHub Trending, news digest, issue triage, weekly
report) exposed as a chip row that pre-fills title + prompt + schedule. Pin
Playwright locale to en-US so E2E selectors stay stable against i18n. No backend
change, no new deps.

* fix(scheduled-tasks): idempotent 0003 migration, update head constants, future-date once test

Merge with main surfaced three CI failures:
- 0003_scheduled_tasks create_table collided with legacy test seeds that
  build from full metadata; guard with inspector.has_table so the revision
  no-ops when the table already exists (0004/0005 are already idempotent via
  _helpers.py).
- persistence bootstrap concurrency/regression tests pinned HEAD to main's
  0002_runs_token_usage; bump to the new head 0005_scheduled_task_thread_nullable.
- once-task router test used a fixed past run_at and tripped the
  must-be-in-the-future validation; use a future date.

* address review: ok-check, 502 for trigger failure, mock fields, migration filename, doc fences

- fetchThreadScheduledTasks now checks response.ok like the other fetchers.
- trigger endpoint returns 502 (not 409) when dispatch fails outright, so
  clients can distinguish a real conflict from a server-side failure.
- E2E mock normalizes scheduled-task objects with context_mode/last_thread_id
  and nullable thread_id, matching the backend contract the UI renders against.
- Rename 0002_scheduled_tasks.py -> 0003_scheduled_tasks.py to match its
  revision id (file was renamed in spirit already; filename now follows).
- CONFIGURATION.md: close the Tool Groups yaml fence and drop the stray fence
  after the Scheduler notes so the sections render correctly.

* fix(scheduled-tasks): harden lease, poller, config, and frontend UX after review

* fix(scheduled-tasks): harden run lifecycle, overlap skip, non_interactive gating, and DST conversion after review

- defer a once task's terminal status to the run-completion hook; the task
  stays running until the real outcome, and a startup sweep cancels once
  tasks orphaned by a crash (launch-time 'completed' could stick forever)
- record interrupted runs as a distinct 'interrupted' run status with a
  readable message; an interrupted once task ends 'cancelled', not 'failed'
- enforce overlap_policy=skip for fresh_thread_per_run via an active-run
  pre-check (same-thread ConflictError can never fire across fresh threads)
- protect terminal run statuses from the late launch-path 'running' write
- honor context.non_interactive only for internally-authenticated callers;
  arbitrary clients can no longer strip ask_clarification
- fix DST-stale timezone offset in zonedLocalToUtcIso by re-deriving the
  offset at the resolved instant (once tasks fired an hour late around
  spring-forward and the create->edit round-trip diverged)
- drop dead ScheduledTaskRunRepository.update_by_run_id; share one Gateway
  API error helper between channels and scheduled-tasks frontends

* fix(scheduled-tasks): close review round-3 gaps in guards, concurrency, and API ergonomics

- scrub internal-only context keys (non_interactive) from the assembled run
  config for non-internal callers: gating body.context alone left the same
  key smuggle-able through the free-form body.config copied verbatim by
  build_run_config
- guard update_after_launch with protect_terminal so the launch bookkeeping
  write cannot clobber a once task already finalized by a fast-failing run's
  completion hook (parent-row sibling of the run-row guard)
- reject a manual trigger while the task has an active run (409) instead of
  launching a duplicate concurrent run on fresh_thread_per_run
- re-arm a terminal once task to enabled when PATCH pushes run_at into the
  future; previously the endpoint returned 200 with a next_run_at that could
  never be claimed
- make max_concurrent_runs a real global cap: each poll claims only into the
  remaining budget of active (queued/running) scheduled runs
- paginate GET /scheduled-tasks/{id}/runs (limit<=200, offset) and push the
  thread filter of /threads/{id}/scheduled-tasks into SQL
- stamp context.user_id on scheduler-launched runs, matching IM channels, so
  user-scoped guardrail providers see the owning user

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-04 21:51:57 +08:00

636 lines
29 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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 = "0003_scheduled_tasks"
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}"