mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 17:06:05 +00:00
838037188e
* feat(channels): share inbound webhook dedupe across pods via Postgres (#4120) * ci: run cross-pod inbound dedupe integration tests in CI Expose the job Postgres service via DEDUPE_TEST_POSTGRES_URL so the integration tests (issue #4120) actually execute instead of silently skipping. Normalize the URL for asyncpg (postgresql:// -> +asyncpg, drop libpq-only sslmode) and await the now-async _is_duplicate_inbound in test_github_dispatcher.
54 lines
2.4 KiB
Python
54 lines
2.4 KiB
Python
"""Migration ``0009_webhook_dedupe`` regression test (issue #4120).
|
|
|
|
Verifies the migration creates ``webhook_deliveries`` with the composite
|
|
primary key (channel, workspace_id, chat_id, message_id) and no legacy
|
|
``dedupe_key`` column, and that re-running it is idempotent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
import deerflow.persistence.models # noqa: F401 -- registers ORM models
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.bootstrap import bootstrap_schema
|
|
from deerflow.persistence.engine import close_engine, init_engine
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def test_migration_0009_creates_composite_pk_table_and_is_idempotent(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "deer.db"
|
|
url = f"sqlite+aiosqlite:///{db_path}"
|
|
engine = create_async_engine(url)
|
|
try:
|
|
# Seed all baseline tables, then drop ONLY webhook_deliveries so the
|
|
# 0009 upgrade actually exercises its create_table path (not the
|
|
# idempotent early-return). Stamp at 0004 so bootstrap upgrades to head.
|
|
sync = sa.create_engine(f"sqlite:///{db_path}")
|
|
Base.metadata.create_all(sync)
|
|
with sync.begin() as conn:
|
|
conn.execute(sa.text("DROP TABLE IF EXISTS webhook_deliveries"))
|
|
conn.execute(sa.text("CREATE TABLE IF NOT EXISTS alembic_version (version_num VARCHAR(32) NOT NULL)"))
|
|
conn.execute(sa.text("DELETE FROM alembic_version"))
|
|
conn.execute(sa.text("INSERT INTO alembic_version (version_num) VALUES ('0004_run_ownership')"))
|
|
|
|
await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
|
|
# Runs upgrade head -> executes 0009.create_table.
|
|
await bootstrap_schema(engine, backend="sqlite")
|
|
|
|
async with engine.connect() as conn:
|
|
cols = {row["name"] for row in await conn.run_sync(lambda c: sa.inspect(c).get_columns("webhook_deliveries"))}
|
|
# Composite PK columns only; the old single-column ``dedupe_key`` must
|
|
# NOT exist (it is illegal in Postgres TEXT and caused schema drift).
|
|
assert cols == {"channel", "workspace_id", "chat_id", "message_id", "first_seen"}
|
|
|
|
# Idempotent: re-running bootstrap at head must not raise (table exists).
|
|
await bootstrap_schema(engine, backend="sqlite")
|
|
finally:
|
|
await close_engine()
|