69 lines
3.0 KiB
Python
69 lines
3.0 KiB
Python
|
"""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"
|