05 — SQL / NoSQL for Test Data & State Validation

JD: "SQL/NoSQL for test data setup, teardown, and database state validation." Testing agents in finance means verifying the side effects landed correctly in the DB — not just what the API returned.

---

1. SQL you must be fluent in

-- filtering, sorting, limiting
SELECT id, amount, status FROM payments
WHERE status = 'pending' AND amount > 1000
ORDER BY created_at DESC
LIMIT 10;

-- joins (know INNER vs LEFT)
SELECT a.id, a.balance, u.email
FROM accounts a
INNER JOIN users u ON u.id = a.owner_id      -- only matching rows
LEFT  JOIN kyc  k ON k.user_id = u.id;       -- all accounts, kyc null if none

-- aggregation
SELECT status, COUNT(*), SUM(amount)
FROM payments GROUP BY status HAVING COUNT(*) > 5;

Join	Returns
INNER	rows matching in both tables
LEFT	all left rows + matched right (null if none)
RIGHT	all right + matched left
FULL	everything, nulls where unmatched

Also know: DISTINCT, subqueries, IN/EXISTS, NULL handling (IS NULL, COALESCE), window functions (ROW_NUMBER() OVER(...)) at a recognition level.

---

2. Setup / teardown patterns (the core JD ask)

Golden rule: each test starts from a known state and leaves nothing behind. Options, best→pragmatic:

a) Transaction rollback (fastest, cleanest)

Wrap each test in a transaction and roll back — DB never actually changes.

@pytest.fixture
def db_session(engine):
    conn = engine.connect()
    txn = conn.begin()
    session = Session(bind=conn)
    yield session
    session.close()
    txn.rollback()          # undo everything the test did
    conn.close()

b) Factory + explicit cleanup (see file 01 factory fixture)

Create rows, track them, delete in teardown.

c) Truncate/reseed between tests

Simple but slower; fine for small schemas.

d) Ephemeral DB via containers (integration)

testcontainers spins a throwaway Postgres/Mongo per session — real engine, isolated.

from testcontainers.postgres import PostgresContainer
@pytest.fixture(scope="session")
def pg():
    with PostgresContainer("postgres:16") as p:
        yield p.get_connection_url()

Interview line: "I prefer transaction-rollback isolation so tests are fast and never leak state; for true integration I use ephemeral containers so I test against the real engine, not a mock."

---

3. Database state validation (assert the side effect)

The API said "payment released" — prove the DB agrees and the audit row exists.

def test_payment_release_persists(client, db_session):
    r = client.post("/payments/42/release")
    assert r.status_code == 200

    # 1. state transition landed
    row = db_session.execute(
        text("SELECT status, released_at FROM payments WHERE id=:id"), {"id": 42}
    ).one()
    assert row.status == "released"
    assert row.released_at is not None

    # 2. no orphaned/duplicate side effects
    count = db_session.execute(
        text("SELECT COUNT(*) FROM ledger_entries WHERE payment_id=:id"), {"id": 42}
    ).scalar()
    assert count == 1          # exactly one ledger entry, not zero or two

    # 3. audit row emitted (file 12)
    audit = db_session.execute(
        text("SELECT event_type, correlation_id FROM audit_log WHERE entity_id=:id"),
        {"id": 42}
    ).one()
    assert audit.event_type == "PAYMENT_RELEASED"
    assert audit.correlation_id            # links the whole request chain

Senior signals: assert exactly one side effect (catches double-writes), assert timestamps/foreign keys, and assert the audit trail — not just the primary row.

---

4. Data integrity checks worth mentioning

• Referential integrity: no orphaned child rows (foreign keys intact).
• No duplicates: unique constraints honoured under retries/idempotency.
• Money precision: stored as NUMERIC/DECIMAL, never FLOAT (finance!).
• Soft-delete vs hard-delete: deleted rows flagged, not lost (audit).
• Consistency after a multi-step agent run: intermediate + final tables agree.

---

5. NoSQL basics

MongoDB (document store)

In plain words: stores JSON-like documents; flexible schema; query by field.

db.payments.find_one({"_id": 42})
db.payments.update_one({"_id": 42}, {"$set": {"status": "released"}})
db.payments.count_documents({"status": "pending"})
# aggregation pipeline (group/sum)
db.payments.aggregate([
    {"$match": {"status": "released"}},
    {"$group": {"_id": "$currency", "total": {"$sum": "$amount"}}},
])

State validation is the same idea: assert the document reached the expected shape/state. Watch eventual consistency — a read right after a write may lag; poll with a bounded retry rather than asserting instantly.

Redis (key-value / cache)

r.get("session:abc"); r.ttl("session:abc"); r.exists("lock:payment:42")

Test-relevant: cache invalidation, TTL/expiry, distributed locks (an agent shouldn't process the same job twice), rate-limit counters.

SQL vs NoSQL one-liner: "SQL for relational, transactional, audit-critical data with strong consistency and joins; NoSQL (Mongo) for flexible documents/high write throughput, Redis for cache/locks/ephemeral state — accepting eventual consistency."

---

6. Test-data strategy (senior view)

• Synthetic > production copy — never test on real customer PII (GDPR, file 12); generate with Faker.
• Deterministic seeds so tests are reproducible.
• Isolation — unique keys per test/worker so pytest -n auto parallel runs don't collide.
• Boundary data — min/max amounts, empty sets, unicode, timezone edges.

---

Rapid-fire recall

• Fast isolation = transaction rollback; real-engine isolation = testcontainers.
• State validation = assert DB row transition + exactly-one side effect + audit row.
• Money = DECIMAL, never float.
• Mongo/Redis: mind eventual consistency — bounded-retry reads.
• Synthetic data (Faker), deterministic seeds, unique keys for parallelism.