import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal import pytest from sqlalchemy import func, select from app.auth.models import AuthRole, AuthUserRole from app.domains.asset.models import ( AssetItem, AssetLocation, AssetLocationAdmin, ) from app.domains.cmdb.models import ( SCOPE_STATUS_SUSPENDED, CmdbCloudScope, CmdbCostFact, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration, PlatformSetting from app.intelligence.models import IntelligenceAnalysis, IntelligenceSnapshot from app.intelligence.provider_registry import ( BUSINESS_AUDIENCE_KEY, PLATFORM_DIMENSIONS, platform_providers, ) from app.intelligence.regional_audiences import ( regional_asset_audience, regional_asset_audiences, ) from app.observability.client_errors import ClientErrorDailyAggregate from app.scheduled_jobs.intelligence_handlers import ( _CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, run_cloud_scope_baseline_bootstrap, run_cloud_scope_baseline_daily, run_intelligence_snapshot_daily, ) from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun from app.scheduled_jobs.registry import get_handler def test_platform_registry_is_explicit_and_complete(): providers = platform_providers() assert tuple(row.dimension for row in providers) == PLATFORM_DIMENSIONS assert len(set(PLATFORM_DIMENSIONS)) == 8 assert get_handler("intelligence.snapshot_daily") is not None assert get_handler("intelligence.cloud_scope_baseline_daily") is not None assert get_handler(_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY) is not None # The daily fleet refresh may take time and must stay behind the heavy # gate. The release warm-up is deliberately light so it cannot wait for # that gate while keeping its single-flight lock and leave employees with # an unexplained empty subscription analysis. assert get_handler("intelligence.snapshot_daily").weight == "heavy" assert get_handler("intelligence.cloud_scope_baseline_daily").weight == "heavy" assert get_handler(_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY).weight == "light" @pytest.mark.asyncio async def test_cloud_scope_baseline_daily_precomputes_all_active_scopes_without_model( db_session, ): integration = Integration( kind="cloud", provider="aws", name="Baseline AWS", slug="baseline-aws", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() active = CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id="baseline-active", display_name="Baseline active", status="active", ) suspended = CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id="baseline-suspended", display_name="Baseline suspended", status=SCOPE_STATUS_SUSPENDED, ) db_session.add_all([active, suspended]) await db_session.commit() first = await run_cloud_scope_baseline_daily(db_session, {}) second = await run_cloud_scope_baseline_daily(db_session, {}) assert first["scope_count"] == 1 assert first["snapshots_created"] == 1 assert first["snapshots_reused"] == 0 assert first["failed_scope_count"] == 0 assert first["model_calls"] == 0 assert second["snapshots_created"] == 0 assert second["snapshots_reused"] == 1 rows = list((await db_session.execute( select(IntelligenceSnapshot).where( IntelligenceSnapshot.subject_type == "cloud_scope", ) )).scalars()) assert [row.subject_id for row in rows] == [str(active.id)] assert rows[0].overall_score is not None assert [row["key"] for row in rows[0].aggregate_facts["data_quality"]["sources"]] == [ "inventory", "efficiency", "cost", "scope_sync", ] @pytest.mark.asyncio async def test_cloud_scope_baseline_bootstrap_disables_itself_after_complete_backfill( db_session, ): integration = Integration( kind="cloud", provider="aws", name="Bootstrap AWS", slug="bootstrap-aws", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() db_session.add_all([ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id="bootstrap-active", display_name="Bootstrap active", status="active", ), ScheduledJob( name=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, display_name="Bootstrap cloud-scope baseline", handler_key=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, trigger_type="interval", interval_seconds=60, timezone="UTC", enabled=True, config={"max_attempts": 6}, ), ]) await db_session.commit() result = await run_cloud_scope_baseline_bootstrap(db_session, {}) assert result["bootstrap_complete"] is True assert result["bootstrap_disabled"] is True job = ( await db_session.execute( select(ScheduledJob).where( ScheduledJob.handler_key == _CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, ) ) ).scalar_one() assert job.enabled is False @pytest.mark.asyncio async def test_cloud_scope_baseline_bootstrap_stops_after_bounded_failed_retries( db_session, monkeypatch, ): job = ScheduledJob( name=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, display_name="Bootstrap cloud-scope baseline", handler_key=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, trigger_type="interval", interval_seconds=60, timezone="UTC", enabled=True, config={"max_attempts": 2}, # ``run_count`` is deliberately left at its 0 default: the # give-up must be driven by dispatched run rows, not by the # terminal-write-back counter. See the regression test below. ) db_session.add(job) await db_session.flush() db_session.add_all([ ScheduledJobRun( job_id=job.id, started_at=datetime.now(UTC) - timedelta(minutes=n), status="failed", triggered_by="scheduler", ) for n in (2, 1) ]) await db_session.commit() async def failed_refresh(_db, **_kw): return { "scope_count": 4, "scopes_processed": 4, "complete": True, "next_scope_id": None, "resumed_after_scope_id": None, "snapshots_created": 2, "snapshots_reused": 0, "failed_scope_count": 2, "failed_scope_ids": ["scope-a", "scope-b"], "model_calls": 0, "run_at": datetime.now(UTC).isoformat(), } monkeypatch.setattr( "app.scheduled_jobs.intelligence_handlers._refresh_cloud_scope_baselines", failed_refresh, ) with pytest.raises(RuntimeError, match="gave up after 2 attempt"): await run_cloud_scope_baseline_bootstrap(db_session, {"max_attempts": 2}) refreshed = await db_session.get(ScheduledJob, job.id) assert refreshed is not None assert refreshed.enabled is False @pytest.mark.asyncio async def test_daily_handler_persists_once_for_unchanged_facts_and_never_calls_model( db_session, ): now = datetime.now(UTC) db_session.add_all([ IdentityPerson( source_system="test", hris_person_id="daily-person", work_email="daily-person@example.test", display_name="Daily Person", employment_status="active", job_title="Engineer", work_location_code="SHA", ), PlatformSetting( key="observability.client_error_telemetry_started", value=True, created_at=now - timedelta(days=8), ), ClientErrorDailyAggregate( day=now.date() - timedelta(days=1), surface="desktop", report_count=1, first_seen_at=now - timedelta(days=1), last_seen_at=now - timedelta(days=1), ), ]) await db_session.commit() first = await run_intelligence_snapshot_daily(db_session, {}) second = await run_intelligence_snapshot_daily(db_session, {}) assert first["created"] is True assert second["created"] is False assert first["snapshot_id"] == second["snapshot_id"] assert first["facts_hash"] == second["facts_hash"] assert first["model_calls"] == second["model_calls"] == 0 snapshot = (await db_session.execute(select(IntelligenceSnapshot))).scalar_one() assert snapshot.subject_type == "platform" assert snapshot.subject_id == "platform" assert snapshot.audience_key == BUSINESS_AUDIENCE_KEY assert [row["key"] for row in snapshot.dimensions] == list(PLATFORM_DIMENSIONS) analysis_count = int((await db_session.execute( select(func.count(IntelligenceAnalysis.id)) )).scalar_one()) assert analysis_count == 0 def _regional_person(index: int, *, status: str = "active") -> IdentityPerson: return IdentityPerson( source_system="test", hris_person_id=f"regional-admin-{index}", work_email=f"regional-admin-{index}@example.test", display_name=f"Regional Admin {index}", employment_status=status, ) @pytest.mark.asyncio async def test_regional_asset_audiences_share_identical_active_location_sets(db_session): locations = [ AssetLocation(code="RGNA", name="Regional A", is_active=True), AssetLocation(code="RGNB", name="Regional B", is_active=True), AssetLocation(code="RGNC", name="Regional C", is_active=True), AssetLocation(code="RGNI", name="Regional Inactive", is_active=False), ] admins = [_regional_person(index) for index in range(1, 6)] inactive_admin = _regional_person(6, status="left") db_session.add_all([*locations, *admins, inactive_admin]) await db_session.flush() # Admins 1 and 2 share the same set; admin 3 owns a disjoint set. for admin in admins[:2]: for location in locations[:2]: db_session.add( AssetLocationAdmin(location_id=location.id, person_id=admin.id), ) db_session.add_all([ AssetLocationAdmin(location_id=locations[2].id, person_id=admins[2].id), AssetLocationAdmin(location_id=locations[0].id, person_id=admins[3].id), AssetLocationAdmin(location_id=locations[1].id, person_id=admins[4].id), AssetLocationAdmin(location_id=locations[3].id, person_id=admins[4].id), AssetLocationAdmin(location_id=locations[0].id, person_id=inactive_admin.id), ]) global_role = ( await db_session.execute(select(AuthRole).where(AuthRole.name == "global_admin")) ).scalar_one() db_session.add(AuthUserRole(user_id=admins[3].id, role_id=global_role.id)) finance_role = ( await db_session.execute(select(AuthRole).where(AuthRole.name == "finance")) ).scalar_one() db_session.add(AuthUserRole(user_id=admins[4].id, role_id=finance_role.id)) await db_session.commit() audiences = await regional_asset_audiences(db_session) location_sets = {row.location_ids for row in audiences} assert location_sets == { frozenset({locations[0].id, locations[1].id}), frozenset({locations[2].id}), } assert len({row.key for row in audiences}) == 2 assert audiences == tuple(sorted(audiences, key=lambda row: row.key)) assert regional_asset_audience({locations[1].id, locations[0].id}).key == ( regional_asset_audience({locations[0].id, locations[1].id}).key ) with pytest.raises(ValueError, match="at least one location"): regional_asset_audience(set()) @pytest.mark.asyncio async def test_daily_handler_builds_one_isolated_snapshot_per_unique_region(db_session): locations = [ AssetLocation(code="SNAPA", name="Snapshot Region A"), AssetLocation(code="SNAPB", name="Snapshot Region B"), ] shared_admins = [_regional_person(index) for index in (10, 11)] other_admin = _regional_person(12) db_session.add_all([*locations, *shared_admins, other_admin]) await db_session.flush() for admin in shared_admins: db_session.add( AssetLocationAdmin(location_id=locations[0].id, person_id=admin.id), ) db_session.add( AssetLocationAdmin(location_id=locations[1].id, person_id=other_admin.id), ) db_session.add_all([ AssetItem( asset_tag="REG-A-1", category="laptop", status="idle", location="Snapshot Region A", ), AssetItem( asset_tag="REG-B-1", category="laptop", status="idle", location="Snapshot Region B", ), ]) await db_session.commit() first = await run_intelligence_snapshot_daily(db_session, {}) second = await run_intelligence_snapshot_daily(db_session, {}) assert first["regional_audience_count"] == 2 assert first["regional_snapshots_created"] == 2 assert second["regional_snapshots_created"] == 0 assert first["model_calls"] == second["model_calls"] == 0 snapshots = list((await db_session.execute(select(IntelligenceSnapshot))).scalars()) regional = [row for row in snapshots if row.subject_type == "domain"] assert len(snapshots) == 3 assert len(regional) == 2 assert all(row.subject_id == "assets" for row in regional) assert all([item["key"] for item in row.dimensions] == ["assets"] for row in regional) assert all( float(row.confidence_score) == pytest.approx(row.dimensions[0]["confidence"]) for row in regional ) assert all(row.aggregate_facts["assets"]["regional_scope"] is True for row in regional) assert {row.aggregate_facts["assets"]["asset_count"] for row in regional} == {1} @pytest.mark.asyncio async def test_cloud_scope_baseline_bootstrap_counts_abandoned_runs_toward_giveup( db_session, monkeypatch, ): """**The 2026-08-17 contract.** Runs killed mid-flight must still count toward the bounded-retry give-up. ``ScheduledJob.run_count`` is bumped only in the scheduler's terminal write-back. A handler that is killed before returning — worker restart, deploy, OOM — never increments it, so keying the give-up on ``run_count`` made this one-shot row's only self-disable path unreachable. In prod that row (60s interval, 271 cloud scopes, longer than the watchdog's 300s grace) was dispatched 874 times with ``run_count`` still 0, crash-looping the worker for ~16h and truncating every long-running cloud-cost ingest mid-flight. Here: ``run_count`` is 0 and every prior run row is an abandoned ``running``/swept-``failed`` — exactly the prod shape. The job must still give up and disable itself. """ job = ScheduledJob( name=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, display_name="Bootstrap cloud-scope baseline", handler_key=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, trigger_type="interval", interval_seconds=60, timezone="UTC", enabled=True, config={"max_attempts": 3}, run_count=0, ) db_session.add(job) await db_session.flush() now = datetime.now(UTC) db_session.add_all([ # Abandoned mid-run, then swept — never reached the write-back. ScheduledJobRun( job_id=job.id, started_at=now - timedelta(minutes=30), status="failed", triggered_by="scheduler", error_message="auto-fail: stale running row swept after 6h", ), ScheduledJobRun( job_id=job.id, started_at=now - timedelta(minutes=20), status="running", triggered_by="scheduler", ), ScheduledJobRun( job_id=job.id, started_at=now - timedelta(minutes=10), status="running", triggered_by="scheduler", ), # Single-flight skips are not attempts and must not be counted — # at a 60s cadence they outnumber real attempts ~10:1 and would # collapse the retry budget to a single pass. ScheduledJobRun( job_id=job.id, started_at=now - timedelta(minutes=5), status="skipped", triggered_by="scheduler", ), ]) await db_session.commit() async def failed_refresh(_db, **_kw): return { "scope_count": 4, "scopes_processed": 4, "complete": True, "next_scope_id": None, "resumed_after_scope_id": None, "snapshots_created": 2, "snapshots_reused": 0, "failed_scope_count": 2, "failed_scope_ids": ["scope-a", "scope-b"], "model_calls": 0, "run_at": datetime.now(UTC).isoformat(), } monkeypatch.setattr( "app.scheduled_jobs.intelligence_handlers._refresh_cloud_scope_baselines", failed_refresh, ) with pytest.raises(RuntimeError, match="gave up after 3 attempt"): await run_cloud_scope_baseline_bootstrap(db_session, {"max_attempts": 3}) refreshed = await db_session.get(ScheduledJob, job.id) assert refreshed is not None assert refreshed.enabled is False, ( "a bootstrap row whose runs are all abandoned must still reach " "its give-up; otherwise it retries forever and, if it outlives " "the watchdog grace, crash-loops the worker" ) @pytest.mark.asyncio async def test_anomaly_index_scans_once_per_integration_with_identical_output( db_session, monkeypatch, ): """**The 2026-08-18 perf contract.** Sibling scopes share one anomaly scan. ``detect_anomalies`` is integration-scoped: its ``resource_keys`` argument only filters in Python *after* the widest GROUP BY in the cost domain. Calling it per subscription re-ran that scan once per sibling — aws-primary has 96 active scopes. Measured against prod: ~51s/scope, ~3.9h for 271 scopes, which is why the fleet refresh had never once completed. Pinned here rather than by an A/B against prod because prod's cost facts are rewritten continuously by the ingest crons — a control pass (old vs old) proved that even two identical runs disagree there, so a live diff cannot distinguish a logic change from data drift. Asserts both halves: the scan collapses to one call per integration, AND every scope still sees exactly the anomalies it saw before. """ from app.intelligence import cloud_scope_facts as facts_mod integration = Integration( kind="cloud", provider="aws", name="Idx AWS", slug="idx-aws", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() scopes = [ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"idx-{n}", display_name=f"Idx {n}", status="active", ) for n in range(3) ] db_session.add_all(scopes) await db_session.flush() # Real cost facts, or this test is vacuous: with no facts, # ``evaluable_keys`` is empty and ``detect_anomalies`` is never called # at all, so a "scanned once" assertion passes without the memo ever # being exercised. Caught by mutation-testing this very test — the # first version stayed green with the memo disabled. # Each resource needs >= 7 days of coverage to become evaluable, and a # late spike so there is a real anomaly to compare across the two paths. anchor = datetime.now(UTC).date() for n, scope in enumerate(scopes): for day_offset in range(2, 14): spike = day_offset == 2 db_session.add(CmdbCostFact( fact_key=uuid.uuid4().hex, date=anchor - timedelta(days=day_offset), integration_id=str(integration.id), scope_id=str(scope.id), service="AmazonEC2", resource_id=f"i-idx-{n}", amount_usd=Decimal("0"), amount_native=Decimal("0"), amount_usd_list=Decimal("9000") if spike else Decimal("10"), currency="USD", )) await db_session.commit() calls: list[dict] = [] real = facts_mod.detect_anomalies async def counting(db, **kw): calls.append(kw) return await real(db, **kw) monkeypatch.setattr(facts_mod, "detect_anomalies", counting) run_at = datetime.now(UTC) per_scope = {} for scope in scopes: per_scope[str(scope.id)] = ( await facts_mod.collect_cloud_scope_facts( db_session, str(scope.id), now=run_at, ) ).aggregate_facts baseline_calls = len(calls) calls.clear() index = facts_mod.ScopeAnomalyIndex() indexed = {} for scope in scopes: indexed[str(scope.id)] = ( await facts_mod.collect_cloud_scope_facts( db_session, str(scope.id), now=run_at, anomaly_index=index, ) ).aggregate_facts assert indexed == per_scope, ( "indexed sweep must produce byte-identical facts; the per-scope key " "filter is post-hoc on (integration_id, resource_id), limit is None, " "and each resource's median/IQR uses only its own series — so " "hoisting the scan cannot change the result" ) assert len(calls) <= 1, ( f"3 sibling scopes must share ONE integration scan, got {len(calls)}; " f"the unindexed path made {baseline_calls}" ) if calls: assert calls[0].get("resource_keys") is None, ( "the shared scan must be unfiltered so every sibling can reuse it" ) @pytest.mark.asyncio async def test_baseline_sweep_is_budgeted_and_resumable(db_session, monkeypatch): """A budgeted sweep stops on a scope boundary and resumes after it. Without this ceiling the sweep's runtime is unbounded in the fleet size, which is how this code path outlived the worker watchdog's 300s grace and crash-looped prod for ~16h on 2026-08-17. """ from app.scheduled_jobs import intelligence_handlers as mod integration = Integration( kind="cloud", provider="aws", name="Budget AWS", slug="budget-aws", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() db_session.add_all([ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"budget-{n}", display_name=f"Budget {n}", status="active", ) for n in range(4) ]) await db_session.commit() # Each scope "costs" 10s of the budget. clock = {"t": 0.0} monkeypatch.setattr(mod.time, "monotonic", lambda: clock["t"]) real_build = mod.build_cloud_scope_snapshot async def slow_build(db, scope_id, **kw): clock["t"] += 10.0 return await real_build(db, scope_id, **kw) monkeypatch.setattr(mod, "build_cloud_scope_snapshot", slow_build) first = await mod._refresh_cloud_scope_baselines(db_session, budget_secs=15) assert first["complete"] is False assert first["scopes_processed"] == 2, first assert first["next_scope_id"] is not None rest = await mod._refresh_cloud_scope_baselines( db_session, budget_secs=10_000, after_scope_id=first["next_scope_id"], ) assert rest["complete"] is True assert rest["resumed_after_scope_id"] == first["next_scope_id"] assert first["scopes_processed"] + rest["scopes_processed"] == 4, ( "resume must cover exactly the remaining scopes — no scope skipped, " "none processed twice" ) @pytest.mark.asyncio async def test_baseline_sweep_always_advances_on_a_tiny_budget( db_session, monkeypatch, ): """A budget smaller than one scope must still make forward progress. Otherwise a resumable sweep re-reads the same cursor forever — the same "never completes" shape, just with a tidier summary. """ from app.scheduled_jobs import intelligence_handlers as mod integration = Integration( kind="cloud", provider="aws", name="Tiny AWS", slug="tiny-aws", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() db_session.add_all([ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"tiny-{n}", display_name=f"Tiny {n}", status="active", ) for n in range(3) ]) await db_session.commit() clock = {"t": 0.0} monkeypatch.setattr(mod.time, "monotonic", lambda: clock["t"]) real_build = mod.build_cloud_scope_snapshot async def slow_build(db, scope_id, **kw): clock["t"] += 999.0 return await real_build(db, scope_id, **kw) monkeypatch.setattr(mod, "build_cloud_scope_snapshot", slow_build) result = await mod._refresh_cloud_scope_baselines(db_session, budget_secs=1) assert result["scopes_processed"] == 1, ( "the first scope must always run regardless of budget; got " f"{result['scopes_processed']}" ) assert result["complete"] is False assert result["next_scope_id"] is not None @pytest.mark.asyncio async def test_bootstrap_keeps_cursor_when_a_scope_fails(db_session, monkeypatch): """A failing scope must not throw away the ground already swept. The failure paths raise, so a cursor persisted *after* them would never land: the next tick would restart from the top and re-walk the same prefix forever — the same "never completes" shape as the original bug, just with a tidier summary. The cursor is therefore committed before any failure verdict. """ from app.scheduled_jobs import intelligence_handlers as mod job = ScheduledJob( name=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, display_name="Bootstrap cloud-scope baseline", handler_key=_CLOUD_SCOPE_BASELINE_BOOTSTRAP_KEY, trigger_type="interval", interval_seconds=60, timezone="UTC", enabled=True, config={"max_attempts": 6}, ) db_session.add(job) await db_session.commit() async def partial_and_failing(_db, **_kw): return { "scope_count": 10, "scopes_processed": 4, "complete": False, "next_scope_id": "scope-cursor-4", "resumed_after_scope_id": None, "snapshots_created": 3, "snapshots_reused": 0, "failed_scope_count": 1, "failed_scope_ids": ["scope-bad"], "model_calls": 0, "run_at": datetime.now(UTC).isoformat(), } monkeypatch.setattr(mod, "_refresh_cloud_scope_baselines", partial_and_failing) with pytest.raises(RuntimeError, match="incomplete"): await run_cloud_scope_baseline_bootstrap(db_session, {}) refreshed = await db_session.get(ScheduledJob, job.id) assert refreshed is not None assert refreshed.enabled is True, "a recoverable failure must not disable the row" assert (refreshed.config or {}).get("resume_after_scope_id") == "scope-cursor-4", ( "the cursor must survive the raise; otherwise every tick re-walks " "the same prefix and the sweep never reaches its completion branch" )