Testing

When writing pytest (Python):

• No fake tests — every assertion must test real behavior (invariants, boundaries, error semantics)
• Mock ONLY at system boundaries: DB, external HTTP, filesystem, clock. Never mock internal logic
• Real objects for domain types — instantiate real pydantic schemas, ORM rows, internal models with real values. MagicMock for a domain object silently accepts any attribute access and lets schema breaks pass — production-only failure mode. Mocks remain ok for system boundaries (DB session, HTTP client, scraper, logger).
• Test factories live in conftest — when a test needs a real model/schema/ORM object, add a make_* factory fixture to tests/conftest.py, not a private _make_* helper inside a single test file. Use via dependency injection: def test_x(make_main_table_client): .... The factory takes **overrides so each test can customize.
• One test file per production module — mirror the production layout. Prefer tests/test_<module>.py covering everything in app/.../<module>.py over splitting by concern (e.g. test_<module>_report.py + test_<module>_schedules.py). Split files drift apart, duplicate helpers, and hide shared fixtures. Consolidate before adding new tests when a module already has multiple test files.
• Imports at the top of the file — never from foo import Bar inside a test body.
• Use existing fixtures from conftest.py before creating new ones
• When fixing failing tests: fix test inputs/data to match real behavior — do not add more mocks
• Delete flaky tests outright — do not weaken assertions to make them pass
• HTTP response tests assert status code AND headers, not just body. When testing a route, assert response.status_code AND any contract-meaningful headers (X-Cache, Retry-After, WWW-Authenticate, content-type, etc.) — not just the body shape. Weak assertions like status != 200 or X-Cache != "HIT" miss real bugs (e.g. a 201 flattened to 200 on cache replay, or a header silently dropped). Pin the exact value.
• Don't hardcode enum/constant copies in fixtures. When a fixture needs an enum value (status, role, type code), import it from the schema/model module, OR add a one-line equality assertion (assert set(FIXTURE_STATUSES) == {s.value for s in StatusEnum}) that fails when the schema drifts EITHER WAY. A subset assertion (<=) only catches harness-has-invalid-value drift; it silently passes when the schema adds a value the harness doesn't cover (verified: a schema added UNPLUGGED=13, the harness still had [1..12], every subset test stayed green, the harness silently stopped covering one production state). A standalone list of "looks right" values is a silent contract test for a contract that doesn't exist — and pydantic / DB constraints will reject the drift in production while every unit test passes.
• Assert every field on the returned contract, not a spot-check subset. When a test inspects a returned object (mapper output, service return, response body, cached envelope), assert EVERY field on the return schema with an explicit value+type assertion — for the happy-path fixture and for every meaningful variant. Nested objects (embedded relations, status-history rows) get the same treatment recursively. Pair the per-field asserts with a one-line field-set-equality guard: assert set(response.keys()) == set(Schema.__fields__.keys()). Column-iteration drift-guard loops (for col in __table__.columns: assert getattr(row, col.name) is not None) are additive — they catch newly-added fields being dropped, but they silently pass when TWO fields swap mappings because both still hold some value. Three complementary layers: per-field pins the intended value, set-equality catches additions, the drift loop catches removals. Any partial spot-check ships the wrong-value bug to prod.
• Match event_loop fixture scope to your async singletons' lifetime. If the suite touches any process-cached async resource — a SQLAlchemy async engine per client_name, a pooled HTTP client, a kafka producer, a redis client — override pytest-asyncio's default function-scoped event_loop with a session-scoped fixture in conftest.py (and set asyncio_default_fixture_loop_scope = session in pytest.ini for pytest-asyncio ≥ 0.24). Otherwise pytest-asyncio spins up a new loop per test, the singleton stays attached to the FIRST loop, and later tests blow up with RuntimeError: got Future <...> attached to a different loop — a lifecycle mismatch that looks like a race or flake. Rule of thumb: fixture scope must be ≥ the resource's cache scope. If the singleton is torn down between tests (via a fresh-per-function fixture), function scope is fine — the mismatch, not either scope, is the bug.
• Tapping/seeding a queue shared with a live consumer: account for delivery semantics. When an integration/E2E test reads (or asserts on) a message queue that a running service also consumes, the transport's delivery model decides whether your tap is even possible — a single-delivery queue (SQS) hands each message to exactly ONE reader, so a naive tap races the real consumer and one of them loses the message. See references/localstack-integration.md §"Tapping a queue shared with a live consumer" for the patterns (non-destructive SQS peek, pause-the-consumer, Kafka unique-group fan-out, purge-stale-then-poll-full-timeout for windowed producers).

When invoked to fix failing tests (not write new ones):

1. NO production code changes — fix tests only (unless explicitly told otherwise)
2. Debug first: print what the function actually returns before asserting
3. Fix test inputs/data to match real behavior — do not add more mocks
4. Use existing conftest fixtures — do not create duplicates
5. Remove any @pytest.mark.skip once the test is passing