From ae59fbcf047cf4adeaf6c9c53886db77a8044c7d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 03:51:39 +0000 Subject: [PATCH] test(reporting): run the reporting-subscription stores against PostgreSQL Covers PgReportingSubscriptionStateStore and PgPrincipalRecordStore on a real database: idempotent schema creation, compare-and-swap semantics, 20 concurrent writers losing nothing, and a restarted seller (a new PrincipalService and engine on the same database) serving the same get_principal readback and owing the same periods. The module skips without ADCP_PG_TEST_URL; CI's core Postgres lane runs it. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../test_reporting_subscriptions_pg.py | 147 ++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 tests/conformance/reporting/test_reporting_subscriptions_pg.py diff --git a/tests/conformance/reporting/test_reporting_subscriptions_pg.py b/tests/conformance/reporting/test_reporting_subscriptions_pg.py new file mode 100644 index 000000000..0d234c974 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_subscriptions_pg.py @@ -0,0 +1,147 @@ +"""PostgreSQL stores for AdCP 3.3 reporting subscriptions (#1695). + +Requires a real PostgreSQL instance (``ADCP_PG_TEST_URL``); the module skips +without one. Each test uses a fresh schema, so runs never collide. +""" + +from __future__ import annotations + +import asyncio +import os +import secrets +from collections.abc import AsyncIterator + +import pytest + +psycopg_pool = pytest.importorskip("psycopg_pool") + +TEST_URL = os.environ.get("ADCP_PG_TEST_URL") +if not TEST_URL: + pytest.skip("ADCP_PG_TEST_URL not set", allow_module_level=True) + +from adcp_reporting_storage import ( # noqa: E402 + PgPrincipalRecordStore, + PgReportingSubscriptionStateStore, +) + +from adcp.reporting.subscriptions import ReportingSubscriptionEngine # noqa: E402 +from adcp.server import PrincipalService # noqa: E402 +from tests.test_reporting_subscription_engine import ( # noqa: E402 + IDENTITY, + MANAGED, + OFFERINGS, + World, +) +from tests.test_reporting_subscription_state_store import ( # noqa: E402 + CAPACITY, + _observed, + _scenario, +) + + +@pytest.fixture +async def pool() -> AsyncIterator[psycopg_pool.AsyncConnectionPool]: + schema = f"test_subs_{secrets.token_hex(6)}" + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, min_size=1, max_size=8, open=False, kwargs={"autocommit": True} + ) as admin: + async with admin.connection() as connection: + await connection.execute(f"CREATE SCHEMA {schema}") + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, + min_size=1, + max_size=8, + open=False, + kwargs={"autocommit": True, "options": f"-c search_path={schema}"}, + ) as scoped: + yield scoped + async with psycopg_pool.AsyncConnectionPool( + TEST_URL, min_size=1, max_size=2, open=False, kwargs={"autocommit": True} + ) as admin: + async with admin.connection() as connection: + await connection.execute(f"DROP SCHEMA {schema} CASCADE") + + +def _world(pool) -> World: + world = World(support=CAPACITY) + records = PgPrincipalRecordStore(pool=pool) + world.principals = PrincipalService(store=records, emit_change=_record(world)) + world.engine = ReportingSubscriptionEngine( + principals=world.principals, + host=world.host, + offerings=OFFERINGS, + support=CAPACITY, + store=PgReportingSubscriptionStateStore(pool=pool), + clock=world.clock, + ) + world.principals.set_reporting_subscriptions(world.engine.principal_options()) + return world + + +def _record(world: World): + async def emit(change) -> None: + world.changes.append(change) + + return emit + + +async def test_schema_creation_is_idempotent(pool) -> None: + store = PgReportingSubscriptionStateStore(pool=pool) + await store.create_schema() + await store.create_schema() + + +async def test_state_compare_and_swap(pool) -> None: + store = PgReportingSubscriptionStateStore(pool=pool) + await store.create_schema() + assert await store.load("p") is None + assert await store.compare_and_swap("p", None, {"a": 1}) + assert not await store.compare_and_swap("p", None, {"a": 2}) + stored = await store.load("p") + assert stored is not None and (stored.document, stored.revision) == ({"a": 1}, 1) + assert not await store.compare_and_swap("p", 7, {"a": 3}) + assert await store.compare_and_swap("p", 1, {"a": 4}) + stored = await store.load("p") + assert stored is not None and (stored.document, stored.revision) == ({"a": 4}, 2) + + +async def test_concurrent_writers_never_lose_a_change(pool) -> None: + store = PgReportingSubscriptionStateStore(pool=pool) + await store.create_schema() + + async def bump() -> None: + while True: + stored = await store.load("p") + count = 0 if stored is None else stored.document["count"] + if await store.compare_and_swap( + "p", None if stored is None else stored.revision, {"count": count + 1} + ): + return + + await asyncio.gather(*(bump() for _ in range(20))) + stored = await store.load("p") + assert stored is not None and stored.document == {"count": 20} and stored.revision == 20 + + +async def test_a_restarted_seller_on_postgres_owes_the_same_periods(pool) -> None: + await PgReportingSubscriptionStateStore(pool=pool).create_schema() + world = _world(pool) + await _scenario(world) + before = await _observed(world, world.engine) + readback = await world.principals.get_principal(IDENTITY) + + restarted_principals = PrincipalService( + store=PgPrincipalRecordStore(pool=pool), emit_change=_record(world) + ) + restarted = ReportingSubscriptionEngine( + principals=restarted_principals, + host=world.host, + offerings=(*OFFERINGS, MANAGED), + support=CAPACITY, + store=PgReportingSubscriptionStateStore(pool=pool), + clock=world.clock, + ) + restarted_principals.set_reporting_subscriptions(restarted.principal_options()) + assert await restarted_principals.get_principal(IDENTITY) == readback + assert await _observed(world, restarted) == before + assert before["owed"][("files", "acct-1")]