"""chain_deps._libpq_conn_string 单测——剥离 SQLAlchemy 驱动后缀(回归)。 bug:runtime checkpointer 工厂曾把 `postgresql+psycopg://…`(settings 的 sync URL) 原样喂给 `AsyncPostgresSaver.from_conn_string`,psycopg 不识别 `+psycopg` 方言段 → `ProgrammingError: missing "=" …` → 多章链 job 直接 failed。E2E 用 MemorySaver 覆盖, 从不走真 conn string,故此路径无回归保护。本用例锁住「剥离后缀」语义。 """ from __future__ import annotations import pytest from ww_api.services import chain_deps class _StubSettings: def __init__(self, url: str) -> None: self.database_url_sync = url @pytest.mark.parametrize( ("raw", "expected"), [ ( "postgresql+psycopg://writer:writer@localhost:5432/writer", "postgresql://writer:writer@localhost:5432/writer", ), ("postgresql+asyncpg://u:p@h:5432/db", "postgresql://u:p@h:5432/db"), ("postgresql://u:p@h:5432/db", "postgresql://u:p@h:5432/db"), ], ) def test_libpq_conn_string_strips_driver_suffix( monkeypatch: pytest.MonkeyPatch, raw: str, expected: str ) -> None: # Arrange monkeypatch.setattr(chain_deps, "get_settings", lambda: _StubSettings(raw)) # Act result = chain_deps._libpq_conn_string() # Assert assert result == expected assert "+psycopg" not in result assert "+asyncpg" not in result