deer-flow/backend/tests/test_persistence_bootstrap_url.py

"""Tests for the Postgres URL / ConfigParser pitfalls in ``bootstrap``.

Two failure modes the ``_alembic_safe_url`` helper exists to prevent:

1. ``str(engine.url)`` (and the default ``URL.render_as_string()``) masks the
   password as ``***``. The live engine would still work because it carries
   the password in-memory, but alembic ``stamp`` / ``upgrade`` (which open
   their own connection from the URL we pass in) would authenticate with
   garbage and fail at runtime.
2. ``alembic.config.Config.set_main_option`` forwards to ``ConfigParser.set``,
   which performs ``%(name)s``-style interpolation on the value. A URL-encoded
   password containing ``%`` (e.g. ``p%40ss`` for ``p@ss``) raises
   ``InterpolationSyntaxError``. Every literal ``%`` must be doubled.
"""

from __future__ import annotations

from types import SimpleNamespace

from sqlalchemy.engine.url import make_url

from deerflow.persistence.bootstrap import _alembic_safe_url, _escape_url_for_alembic, _get_alembic_config


def _fake_engine(url: str) -> SimpleNamespace:
    """Build a minimal stand-in for ``AsyncEngine`` so we don't need a real
    driver (e.g. asyncpg) installed just to exercise the URL path."""
    return SimpleNamespace(url=make_url(url))


def test_safe_url_preserves_password_for_postgres() -> None:
    engine = _fake_engine("postgresql://alice:s3cret@db.example.com/app")
    safe = _alembic_safe_url(engine)
    assert "s3cret" in safe, "password got masked: alembic would auth with garbage"
    assert "***" not in safe


def test_safe_url_escapes_percent_for_configparser() -> None:
    # URL-encoded ``@`` in password -> raw ``%40`` in URL -> ConfigParser
    # would treat it as an interpolation marker.
    engine = _fake_engine("postgresql://alice:p%40ss@db.example.com/app")
    safe = _alembic_safe_url(engine)
    assert "p%%40ss" in safe, f"percent not doubled, ConfigParser will fail: {safe}"


def test_alembic_config_accepts_url_with_percent_and_round_trips() -> None:
    # The whole point: build_config should not raise, and the URL alembic
    # reads back should match the original (single ``%``, real password).
    original = "postgresql://alice:p%40ss@db.example.com/app"
    engine = _fake_engine(original)
    cfg = _get_alembic_config(engine)
    roundtrip = cfg.get_main_option("sqlalchemy.url")
    assert roundtrip == original, f"alembic sees a different URL than we set: {roundtrip}"


def test_sqlite_url_does_not_double_percent_unnecessarily() -> None:
    # No percent in the URL -> no escaping needed -> output equals input.
    engine = _fake_engine("sqlite+aiosqlite:///tmp/db.sqlite")
    safe = _alembic_safe_url(engine)
    assert safe == "sqlite+aiosqlite:///tmp/db.sqlite"


def test_escape_url_for_alembic_doubles_only_percent_signs() -> None:
    # Shared helper used by both ``bootstrap._alembic_safe_url`` and
    # ``scripts/_autogen_revision._alembic_config`` -- pins the round-trip
    # rule so any future URL/ConfigParser corner case is fixed in one place.
    assert _escape_url_for_alembic("postgresql://a:p%40ss@h/d") == "postgresql://a:p%%40ss@h/d"
    assert _escape_url_for_alembic("sqlite:///x.db") == "sqlite:///x.db"
    # Idempotency is intentionally NOT a property -- doubling is one-way;
    # callers must escape exactly once on the way into set_main_option.
    assert _escape_url_for_alembic("a%%b") == "a%%%%b"