diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py index d93a2e8..6afc26f 100644 --- a/faststream_outbox/autovacuum.py +++ b/faststream_outbox/autovacuum.py @@ -60,10 +60,10 @@ def outbox_autovacuum_ddl( ``schema`` defaults to ``None``, which renders an unqualified table name that resolves via the connection's ``search_path`` -- matching both ``Table.schema=None`` and - the ``validate_schema(check_autovacuum=True)`` reloptions lookup's - ``COALESCE(:schema, current_schema())``. Pass the same ``schema`` as the outbox - ``Table`` (e.g. ``table.schema``) when it lives in a named schema, so this DDL - targets the same table the check reads. + the ``validate_schema(check_autovacuum=True)`` reloptions lookup, which resolves the + table the same way. Pass the same ``schema`` as the outbox ``Table`` (e.g. + ``table.schema``) when it lives in a named schema, so this DDL targets the same table + the check reads. """ quoted_table = _IDENTIFIER_PREPARER.quote(table_name) quoted_name = quoted_table if schema is None else f"{_IDENTIFIER_PREPARER.quote(schema)}.{quoted_table}" @@ -82,8 +82,7 @@ def outbox_autovacuum_ddl( # reloptions come back from asyncpg as a ``list[str]`` of ``"key=value"`` items, or -# None when the table has no options set (or does not exist). NULL nspname match uses -# current_schema() so a search_path-relative table resolves the same way the app does. +# None when the table has no options set (or does not exist). _RELOPTIONS_QUERY = text( "SELECT c.reloptions FROM pg_class c " "JOIN pg_namespace n ON n.oid = c.relnamespace " diff --git a/faststream_outbox/schema_validation.py b/faststream_outbox/schema_validation.py index 055d734..8441a26 100644 --- a/faststream_outbox/schema_validation.py +++ b/faststream_outbox/schema_validation.py @@ -66,22 +66,24 @@ async def validate_schema( ``False`` never checks it. """ async with engine.connect() as conn: - errors = await conn.run_sync(_validate_schema_sync, table) + schema = await conn.run_sync(_resolve_schema, table) + errors = await conn.run_sync(_validate_schema_sync, table, schema) # S2 / lease_ck: these two probes catch drift that `alembic revision # --autogenerate` cannot remediate — its index comparator ignores # postgresql_where and it has no check-constraint comparator at all. # Collect them separately so the raised error can point operators at # the hand-written-migration recipe (_AUTOGEN_BLIND_HINT) only when # one of them actually fired. - blind_errors = await conn.run_sync(_validate_index_predicates_sync, table) - blind_errors.extend(await conn.run_sync(_validate_check_constraints_sync, table)) + blind_errors = await conn.run_sync(_validate_index_predicates_sync, table, schema) + blind_errors.extend(await conn.run_sync(_validate_check_constraints_sync, table, schema)) errors.extend(blind_errors) if dlq_table is not None: - errors.extend(await conn.run_sync(_validate_dlq_schema_sync, dlq_table)) + dlq_schema = await conn.run_sync(_resolve_schema, dlq_table) + errors.extend(await conn.run_sync(_validate_dlq_schema_sync, dlq_table, dlq_schema)) autovacuum_errors: list[str] = [] if check_autovacuum: reloptions = ( - await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema}) + await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": schema}) ).scalar_one_or_none() autovacuum_errors = autovacuum_findings(table.name, reloptions) message_parts: list[str] = [] @@ -93,6 +95,24 @@ async def validate_schema( raise RuntimeError("\n\n".join(message_parts)) +_RESOLVE_SCHEMA_QUERY = text( + "SELECT n.nspname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE c.oid = to_regclass(format('%I', CAST(:table AS text)))", +) + + +def _resolve_schema(connection: "Connection", table: "Table") -> str | None: + """Return the schema the app's queries reach *table* in. + + An unqualified table resolves through the whole ``search_path``, not just ``current_schema()``, + so ask Postgres where the name lands. ``None`` means it doesn't resolve; the probes then fall + back to ``current_schema()`` and report the table missing. + """ + if table.schema is not None: + return table.schema + return connection.execute(_RESOLVE_SCHEMA_QUERY, {"table": table.name}).scalar_one_or_none() + + def _normalize_predicate(predicate: str) -> str: """Canonicalize a Postgres partial-index predicate for comparison (lowercase, drop parens/space).""" return " ".join(predicate.lower().replace("(", " ").replace(")", " ").split()) @@ -123,7 +143,7 @@ def _normalize_predicate(predicate: str) -> str: ) -def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> list[str]: +def _validate_index_predicates_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]: """Compare the live partial-index WHERE predicates against what the package expects (S2). Alembic's index diff ignores ``postgresql_where``, so the alembic autogenerate pass @@ -140,7 +160,7 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> rows = ( connection.execute( _INDEX_PREDICATE_QUERY, - {"table": table.name, "schema": table.schema}, + {"table": table.name, "schema": schema}, ) .mappings() .all() @@ -180,7 +200,7 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> } -def _validate_check_constraints_sync(connection: "Connection", table: "Table") -> list[str]: +def _validate_check_constraints_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]: """Verify the live DB carries a CHECK enforcing each invariant the package needs. Alembic's ``compare_metadata`` registers no check-constraint comparator, so a missing @@ -200,7 +220,7 @@ def _validate_check_constraints_sync(connection: "Connection", table: "Table") - rows = ( connection.execute( _CHECK_CONSTRAINT_QUERY, - {"table": table.name, "schema": table.schema}, + {"table": table.name, "schema": schema}, ) .mappings() .all() @@ -241,19 +261,20 @@ def _compose_schema_mismatch_message(errors: list[str], *, has_blind_drift: bool return msg -def _validate_schema_sync(connection: "Connection", table: "Table") -> list[str]: +def _validate_schema_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]: """Run the outbox-table validation pass; see :func:`_run_validate` for the diff machinery.""" - return _run_validate(connection, table, make_outbox_table) + return _run_validate(connection, table, schema, make_outbox_table) -def _validate_dlq_schema_sync(connection: "Connection", table: "Table") -> list[str]: +def _validate_dlq_schema_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]: """Run the DLQ-table validation pass; see :func:`_run_validate` for the diff machinery.""" - return _run_validate(connection, table, make_dlq_table) + return _run_validate(connection, table, schema, make_dlq_table) def _run_validate( connection: "Connection", table: "Table", + schema: str | None, canonical_factory: "Callable[[MetaData, str], Table]", ) -> list[str]: """Run Alembic's autogenerate diff against the live DB and surface any "missing schema" drift. @@ -286,7 +307,7 @@ def _run_validate( # user's schema onto the canonical copy so the autogenerate diff compares # ``app.outbox`` against ``app.outbox`` rather than the default search_path — # matching the schema-qualified DLQ CTE in ``client._build_dlq_cte`` (B10). - canonical_metadata = MetaData(schema=table.schema) + canonical_metadata = MetaData(schema=schema) canonical_factory(canonical_metadata, table.name) # Alembic reports the connection's DEFAULT schema to ``include_name`` as ``None`` (with @@ -295,7 +316,7 @@ def _run_validate( # search_path) must be normalized to ``None`` before comparing, else ``name == "public"`` # never matches Alembic's ``None`` and a CORRECT table falsely reads as "does not exist". default_schema_name = connection.dialect.default_schema_name - target_schema = None if table.schema == default_schema_name else table.schema + target_schema = None if schema == default_schema_name else schema def _include_name(name: str | None, type_: str, parent_names: "Mapping[str, str | None]") -> bool: # ``include_schemas=True`` makes Alembic enumerate EVERY schema and call this with diff --git a/tests/test_integration.py b/tests/test_integration.py index 0f99b0a..2287ffd 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -167,6 +167,36 @@ async def test_validate_schema_passes_for_explicitly_named_default_schema(pg_eng await conn.run_sync(metadata.drop_all) +async def test_validate_schema_finds_a_table_through_a_later_search_path_entry(pg_engine: AsyncEngine) -> None: + """An unqualified table that resolves through a later ``search_path`` entry validates. + + With ``search_path = , public`` the app reads and writes ``public.``, + so validation must look there too, not only in ``current_schema()``. + """ + schema = f"sch_sp_{uuid.uuid4().hex[:8]}" + suffix = uuid.uuid4().hex[:8] + metadata = MetaData() + table = make_outbox_table(metadata, table_name=f"outbox_sp_{suffix}") + dlq_table = make_dlq_table(metadata, table_name=f"outbox_sp_dlq_{suffix}") + async with pg_engine.begin() as conn: + await conn.execute(text(f'CREATE SCHEMA "{schema}"')) + await conn.run_sync(metadata.create_all) + await conn.execute(text(outbox_autovacuum_ddl(table.name))) + dsn = pg_engine.url.render_as_string(hide_password=False) + search_path_engine = create_async_engine( + dsn, + connect_args={"server_settings": {"search_path": f'"{schema}", public'}}, + ) + try: + client = OutboxClient(search_path_engine, table, dlq_table=dlq_table) + await client.validate_schema(check_autovacuum=True) # must NOT raise + finally: + await search_path_engine.dispose() + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + + async def test_validate_schema_fails_for_missing_table(pg_engine) -> None: metadata = MetaData() table = make_outbox_table(metadata, table_name="does_not_exist_xyz") diff --git a/tests/test_unit.py b/tests/test_unit.py index 8d58b47..6e372a3 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -1411,7 +1411,7 @@ def test_validate_schema_sync_raises_when_alembic_missing() -> None: patch("faststream_outbox.schema_validation.is_alembic_installed", new=False), pytest.raises(ImportError, match=r"pip install 'faststream-outbox\[validate\]'"), ): - _validate_schema_sync(MagicMock(), t) + _validate_schema_sync(MagicMock(), t, None) def test_compose_schema_mismatch_message_appends_hint_on_blind_drift() -> None: @@ -1461,7 +1461,7 @@ def test_validate_check_constraints_passes_under_convention_resolved_name() -> N connection = _mock_check_constraint_connection( [{"name": "ck_outbox_faststream_outbox_faststream_lease_ck", "definition": _LEASE_CK_DEFINITION}], ) - assert _validate_check_constraints_sync(connection, table) == [] + assert _validate_check_constraints_sync(connection, table, None) == [] def test_validate_check_constraints_passes_under_literal_name() -> None: @@ -1470,14 +1470,14 @@ def test_validate_check_constraints_passes_under_literal_name() -> None: connection = _mock_check_constraint_connection( [{"name": "outbox_faststream_lease_ck", "definition": _LEASE_CK_DEFINITION}], ) - assert _validate_check_constraints_sync(connection, table) == [] + assert _validate_check_constraints_sync(connection, table, None) == [] def test_validate_check_constraints_missing_describes_predicate() -> None: """When no live CHECK enforces the predicate, the error names the invariant, not a guessed name.""" table = make_outbox_table(MetaData(), table_name="outbox_faststream") connection = _mock_check_constraint_connection([]) # constraint absent from the live DB - errors = _validate_check_constraints_sync(connection, table) + errors = _validate_check_constraints_sync(connection, table, None) assert errors == [ ( f"missing CHECK constraint enforcing '{_LEASE_CK_PREDICATE}' " @@ -1493,7 +1493,7 @@ def test_validate_check_constraints_drifted_predicate_reported_as_missing() -> N connection = _mock_check_constraint_connection( [{"name": "outbox_faststream_lease_ck", "definition": drifted}], ) - errors = _validate_check_constraints_sync(connection, table) + errors = _validate_check_constraints_sync(connection, table, None) assert errors == [ ( f"missing CHECK constraint enforcing '{_LEASE_CK_PREDICATE}' "