mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
* 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>
636 lines
29 KiB
Python
636 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 = "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}"
|