"""Per-test database setup for the Cyclone test suite. Every test gets a fresh SQLite DB at ``tmp_path/test.db`` so the suite can run in parallel and never touches the user's ``~/.local/share/cyclone`` database. ``db.init_db()`` is idempotent — tests that already call it explicitly are unaffected (the second call short-circuits). This fixture exists because the SQLAlchemy-backed ``CycloneStore`` requires the engine to be initialized before any DB access, whereas the old in-memory store did not. Adding the init here keeps existing test modules (test_api.py, test_api_835.py, test_api_gets.py, test_api_parse_persists.py) working unchanged. """ from __future__ import annotations from datetime import datetime, timezone from decimal import Decimal import pytest @pytest.fixture(autouse=True) def _auto_init_db(tmp_path, monkeypatch): """Point CYCLONE_DB_URL at a per-test SQLite file and init the schema. Also wires a fresh ``EventBus`` onto ``app.state`` because ``TestClient`` does not invoke the FastAPI lifespan handler unless used as a context manager. The bus is reset between tests so subscribers don't leak. Auth gating is enabled at the router/endpoint level via the ``Depends(matrix_gate)`` wiring in ``cyclone.api``. The auth tests (``test_auth_*``) explicitly flip ``AUTH_DISABLED = False`` and authenticate via the public login route to exercise the real auth path. Every other test — the original test suite predates auth — gets ``AUTH_DISABLED = True`` here so the existing tests keep working without each one having to login first. """ monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") from cyclone import db from cyclone.pubsub import EventBus from cyclone.auth import deps db._reset_for_tests() db.init_db() # Re-resolve `app` each fixture invocation because some tests # (test_cors_extra_origins_via_env) call ``importlib.reload`` on # ``cyclone.api`` to mutate the CORS allow-list. If we cached # the app reference at conftest module load, we'd be setting # ``event_bus`` on a stale instance that no test client is # actually using. from cyclone import api as _api_mod _api_mod.app.state.event_bus = EventBus() deps.AUTH_DISABLED = True # The rate-limit middleware keeps a per-IP sliding window in # ``_buckets`` (now a class-level dict — see # ``cyclone.security.RateLimitMiddleware._buckets``). Without a # reset between tests, later tests in a full-suite run get # ``429 Too Many Requests`` once the testclient IP exhausts its # 300 req/60s budget. Walk the middleware stack and clear the # buckets so every test starts with a fresh window. Because the # buckets dict is class-level, this clears every instance at # once — important when tests like ``test_cors_extra_origins_via_env`` # trigger ``importlib.reload(cyclone.api)`` mid-suite. # Trigger the stack build with a cheap health probe (the only # request exempt from the limiter — see RateLimitMiddleware.EXEMPT_PATHS). _reset_rate_limit_buckets(_api_mod.app) try: yield finally: deps.AUTH_DISABLED = False _api_mod.app.state.event_bus = None _reset_rate_limit_buckets(_api_mod.app) db._reset_for_tests() def _reset_rate_limit_buckets(app) -> None: """Clear the rate-limit middleware's per-IP sliding-window buckets. SP27 Task 13b follow-up: ``RateLimitMiddleware._buckets`` is shared across all tests in a process (TestClient reuses the same ``app`` instance), so without a reset between tests the full suite trips the limiter after ~300 requests and later tests get 429s. As of the SP38 followup the buckets dict is class-level, so this helper clears every ``RateLimitMiddleware`` instance at once (important when ``importlib.reload(cyclone.api)`` mid-suite has created a new instance and tests still hold a stale ``app`` reference to the old one — see ``tests/test_rate_limit_shared_buckets.py``). The middleware stack is only built on the first request, so we prime it with an exempt health probe before walking to the RateLimit layer. If the stack ever stops being a single chain of ``.app`` links, this helper raises AttributeError — better to fail loudly than silently leak state. """ from cyclone.security import RateLimitMiddleware # Class-level dict — one clear reaches every instance. RateLimitMiddleware._buckets.clear() # --------------------------------------------------------------------------- # SP31: shared DB-session + Claim/Remit factory fixtures. # # `db_session` yields a fresh session per test against the per-test DB set # up by the autouse `_auto_init_db` fixture above. The session rolls back at # teardown so a test that mutates rows doesn't leak into siblings. # # `make_claim` / `make_remit` build ORM rows with a small ergonomic surface: # the planned parameter names follow the content-keys helper (PCN, charge, # rendering NPI), but the ORM attribute names differ (`charge_amount` on # Claim, `total_charge` on Remittance, `provider_npi` on Claim — and no # `payer_id` / `rendering_provider_npi` column on Remittance at all). The # factories map the planned names onto the real ORM attributes and stash # `rendering_provider_npi` as a transient attribute so # ``reconcile._content_keys_match`` can still read it via ``getattr``. # --------------------------------------------------------------------------- @pytest.fixture def db_session(): """Yield a fresh SQLAlchemy session, rolling back at teardown.""" from cyclone import db as _db session = _db.SessionLocal()() try: yield session session.rollback() finally: session.close() def _ensure_batch(session, batch_id: str = "test-batch") -> None: """Create the Batch row a Claim/Remittance FKs to (idempotent).""" from cyclone.db import Batch if session.get(Batch, batch_id) is None: session.add(Batch( id=batch_id, kind="837p", input_filename="test.txt", parsed_at=datetime.now(timezone.utc), )) session.flush() @pytest.fixture def make_claim(db_session): """Factory: build & flush a Claim with the planned content-keys params. Planned params (match the SP31 spec / content-keys test surface): patient_control_number, total_charge, rendering_provider_npi, service_date_from, matched_remittance_id=None, state=None, claim_id=None `state=None` defaults to ``ClaimState.SUBMITTED`` so existing tests that omit it keep behaving. `rendering_provider_npi` is wired through the real ``Claim.rendering_provider_npi`` column (SP32 migration 0019). """ def _make( patient_control_number: str, total_charge, rendering_provider_npi: str, service_date_from, matched_remittance_id=None, state=None, claim_id: str | None = None, ): from cyclone.db import Claim, ClaimState as _CS _ensure_batch(db_session) cid = claim_id or f"clm-{patient_control_number}-{service_date_from.isoformat()}" c = Claim( id=cid, batch_id="test-batch", patient_control_number=patient_control_number, service_date_from=service_date_from, charge_amount=total_charge, state=state if state is not None else _CS.SUBMITTED, matched_remittance_id=matched_remittance_id, rendering_provider_npi=rendering_provider_npi, ) db_session.add(c) db_session.flush() return c return _make @pytest.fixture def make_remit(db_session): """Factory: build & flush a Remittance with the planned content-keys params. Planned params: payer_claim_control_number, total_charge_amount, rendering_provider_npi, service_date, remit_id=None `total_charge_amount` is mapped to ``Remittance.total_charge`` (real ORM field). `rendering_provider_npi` is wired through the real ``Remittance.rendering_provider_npi`` column (SP32 migration 0019). """ def _make( payer_claim_control_number: str, total_charge_amount, rendering_provider_npi: str, service_date, remit_id: str | None = None, ): from cyclone.db import Remittance _ensure_batch(db_session) rid = remit_id or f"remit-{payer_claim_control_number}-{service_date.isoformat()}" r = Remittance( id=rid, batch_id="test-batch", payer_claim_control_number=payer_claim_control_number, status_code="1", total_charge=total_charge_amount, total_paid=Decimal("0"), received_at=datetime.now(timezone.utc), service_date=service_date, is_reversal=False, rendering_provider_npi=rendering_provider_npi, ) db_session.add(r) db_session.flush() return r return _make