deer-flow/backend/tests/test_postgres_schema_helper.py
Ryker_Feng 063d62c3c3
feat(persistence): support custom postgres schema (#3442)
* feat(persistence): support custom postgres schema

* fix(persistence): address CI lint/test failures and review feedback

- Map missing psycopg import to actionable POSTGRES_INSTALL guidance in
  sync/async schema-creation helpers
- Accept SQLAlchemy compound DSN schemes (postgresql+asyncpg) when
  injecting search_path, normalizing to a libpq-consumable DSN
- Guard keyword-DSN tests with importorskip so they skip without psycopg
- Set database=None in sync checkpointer none-fix test to avoid MagicMock
  backend resolution
- Apply ruff import sort and format

* fix(persistence): address pg-schema review feedback

- Restrict postgres_schema regex to lowercase-only so the quoted CREATE
  SCHEMA matches the unquoted search_path (PG case-folds it), fixing the
  mixed-case bug where tables silently fell back to public.
- Replace shlex.join/split with libpq-correct backslash escaping for the
  options parameter so values containing spaces survive intact.
- Add normalize_libpq_dsn() and route the async checkpointer pool through
  dsn_with_search_path() so a +asyncpg suffix is stripped and existing DSN
  options (e.g. statement_timeout) are merged instead of overridden.
- Extract shared ensure_postgres_schema()/ensure_postgres_schema_async()
  helpers (mapping missing psycopg to the install hint) used by all four
  provider sites.
- Tests: reject mixed-case schemas, preserve space-containing libpq option,
  cover normalize_libpq_dsn, and assert pool search_path via DSN.

* fix(persistence): align pg-schema test with merged store API

The main merge moved the sync Store factory to the single-path
_resolve_store_config/_sync_store_cm design, dropping the PR's
_sync_store_from_database helper. The integration test still imported
the removed symbol, breaking test collection (backend-unit-tests).
Resolve the store config from a DatabaseConfig and drive it through
_sync_store_cm instead.

* fix(persistence): address pg-schema review feedback

- reject trailing/leading whitespace in postgres_schema via re.fullmatch
  (a $-anchored re.match let "deerflow\n" through, silently landing tables
  in public)
- re-escape all whitespace (TAB/CR/LF) when re-joining libpq options so a
  caller's pre-existing options value round-trips losslessly
- re-validate the identifier inside create_schema_sql as defense-in-depth
  at the SQL-emitting boundary
- accept the postgres:// short scheme in the alembic search_path injection
- close the sync psycopg connection explicitly (psycopg3 __exit__ does not
  close()), mirroring the async path
- drop the partial checkpointer/store reset on a database config change;
  database is restart-required and the ORM engine is not rebuilt, so a
  partial reset would half-migrate the deployment

* docs(config): complete the postgres_schema migration checklist

Address PR review (P1): the documented `public`->schema migration only
moved runs, run_events, threads_meta, feedback, and users. That strands
every other DeerFlow-owned table -- the four channel_* tables, both
scheduled_* tables, agents, and (critically) alembic_version -- in
`public`. On restart bootstrap treats the partially-populated target
schema as unversioned, re-baselines it, and replays migrations while the
real rows stay invisible in `public`.

List the full owned set explicitly, call out alembic_version as required,
and keep the "discover the rest" query for version-drift safety.

* refactor(checkpointer): drop test-only _sync_checkpointer_from_database

Address PR review: the helper was only reached by the env-gated
integration test and re-implemented the DatabaseConfig->CheckpointerConfig
backend resolution that _resolve_checkpointer_config already owns, so a
future backend added there would silently miss this path. Mirror the store
side of the same test, which reuses the production path directly:
_resolve_checkpointer_config(...) + _sync_checkpointer_cm(...).
2026-07-30 13:51:17 +08:00

179 lines
7.4 KiB
Python

"""Tests for the PostgreSQL schema helpers (Issue #3380)."""
from urllib.parse import parse_qs, urlsplit
import pytest
from deerflow.persistence.postgres_schema import (
build_asyncpg_connect_args,
build_psycopg_options,
create_schema_sql,
dsn_with_search_path,
normalize_libpq_dsn,
)
class TestBuildAsyncpgConnectArgs:
def test_sets_search_path_for_schema(self):
assert build_asyncpg_connect_args("deerflow") == {"server_settings": {"search_path": "deerflow"}}
def test_empty_schema_returns_empty_dict(self):
assert build_asyncpg_connect_args("") == {}
class TestBuildPsycopgOptions:
def test_builds_libpq_options(self):
assert build_psycopg_options("deerflow") == "-c search_path=deerflow"
def test_empty_schema_returns_none(self):
assert build_psycopg_options("") is None
class TestCreateSchemaSql:
def test_builds_create_schema_statement(self):
assert create_schema_sql("deerflow") == 'CREATE SCHEMA IF NOT EXISTS "deerflow"'
def test_empty_schema_returns_none(self):
assert create_schema_sql("") is None
@pytest.mark.parametrize("schema", ['a"; DROP SCHEMA public; --', "MySchema", "a b", "deerflow\n"])
def test_rejects_non_plain_identifier(self, schema):
# Defense-in-depth: the SQL-emitting boundary re-validates so a caller
# that bypasses the pydantic config validator cannot inject.
with pytest.raises(ValueError):
create_schema_sql(schema)
class TestDsnWithSearchPath:
def test_empty_schema_returns_dsn_unchanged(self):
dsn = "postgresql://u:p@h:5432/db"
assert dsn_with_search_path(dsn, "") == dsn
def test_appends_options_query_encoded(self):
dsn = "postgresql://u:p@h:5432/db"
out = dsn_with_search_path(dsn, "deerflow")
# libpq only decodes %XX in URI query values; '+' is NOT treated as a
# space. The space MUST therefore be encoded as %20, never as '+'.
assert "+" not in out
assert "options=-c%20search_path%3Ddeerflow" in out
parts = urlsplit(out)
query = parse_qs(parts.query)
assert query["options"] == ["-c search_path=deerflow"]
def test_merges_with_existing_query(self):
dsn = "postgresql://u:p@h:5432/db?sslmode=require"
out = dsn_with_search_path(dsn, "deerflow")
query = parse_qs(urlsplit(out).query)
assert query["sslmode"] == ["require"]
assert query["options"] == ["-c search_path=deerflow"]
def test_replaces_existing_options_query(self):
dsn = "postgresql://u:p@h:5432/db?options=-c%20search_path%3Dpublic"
out = dsn_with_search_path(dsn, "deerflow")
query = parse_qs(urlsplit(out).query)
assert query["options"] == ["-c search_path=deerflow"]
def test_preserves_existing_options_query(self):
dsn = "postgresql://u:p@h:5432/db?options=-c%20statement_timeout%3D5000"
out = dsn_with_search_path(dsn, "deerflow")
query = parse_qs(urlsplit(out).query)
assert query["options"] == ["-c statement_timeout=5000 -c search_path=deerflow"]
def test_replaces_only_existing_search_path_option(self):
dsn = "postgresql://u:p@h:5432/db?options=-c%20statement_timeout%3D5000%20-c%20search_path%3Dpublic"
out = dsn_with_search_path(dsn, "deerflow")
query = parse_qs(urlsplit(out).query)
assert query["options"] == ["-c statement_timeout=5000 -c search_path=deerflow"]
def test_supports_keyword_dsn(self):
pytest.importorskip("psycopg")
from psycopg.conninfo import conninfo_to_dict
dsn = "host=localhost dbname=deerflow user=postgres"
out = dsn_with_search_path(dsn, "deerflow")
assert conninfo_to_dict(out) == {
"host": "localhost",
"dbname": "deerflow",
"user": "postgres",
"options": "-c search_path=deerflow",
}
def test_preserves_keyword_dsn_options(self):
pytest.importorskip("psycopg")
from psycopg.conninfo import conninfo_to_dict
dsn = "host=localhost dbname=deerflow options='-c statement_timeout=5000'"
out = dsn_with_search_path(dsn, "deerflow")
assert conninfo_to_dict(out)["options"] == "-c statement_timeout=5000 -c search_path=deerflow"
def test_normalizes_sqlalchemy_driver_scheme(self):
# DatabaseConfig.postgres_url may carry a +asyncpg suffix; the libpq DSN
# produced for psycopg must drop the driver and still inject search_path.
dsn = "postgresql+asyncpg://u:p@h:5432/db"
out = dsn_with_search_path(dsn, "deerflow")
parts = urlsplit(out)
assert parts.scheme == "postgresql"
query = parse_qs(parts.query)
assert query["options"] == ["-c search_path=deerflow"]
def test_rejects_non_postgres_url_scheme(self):
try:
dsn_with_search_path("mysql://localhost/db", "deerflow")
except ValueError as exc:
assert "Unsupported PostgreSQL DSN scheme" in str(exc)
else:
raise AssertionError("Expected ValueError")
def test_roundtrip_preserves_host_and_db(self):
dsn = "postgresql://u:p@h:5432/db"
out = dsn_with_search_path(dsn, "deerflow")
parts = urlsplit(out)
assert parts.hostname == "h"
assert parts.port == 5432
assert parts.path == "/db"
def test_preserves_option_value_containing_space(self):
# libpq's options parameter separates args on spaces unless they are
# backslash-escaped. shlex.join would emit single-quotes, which libpq
# treats as literal characters and would corrupt the option. A token
# carrying a space must round-trip as a single backslash-escaped token.
from deerflow.persistence.postgres_schema import _merge_search_path_option
merged = _merge_search_path_option(r"-c application_name=My\ App", "deerflow")
assert "'" not in merged
assert r"application_name=My\ App" in merged
assert merged.endswith("-c search_path=deerflow")
def test_preserves_option_value_containing_tab(self):
# Non-space whitespace (TAB/CR/LF) inside an existing escaped token must
# also be re-escaped on re-join, otherwise libpq re-tokenizes on the bare
# whitespace byte and the round-trip is lossy.
from deerflow.persistence.postgres_schema import (
_merge_search_path_option,
_split_libpq_options,
)
merged = _merge_search_path_option("-c application_name=My\\\tApp", "deerflow")
assert "'" not in merged
# The tab-bearing value must round-trip back to a single token.
tokens = _split_libpq_options(merged)
assert "application_name=My\tApp" in tokens
assert merged.endswith("-c search_path=deerflow")
class TestNormalizeLibpqDsn:
def test_strips_asyncpg_driver_suffix(self):
assert normalize_libpq_dsn("postgresql+asyncpg://u:p@h:5432/db") == "postgresql://u:p@h:5432/db"
def test_leaves_bare_postgres_scheme_unchanged(self):
dsn = "postgresql://u:p@h:5432/db"
assert normalize_libpq_dsn(dsn) == dsn
def test_leaves_keyword_dsn_unchanged(self):
dsn = "host=localhost dbname=deerflow"
assert normalize_libpq_dsn(dsn) == dsn
def test_rejects_non_postgres_scheme(self):
with pytest.raises(ValueError, match="Unsupported PostgreSQL DSN scheme"):
normalize_libpq_dsn("mysql://localhost/db")