"""G3 -- async work happens after the response, via two independent consumer groups.""" import time import psycopg import pytest from harness.config import STREAM_NAME from harness.payloads import tx_payload pytestmark = pytest.mark.contract async def test_two_independent_consumer_groups_exist(redis_client): """G3: XINFO GROUPS reports exactly two groups, `notifier` and `audit`.""" groups = redis_client.xinfo_groups(STREAM_NAME) names = {g["name"] for g in groups} assert names == {"notifier", "audit"}, groups async def test_notification_and_audit_rows_appear(api, manifest, fresh_balances, dsn): """G3: after a successful transaction, rows appear in both `notifications` and `audit_events` for that transaction_id within 10s -- proving two independent consumer groups, not one consumer doing both jobs.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) resp = await api.post("/api/transactions", json=tx_payload(account_id, merchant_name="Async Test")) assert resp.status_code == 201 tx_id = resp.json()["transaction_id"] deadline = time.monotonic() + 10 found_notification = found_audit = False with psycopg.connect(dsn, autocommit=True) as conn: while time.monotonic() < deadline and not (found_notification and found_audit): with conn.cursor() as cur: cur.execute("SELECT 1 FROM notifications WHERE transaction_id = %s", (tx_id,)) found_notification = cur.fetchone() is not None cur.execute("SELECT 1 FROM audit_events WHERE transaction_id = %s", (tx_id,)) found_audit = cur.fetchone() is not None if not (found_notification and found_audit): time.sleep(0.5) assert found_notification, f"no notifications row for {tx_id} within 10s" assert found_audit, f"no audit_events row for {tx_id} within 10s" async def test_response_returns_before_notification_exists_at_least_once(api, manifest, fresh_balances, dsn): """G3: across 20 attempts, at least once the API response returns before the notification row exists -- proving the work is genuinely off the request path. Tolerant of a fast machine winning the race sometimes.""" account_id = manifest["clean_account_id"] fresh_balances(account_id) saw_a_miss = False with psycopg.connect(dsn, autocommit=True) as conn: for _ in range(20): resp = await api.post( "/api/transactions", json=tx_payload(account_id, amount_minor=100, merchant_name="Race Test") ) assert resp.status_code == 201 tx_id = resp.json()["transaction_id"] with conn.cursor() as cur: cur.execute("SELECT 1 FROM notifications WHERE transaction_id = %s", (tx_id,)) if cur.fetchone() is None: saw_a_miss = True break assert saw_a_miss, "notification row existed immediately on every one of 20 attempts"