wehub-resource-sync
72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
"""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"
|