"""Monthly evidence pre-compile. The point of the job is that evidence for a period exists from within that period. Two properties make that true rather than merely intended: it runs over the **closed** month, and it fails loudly when a query cannot be answered. """ from __future__ import annotations import uuid from datetime import UTC, date, datetime, timedelta import pytest from sqlalchemy import select from app.core.access_cycle_listener import register_listener from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.domains.auditor.models import AuditorEvidenceBundle from app.scheduled_jobs.evidence_precompile_handler import ( closed_month_window, precompilable_queries, run_evidence_precompile, ) register_listener() def test_window_is_the_previous_whole_month(): """Running over the CURRENT month would freeze a partial picture and stamp it with a hash; the next run would then differ for no reason other than time passing, which is the one meaning the hash must keep reserved for "the underlying data changed".""" assert closed_month_window(date(2026, 3, 15)) == ("2026-02-01", "2026-02-28") assert closed_month_window(date(2026, 3, 1)) == ("2026-02-01", "2026-02-28") # Year boundary and leap February. assert closed_month_window(date(2026, 1, 9)) == ("2025-12-01", "2025-12-31") assert closed_month_window(date(2024, 3, 5)) == ("2024-02-01", "2024-02-29") def test_precompilable_set_is_derived_not_listed(): """A hard-coded list stops covering the catalogue the moment someone registers a new query — which is exactly when nobody is looking.""" keys = {d.key for d in precompilable_queries()} assert "software_seat_roster" in keys assert "privileged_role_roster" in keys # Subject-scoped questions cannot be run company-wide. assert "person_footprint" not in keys assert "revocation_proof" not in keys assert "object_lifecycle" not in keys @pytest.mark.asyncio async def test_precompile_freezes_a_bundle_per_company_wide_query( db_session, employee_user, ): employee, _ = employee_user software = AssetSoftwareCatalog(name=f"PreSW {uuid.uuid4().hex[:6]}") db_session.add(software) await db_session.flush() plan = SoftwareLicensePlan( software_id=software.id, name="Pro", price=1, billing_cycle="monthly", ) db_session.add(plan) await db_session.flush() db_session.add(SoftwareSeatAllocation( plan_id=plan.id, user_id=employee.id, status="active", allocated_at=datetime(2026, 5, 10, tzinfo=UTC), )) await db_session.flush() summary = await run_evidence_precompile(db_session, {"month": "2026-05"}) assert summary["period_start"] == "2026-05-01" assert summary["period_end"] == "2026-05-31" assert summary["queries_compiled"] == len(precompilable_queries()) assert summary["model_calls"] == 0 # Control coverage comes from the catalogue's own declarations, not a # second mapping table that would drift. assert "CC6.1" in summary["controls_evidenced"] stored = (await db_session.execute( select(AuditorEvidenceBundle).where( AuditorEvidenceBundle.generated_via == "scheduled", ) )).scalars().all() assert len(stored) == summary["queries_compiled"] seat_bundle = next( b for b in stored if b.query_key == "software_seat_roster" ) assert seat_bundle.row_count == 1 assert seat_bundle.generated_by_person_id is None @pytest.mark.asyncio async def test_precompile_can_be_limited_to_named_queries(db_session): summary = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["privileged_role_roster"]}, ) assert summary["queries_compiled"] == 1 assert summary["bundles"][0]["query_key"] == "privileged_role_roster" @pytest.mark.asyncio async def test_a_failing_query_fails_the_run(db_session, monkeypatch): """A run that reports ok while an exhibit is missing would let a monthly control look healthy for months.""" from app.scheduled_jobs import evidence_precompile_handler as handler async def _boom(*_args, **_kwargs): raise RuntimeError("source unavailable") monkeypatch.setattr(handler, "run_evidence_query", _boom) with pytest.raises(RuntimeError, match="evidence pre-compile"): await run_evidence_precompile(db_session, {"month": "2026-05"}) @pytest.mark.asyncio async def test_point_in_time_queries_get_the_month_end(db_session): """"Who held admin in May" is answered as of the end of May, not as an interval — the ledger query takes a single instant.""" summary = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["privileged_role_roster"]}, ) bundle = await db_session.get( AuditorEvidenceBundle, uuid.UUID(summary["bundles"][0]["bundle_id"]), ) assert bundle.params["at"] == "2026-05-31" @pytest.mark.asyncio async def test_rerunning_the_same_month_is_a_no_op(db_session): """A backfill must not double the month's evidence. ``config.month`` exists so a missed month can be re-run by hand, and the job runner retries — so an unconditional INSERT stacked a second identical bundle per query and the pack section then rendered each exhibit twice. The second run finds this month's scheduled bundle and skips it. Only the SCHEDULED producer is idempotent. An auditor asking the same question twice by hand still gets two bundles: two hashes that differ IS the finding, and collapsing them would delete the signal. """ first = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["sso_grant_roster"]}, ) second = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["sso_grant_roster"]}, ) assert first["queries_compiled"] == 1 assert second["queries_compiled"] == 0 assert second["queries_already_present"] == 1 @pytest.mark.asyncio async def test_a_point_in_time_query_is_idempotent_too(db_session): """The half the first idempotency fix missed. Point-in-time questions are asked ``at`` the month's LAST day, and ``_period_bounds`` writes that single instant into both period columns. Comparing ``period_start.date()`` against the month's FIRST day therefore never matched, so ``privileged_role_roster`` duplicated on every retry while the other eight skipped correctly — and the test meant to cover this used a period query, so it stayed green. """ first = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["privileged_role_roster"]}, ) second = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["privileged_role_roster"]}, ) assert first["queries_compiled"] == 1 assert second["queries_compiled"] == 0 assert second["queries_already_present"] == 1 @pytest.mark.asyncio async def test_an_unknown_query_key_raises_instead_of_compiling_nothing( db_session, ): """A typo selected nothing, compiled nothing, and recorded ``ok`` — a month with no evidence reported as a healthy run.""" with pytest.raises(RuntimeError, match="query_keys"): await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["sso_grant_rosterr"]}, ) @pytest.mark.asyncio async def test_job_is_registered_for_both_processes(): """Registered in ``main.py`` AND ``worker.py`` — the worker is what actually runs it, so a web-only import means the cron never fires.""" from pathlib import Path from app.scheduled_jobs.registry import get_handler assert get_handler("auditor.evidence_precompile") is not None root = Path(__file__).resolve().parents[2] / "app" for process in ("main.py", "worker.py"): source = (root / process).read_text(encoding="utf-8") assert "evidence_precompile_handler" in source, process @pytest.mark.asyncio async def test_scheduled_bundles_are_distinguishable_from_manual_runs( db_session, admin_user, ): """"Someone went looking" and "routine collection" are different facts about a control, and an auditor reads them differently.""" from app.domains.auditor.evidence_query.service import run_evidence_query admin, _ = admin_user manual = await run_evidence_query( db_session, query_key="sso_grant_roster", params={}, actor_user_id=admin.id, ) summary = await run_evidence_precompile( db_session, {"month": "2026-05", "query_keys": ["sso_grant_roster"]}, ) scheduled = await db_session.get( AuditorEvidenceBundle, uuid.UUID(summary["bundles"][0]["bundle_id"]), ) assert manual.generated_via == "manual" assert manual.generated_by_person_id == admin.id assert scheduled.generated_via == "scheduled" assert scheduled.generated_by_person_id is None @pytest.mark.asyncio async def test_backfill_of_an_old_month_is_supported(db_session): """Ops needs to fill a month that predates the job without editing code — the config override is the documented path.""" summary = await run_evidence_precompile( db_session, {"month": "2025-11", "query_keys": ["api_key_roster"]}, ) assert summary["period_start"] == "2025-11-01" assert summary["period_end"] == "2025-11-30" @pytest.mark.asyncio async def test_default_window_tracks_the_calendar(db_session): """No month override → the run must land on the previous month.""" expected = closed_month_window(datetime.now(UTC).date()) summary = await run_evidence_precompile( db_session, {"query_keys": ["api_key_roster"]}, ) assert (summary["period_start"], summary["period_end"]) == expected # Sanity: that window is entirely in the past. assert date.fromisoformat(summary["period_end"]) < ( datetime.now(UTC).date() + timedelta(days=1) )