from datetime import UTC, date, datetime, timedelta from decimal import Decimal from uuid import uuid4 from zoneinfo import ZoneInfo import pytest from sqlalchemy import select from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.asset.models import ( AssetInventoryCycle, AssetInventoryItem, AssetItem, AssetLocation, AssetOperation, AssetSoftwareCatalog, ExternalAccountRegistry, SoftwareExternalHolder, SoftwareLicensePlan, SoftwareSeatAllocation, SoftwareSeatSyncStat, ) from app.domains.auditor.models import ( AccessCertification, AccessCertificationItem, AuditorMfaStatus, AuditorPatAudit, AuditorPublicExposureFinding, AuditorSecretRotationRecord, ) from app.domains.cmdb.models import ( EFFICIENCY_TRIAGE_KEPT, CmdbCertificate, CmdbCloudAccessGrant, CmdbCloudScope, CmdbCostFact, CmdbDomain, CmdbServer, CmdbServerEfficiency, CmdbServerEfficiencyTriage, ) from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_READY, RES_CLOUD_SCOPE_OWNER, RESO_PENDING, RESO_TRANSFERRED, OffboardingBlocker, OffboardingProcess, ) from app.domains.offboarding.models import ( SLA_BREACH as OFFBOARDING_BREACH, ) from app.domains.onboarding.models import ( PROC_COMPLETED, PROC_PREPARING, OnboardingBlocker, OnboardingProcess, ) from app.domains.onboarding.models import ( SLA_BREACH as ONBOARDING_BREACH, ) from app.domains.provisioning.models import ( EXEC_AUTO_API, EXEC_MANUAL_NOTIFY, STATUS_COMPLETED, STATUS_FAILED, ProvisioningTask, ) from app.identity.data_quality_models import ( IdentityDataQualityFinding, IdentityDataQualityRun, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson from app.integrations.models import Integration, PlatformSetting from app.intelligence.privacy import EvidenceObject, assert_llm_safe from app.intelligence.providers import ( AssetFactProvider, CloudNetworkFactProvider, CostEfficiencyFactProvider, LifecycleFactProvider, PeopleOrgFactProvider, PlatformOperationsFactProvider, SecurityComplianceFactProvider, SoftwareApiFactProvider, ) from app.intelligence.snapshot_service import SnapshotSubject, build_snapshot from app.notifications.models import NotificationOutbox from app.observability.client_errors import ClientErrorDailyAggregate from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun NOW = datetime(2026, 8, 15, 4, tzinfo=UTC) def person(index: int, **values) -> IdentityPerson: defaults = { "source_system": "test", "hris_person_id": f"person-{index}", "work_email": f"person-{index}@example.test", "display_name": f"Person {index}", "employment_status": "active", "job_title": "Engineer", "work_location_code": "SHA", } defaults.update(values) return IdentityPerson(**defaults) @pytest.mark.asyncio async def test_people_provider_reuses_identity_quality_semantics_and_stays_private(db_session): apex = person(1) db_session.add(apex) await db_session.flush() covered = person(2, manager_person_id=apex.id) gap = person(3, work_location_code=None, work_location_name=None) db_session.add_all([covered, gap]) await db_session.flush() db_session.add_all( [ IdentityOrgMembership( person_id=row.id, source_system="test", org_node_source_id=f"org-{index}", org_node_type="department", org_node_name="private org label", is_primary=True, membership_status="active", ) for index, row in enumerate((apex, covered), start=1) ] ) db_session.add( ScheduledJob( name="hris.incremental_sync", display_name="HRIS", handler_key="hris.incremental_sync", trigger_type="interval", interval_seconds=14_400, enabled=True, last_run_at=NOW - timedelta(hours=2), # Due instant is derived as last_run + interval = NOW+2h; # scheduled_jobs stores no next-run time (migration 620). last_status="ok", ) ) await db_session.commit() result = await PeopleOrgFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) metrics = {row.key: row for row in result.metrics} assert metrics["people_org.hris_freshness"].value == 0 assert metrics["people_org.org_coverage"].value == Decimal("66.67") # The org apex has a direct report and is intentionally excluded. assert metrics["people_org.manager_coverage"].value == 50 assert metrics["people_org.prediction_data_coverage"].value == Decimal("66.67") assert result.aggregate_facts["manager_eligible_count"] == 2 assert len(metrics["people_org.manager_coverage"].evidence) == 1 assert_llm_safe(result.aggregate_facts) assert "private org label" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_people_provider_marks_absent_sources_as_missing_not_bad_health(db_session): result = await build_snapshot( db_session, subject=SnapshotSubject("platform", "platform", "shared"), providers=[PeopleOrgFactProvider(now=lambda: NOW)], ) people = next(row for row in result.snapshot.dimensions if row["key"] == "people_org") assert people["score"] is None assert people["confidence"] == 0 assert result.snapshot.aggregate_facts["people_org"]["hris_freshness_state"] == "job_missing" @pytest.mark.asyncio async def test_hris_freshness_counts_consecutive_missed_cycles(db_session): db_session.add( ScheduledJob( name="hris.incremental_sync", display_name="HRIS", handler_key="hris.incremental_sync", trigger_type="interval", interval_seconds=3_600, enabled=True, last_run_at=NOW - timedelta(hours=4), # Derived due = last_run + 1h interval = NOW-3h, i.e. 3 full # intervals overdue -> 4 missed cycles. last_status="ok", ) ) await db_session.commit() result = await PeopleOrgFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "people_org", "hr") ) freshness = next(row for row in result.metrics if row.key == "people_org.hris_freshness") assert freshness.value == 4 assert result.aggregate_facts["hris_freshness_state"] == "stale" @pytest.mark.asyncio async def test_lifecycle_provider_uses_existing_joiner_mover_leaver_records(db_session): people = [person(index) for index in range(10, 17)] db_session.add_all(people) await db_session.flush() hire_date = date(2026, 8, 1) onboarding_ok = OnboardingProcess( person_id=people[0].id, hire_date=hire_date, status=PROC_COMPLETED, hard_ready_at=datetime(2026, 7, 31, 12, tzinfo=UTC), ) onboarding_gap = OnboardingProcess( person_id=people[1].id, hire_date=hire_date, status=PROC_PREPARING, sla_breach_level=ONBOARDING_BREACH, ) offboarding_gap = OffboardingProcess( person_id=people[2].id, leaver_date=hire_date, status=PROC_BLOCKED, sla_breach_level=OFFBOARDING_BREACH, ) offboarding_ok = OffboardingProcess( person_id=people[3].id, leaver_date=hire_date, status=PROC_READY, ) future_offboarding = OffboardingProcess( person_id=people[6].id, leaver_date=date(2026, 9, 1), status=PROC_BLOCKED, ) db_session.add_all( [onboarding_ok, onboarding_gap, offboarding_gap, offboarding_ok, future_offboarding] ) await db_session.flush() db_session.add_all( [ OffboardingBlocker( process_id=offboarding_gap.id, resource_type=RES_CLOUD_SCOPE_OWNER, resource_id=uuid4(), label="private scope label", resolution=RESO_PENDING, error_count=1, ), OffboardingBlocker( process_id=offboarding_ok.id, resource_type=RES_CLOUD_SCOPE_OWNER, resource_id=uuid4(), label="another private label", resolution=RESO_TRANSFERRED, ), OffboardingBlocker( process_id=future_offboarding.id, resource_type=RES_CLOUD_SCOPE_OWNER, resource_id=uuid4(), label="future private label", resolution=RESO_PENDING, ), ProvisioningTask( user_id=people[4].id, action="revoke", target_type="software", source="role_change", execution_mode="auto", status=STATUS_COMPLETED, created_at=NOW, ), ProvisioningTask( user_id=people[5].id, action="revoke", target_type="software", source="role_change", execution_mode="auto", status=STATUS_FAILED, created_at=NOW, ), ] ) await db_session.commit() result = await LifecycleFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "lifecycle", "hr") ) metrics = {row.key: row for row in result.metrics} assert metrics["lifecycle.day1_ready"].value == 50 assert metrics["lifecycle.onboarding_sla"].value == 50 assert metrics["lifecycle.leaver_failure_rate"].value == 50 # health-v2 retired ``lifecycle.transfer_recovery``. The two # ``source='role_change'`` revoke tasks above are deliberately left in the # fixture: the metric must stay absent even when its old signal source has # rows, so a future re-add cannot slip back in unversioned. assert "lifecycle.transfer_recovery" not in metrics assert metrics["lifecycle.owner_continuity"].value == 50 assert result.aggregate_facts["ownership_resolved_count"] == 1 assert result.aggregate_facts["ownership_total_count"] == 2 assert "leaver_risk" in result.aggregate_facts # Risk facts intentionally cover every non-cancelled offboarding process; # this includes the future-dated fixture, unlike the six-month KPI window. assert result.aggregate_facts["leaver_risk"]["summary"]["leaver_count"] == 3 assert len(metrics["lifecycle.owner_continuity"].evidence) == 1 assert_llm_safe(result.aggregate_facts) assert "private scope label" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_leaver_failure_rate_counts_unresolved_failures_not_retry_history( db_session, ): """health-v2 口径 for ``lifecycle.leaver_failure_rate`` (2026-08-18). The v1 predicate ("any blocker with ``error_count > 0``, or process status ``blocked``") reported 12 of 78 production leavers as failed while zero had an unresolved failure: ``error_count`` is a monotonic since-creation counter that never resets on resolve, and ``blocked`` is the ordinary in-flight status of every offboarding before the T+0+24h force-resolve. Each fixture below isolates one arm so a regression cannot pass by accident — the two "clean" leavers are exactly the shapes v1 got wrong. """ people = [person(index) for index in range(60, 65)] db_session.add_all(people) await db_session.flush() old_leaver_date = date(2026, 8, 1) # NOW is 2026-08-15 → past the 24h grace fresh_leaver_date = NOW.astimezone(ZoneInfo("Asia/Shanghai")).date() # (1) CLEAN: retried but every blocker resolved. v1 counted this forever. retried_then_resolved = OffboardingProcess( person_id=people[0].id, leaver_date=old_leaver_date, status=PROC_READY, ) # (2) CLEAN: in-flight today. ``blocked`` == "has a pending row", not failed. in_flight = OffboardingProcess( person_id=people[1].id, leaver_date=fresh_leaver_date, status=PROC_BLOCKED, ) # (3) FAILED: pending blocker whose auto-handler actually errored. pending_errored = OffboardingProcess( person_id=people[2].id, leaver_date=old_leaver_date, status=PROC_BLOCKED, ) # (4) FAILED: pending blocker whose linked provisioning task failed. pending_task_failed = OffboardingProcess( person_id=people[3].id, leaver_date=old_leaver_date, status=PROC_BLOCKED, ) # (5) FAILED: no error recorded, but stuck well past the force-resolve grace. stuck_past_deadline = OffboardingProcess( person_id=people[4].id, leaver_date=old_leaver_date, status=PROC_BLOCKED, ) db_session.add_all([ retried_then_resolved, in_flight, pending_errored, pending_task_failed, stuck_past_deadline, ]) await db_session.flush() failed_task = ProvisioningTask( user_id=people[3].id, action="revoke", target_type="software", source="offboarding", execution_mode=EXEC_AUTO_API, status=STATUS_FAILED, created_at=NOW, ) db_session.add(failed_task) await db_session.flush() def blocker(process, **values): defaults = { "process_id": process.id, "resource_type": RES_CLOUD_SCOPE_OWNER, "resource_id": uuid4(), "label": "private blocker label", "resolution": RESO_PENDING, } defaults.update(values) return OffboardingBlocker(**defaults) db_session.add_all([ # 58 historical failures, but the blocker landed transferred. blocker(retried_then_resolved, resolution=RESO_TRANSFERRED, error_count=58), blocker(in_flight, error_count=0), blocker(pending_errored, error_count=3), blocker(pending_task_failed, error_count=0, provisioning_task_id=failed_task.id), blocker(stuck_past_deadline, error_count=0), ]) await db_session.commit() result = await LifecycleFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "lifecycle", "hr") ) metrics = {row.key: row for row in result.metrics} # 3 failed of 5 window leavers. Under v1 this fixture would read 100%: # every process is either ``blocked`` or carries a retry counter. assert metrics["lifecycle.leaver_failure_rate"].value == 60 assert result.aggregate_facts["leaver_total_count"] == 5 assert result.aggregate_facts["leaver_failed_count"] == 3 evidence_ids = {row.object_id for row in metrics["lifecycle.leaver_failure_rate"].evidence} assert evidence_ids == { str(pending_errored.id), str(pending_task_failed.id), str(stuck_past_deadline.id), } assert str(retried_then_resolved.id) not in evidence_ids assert str(in_flight.id) not in evidence_ids assert_llm_safe(result.aggregate_facts) @pytest.mark.asyncio async def test_fact_providers_reject_cross_domain_subjects(db_session): with pytest.raises(ValueError, match="People facts"): await PeopleOrgFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("cloud_scope", "scope-1", "shared") ) with pytest.raises(ValueError, match="Lifecycle facts"): await LifecycleFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "assets", "asset_admin") ) with pytest.raises(ValueError, match="Asset facts"): await AssetFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "security_compliance", "auditor") ) with pytest.raises(ValueError, match="Cloud/network facts"): await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("cloud_scope", "scope-1", "shared") ) with pytest.raises(ValueError, match="Cost facts"): await CostEfficiencyFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "assets", "finance") ) with pytest.raises(ValueError, match="Security/compliance facts"): await SecurityComplianceFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "platform_ops", "auditor") ) @pytest.mark.asyncio async def test_asset_provider_scores_real_records_and_enforces_region_scope(db_session): shanghai = AssetLocation(code="RGNSH", name="上海") singapore = AssetLocation(code="RGNSG", name="新加坡") holder = person(30, work_location_name="上海市", city="上海") hires = [person(index, work_location_name="上海市", city="上海") for index in range(31, 34)] db_session.add_all([shanghai, singapore, holder, *hires]) await db_session.flush() complete = AssetItem( asset_tag="SH-1", category="laptop", status="assigned", assigned_to=holder.id, purchase_date=date(2025, 1, 1), purchase_cost=1000, currency="CNY", purchase_entity="CN", warranty_until=date(2028, 1, 1), updated_at=NOW, ) old_idle = AssetItem( asset_tag="SH-2", category="laptop", status="idle", location="上海市", purchase_date=date(2024, 1, 1), updated_at=NOW, created_at=NOW - timedelta(days=120), ) leaver_held = AssetItem( asset_tag="SH-3", category="laptop", status="held_by_leaver", location="上海", updated_at=NOW, ) reserved = AssetItem( asset_tag="SH-4", category="laptop", status="idle", location="上海", reserved_for_person_id=hires[0].id, updated_at=NOW, ) # The asset's own managed location wins over its assignee's: this # laptop is held by a Shanghai person but sits in Singapore, and # must stay out of the Shanghai administrator's regional result. foreign = AssetItem( asset_tag="SG-1", category="laptop", status="assigned", assigned_to=holder.id, location="新加坡", purchase_date=date(2025, 1, 1), purchase_cost=1000, currency="SGD", purchase_entity="SG", warranty_until=date(2028, 1, 1), updated_at=NOW, ) db_session.add_all([complete, old_idle, leaver_held, reserved, foreign]) await db_session.flush() for hire in hires: process = OnboardingProcess( person_id=hire.id, hire_date=date(2026, 8, 20), status=PROC_PREPARING, ) db_session.add(process) await db_session.flush() db_session.add( OnboardingBlocker( process_id=process.id, resource_type="hardware_allocation", label="private demand label", resolution="pending", ) ) cycle = AssetInventoryCycle( name="private cycle name", status="active", started_at=NOW, total_items=5, pending_count=1, confirmed_count=3, issue_count=1, ) db_session.add(cycle) await db_session.flush() for asset, status in zip( (complete, old_idle, leaver_held, reserved, foreign), ("confirmed", "confirmed", "pending", "issue_reported", "confirmed"), strict=True, ): db_session.add( AssetInventoryItem( cycle_id=cycle.id, asset_id=asset.id, holder_kind="it_fallback", status=status, ) ) await db_session.commit() result = await AssetFactProvider( managed_location_ids=frozenset({shanghai.id}), now=lambda: NOW, ).collect(db_session, SnapshotSubject("domain", "assets", f"region:{shanghai.id}")) metrics = {row.key: row for row in result.metrics} assert result.aggregate_facts["asset_count"] == 4 assert result.aggregate_facts["regional_scope"] is True assert metrics["assets.demand_coverage"].value == Decimal("66.67") # 2 of the 3 answered items match; the 4th is unanswered, so it moves the # response rate (confidence), not the accuracy. assert metrics["assets.inventory_accuracy"].value == Decimal("66.67") assert metrics["assets.inventory_accuracy"].confidence == 75 # One long-idle laptop, but two uncovered hires still need hardware — # that device is capacity, not surplus. assert metrics["assets.idle_rate"].value == 0 assert metrics["assets.leaver_hold_rate"].value == 25 assert metrics["assets.financial_coverage"].value == 25 # Retired in health-v3 while the fixtures still carry ``warranty_until``: # a scored warranty metric must not come back without a producer. assert "assets.warranty_coverage" not in metrics assert result.aggregate_facts["warranty_complete_count"] == 1 optimization = result.aggregate_facts["optimization"] assert optimization["optimization_version"] == "asset-optimization-v1" assert optimization["idle_surplus_count"] <= optimization["idle_candidate_count"] assert optimization["future_demand_gap_count"] >= 0 assert optimization["inventory_unconfirmed_count"] >= result.aggregate_facts["inventory_issue_count"] assert len(metrics["assets.demand_coverage"].evidence) == 1 assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_inventory_accuracy_scores_answered_items_not_unanswered_ones(db_session): """health-v3 口径 for ``assets.inventory_accuracy`` (2026-08-18). The v2 predicate was ``confirmed / all snapshot items``, so an item nobody had answered yet scored exactly like a record found to be wrong. A cycle therefore opened at its worst score, and because closing a cycle freezes the remainder at ``pending`` the metric stayed pinned there forever. Unanswered items are missing evidence — they belong in confidence, not in the numerator. """ assets = [ AssetItem(asset_tag=f"INV-{index}", category="laptop", status="claimed", updated_at=NOW) for index in range(10) ] db_session.add_all(assets) await db_session.flush() cycle = AssetInventoryCycle(name="cycle", status="active", started_at=NOW) db_session.add(cycle) await db_session.flush() # 8 answered (7 match, 1 mismatch) + 2 nobody has responded to yet. statuses = ["confirmed"] * 7 + ["issue_reported"] + ["pending"] * 2 for asset, status in zip(assets, statuses, strict=True): db_session.add( AssetInventoryItem( cycle_id=cycle.id, asset_id=asset.id, holder_kind="it_fallback", status=status, ) ) await db_session.commit() result = await AssetFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "assets", "global") ) accuracy = {row.key: row for row in result.metrics}["assets.inventory_accuracy"] assert accuracy.value == Decimal("87.50") # 7 of 8 answered, not 7 of 10 assert accuracy.confidence == 80 # 8 of 10 answered assert result.aggregate_facts["inventory_responded_count"] == 8 # Both the mismatch and the two unanswered items stay actionable. assert len(accuracy.evidence) == 3 @pytest.mark.asyncio async def test_idle_rate_dates_idleness_from_custody_history_not_row_edits(db_session): """health-v3 口径 for ``assets.idle_rate`` (2026-08-18). Two v2 defects, both provable on production data: idleness was dated from ``asset_items.updated_at`` — which *answering a stocktake* bumps, so doing inventory reset the idle clock on 70 idle computers in one day without a device moving — and every category counted, so monitors and office network gear were scored as reclaimable capital. """ stocktaken = AssetItem( asset_tag="IDLE-1", category="laptop", status="idle", updated_at=NOW, created_at=NOW - timedelta(days=300), last_inventory_at=NOW, ) returned_long_ago = AssetItem( asset_tag="IDLE-2", category="laptop", status="idle", updated_at=NOW, created_at=NOW - timedelta(days=300), ) just_returned = AssetItem( asset_tag="IDLE-3", category="laptop", status="idle", updated_at=NOW - timedelta(days=300), created_at=NOW - timedelta(days=300), ) shelf_monitor = AssetItem( asset_tag="IDLE-4", category="monitor", status="idle", updated_at=NOW, created_at=NOW - timedelta(days=300), ) office_switch = AssetItem( asset_tag="IDLE-5", category="network_device", status="held_by_office", updated_at=NOW, created_at=NOW - timedelta(days=300), ) in_use = AssetItem( asset_tag="IDLE-6", category="laptop", status="claimed", updated_at=NOW, created_at=NOW - timedelta(days=300), ) operator = person(70) db_session.add_all( [operator, stocktaken, returned_long_ago, just_returned, shelf_monitor, office_switch, in_use] ) await db_session.flush() db_session.add_all([ AssetOperation( operation_number="OP-OLD", asset_id=returned_long_ago.id, operation_type="return", operator_id=operator.id, created_at=NOW - timedelta(days=150), ), AssetOperation( operation_number="OP-NEW", asset_id=just_returned.id, operation_type="return", operator_id=operator.id, created_at=NOW - timedelta(days=3), ), ]) await db_session.commit() result = await AssetFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "assets", "global") ) facts = result.aggregate_facts # IDLE-1 (row touched by its own stocktake response) and IDLE-2 (last # real movement 150 days ago) are long idle; IDLE-3 came back 3 days ago. assert facts["optimization"]["idle_candidate_count"] == 2 assert facts["reclaimable_count"] == 4 idle = {row.key: row for row in result.metrics}["assets.idle_rate"] assert idle.value == Decimal("50.00") assert {ref.object_id for ref in idle.evidence} == { str(stocktaken.id), str(returned_long_ago.id), } # The monitor on the shelf and the office switch are long idle by the # clock, but neither is capital anyone can reclaim and re-issue. assert facts["long_idle_count"] == 4 assert facts["optimization"]["idle_surplus_count"] == 2 @pytest.mark.asyncio async def test_asset_provider_empty_regional_scope_cannot_fall_back_to_global(db_session): db_session.add( AssetItem(asset_tag="GLOBAL-1", category="laptop", status="idle", updated_at=NOW) ) await db_session.commit() result = await AssetFactProvider( managed_location_ids=frozenset(), now=lambda: NOW, ).collect(db_session, SnapshotSubject("domain", "assets", "region:none")) assert result.aggregate_facts["asset_count"] == 0 assert all(row.confidence == 0 for row in result.metrics) assert result.data_as_of is None @pytest.mark.asyncio async def test_asset_provider_excludes_soft_deleted_identity_from_global_demand(db_session): deleted_person = person(39, deleted_at=NOW - timedelta(days=1)) db_session.add(deleted_person) await db_session.flush() process = OnboardingProcess( person_id=deleted_person.id, hire_date=date(2026, 8, 20), status=PROC_PREPARING, ) db_session.add(process) await db_session.flush() db_session.add( OnboardingBlocker( process_id=process.id, resource_type="hardware_allocation", label="private orphan demand", resolution="pending", ) ) await db_session.commit() result = await AssetFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "asset_global") ) assert result.aggregate_facts["demand_count"] == 0 @pytest.mark.asyncio async def test_cloud_network_provider_reuses_governance_and_efficiency_ledgers(db_session): owner = person(60) departed = person(61, employment_status="left") integration = Integration( id=uuid4(), kind="cloud", provider="aws", name="private cloud", slug=f"intel-cloud-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([owner, departed, integration]) await db_session.flush() scopes = [ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"private-account-{index}", status="active", owner_person_id=person_id, last_synced_at=synced_at, ) for index, (person_id, synced_at) in enumerate( ((owner.id, NOW - timedelta(days=3)), (departed.id, NOW)), start=1, ) ] servers = [ CmdbServer( integration_id=integration.id, instance_id=f"private-instance-{index}", status="running", owner_person_id=person_id, last_synced_at=NOW, ) for index, person_id in enumerate((owner.id, departed.id), start=1) ] domains = [ CmdbDomain( integration_id=integration.id, domain_name=f"private-{index}.example.test", status="active", owner_person_id=person_id, expiry_date=expiry, last_synced_at=NOW, ) for index, (person_id, expiry) in enumerate( ((owner.id, date(2027, 1, 1)), (None, date(2026, 8, 20))), start=1, ) ] db_session.add_all([*scopes, *servers, *domains]) await db_session.flush() db_session.add_all( [ CmdbServerEfficiency( server_id=servers[0].id, integration_id=integration.id, verdict="idle", recommendation="stop", computed_at=NOW, ), CmdbServerEfficiency( server_id=servers[1].id, integration_id=integration.id, verdict="healthy", recommendation="keep", computed_at=NOW, ), CmdbCloudAccessGrant( scope_id=scopes[0].id, person_id=owner.id, role_key="private-role", status="active", expires_at=NOW + timedelta(days=90), ), CmdbCloudAccessGrant( scope_id=scopes[1].id, person_id=departed.id, role_key="private-role", status="active", expires_at=None, ), CmdbCertificate( dedup_key=f"manual:{uuid4()}", source="manual", status="valid", not_after=NOW + timedelta(days=90), last_probed_at=NOW, ), CmdbCertificate( dedup_key=f"manual:{uuid4()}", source="manual", status="expired", not_after=NOW - timedelta(days=1), last_probed_at=NOW, ), ] ) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} assert metrics["cloud_network.sync_freshness"].value == 3 assert metrics["cloud_network.resource_waste_rate"].value == 50 assert metrics["cloud_network.resource_waste_rate"].confidence == 100 assert metrics["cloud_network.resource_waste_rate"].evidence == ( EvidenceObject("server", str(servers[0].id)), ) assert metrics["cloud_network.access_governance"].value == 50 assert metrics["cloud_network.owner_coverage"].value == 50 assert metrics["cloud_network.domain_certificate_health"].value == 50 assert result.aggregate_facts["actionable_efficiency_count"] == 1 assert {row.kind for row in metrics["cloud_network.owner_coverage"].evidence} == { "scope", "server", "domain", } assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_cloud_network_provider_has_no_confident_facts_without_sources(db_session): result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) assert all(row.confidence == 0 for row in result.metrics) assert result.data_as_of is None @pytest.mark.asyncio async def test_cloud_network_access_governance_follows_shipped_grant_policy(db_session): """The governance rate scores Helix's own grant policy, not a bare "expires_at is in the future" test. Prod scopes are outside Helix's remit (both expiry sweeps skip them), perpetual holders carry ``expires_at IS NULL`` on purpose, and a scope owner's grant on their own scope never expires. Measuring those three as failures put the 95% coverage band out of reach: on production 2026-08-18 the metric read 62.6% while only 9 of 740 "gaps" were closable. """ from app.auth.models import AuthRole, AuthUserRole cloud_admin = person(80) scope_owner = person(81) plain = person(82) leaver = person(83, employment_status="left") integration = Integration( id=uuid4(), kind="cloud", provider="aws", name="private policy cloud", slug=f"intel-policy-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([cloud_admin, scope_owner, plain, leaver, integration]) await db_session.flush() cloud_admin_role_id = ( await db_session.execute( select(AuthRole.id).where(AuthRole.name == "cloud_admin") ) ).scalar_one() db_session.add( AuthUserRole(user_id=cloud_admin.id, role_id=cloud_admin_role_id) ) managed_scope = CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"private-managed-{uuid4().hex[:8]}", status="active", environment="dev", owner_person_id=scope_owner.id, last_synced_at=NOW, last_seen_sync_job_id=str(uuid4()), ) prod_scope = CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"private-prod-{uuid4().hex[:8]}", status="active", environment="prod", owner_person_id=scope_owner.id, last_synced_at=NOW, last_seen_sync_job_id=str(uuid4()), ) db_session.add_all([managed_scope, prod_scope]) await db_session.flush() leaver_grant = CmdbCloudAccessGrant( scope_id=managed_scope.id, person_id=leaver.id, role_key="private-leaver", status="active", expires_at=NOW + timedelta(days=30), ) db_session.add_all( [ leaver_grant, # Perpetual holder: NULL expiry is the policy, not a gap. CmdbCloudAccessGrant( scope_id=managed_scope.id, person_id=cloud_admin.id, role_key="private-perpetual", status="active", expires_at=None, ), # Scope owner on their own scope: exempt from both expiry sweeps, # so a lapsed timestamp is not a governance failure. CmdbCloudAccessGrant( scope_id=managed_scope.id, person_id=scope_owner.id, role_key="private-owner", status="active", expires_at=NOW - timedelta(days=30), ), # Ordinary employee inside the TTL: governed. CmdbCloudAccessGrant( scope_id=managed_scope.id, person_id=plain.id, role_key="private-ttl", status="active", expires_at=NOW + timedelta(days=30), ), # Prod grant, overdue: Helix does not manage prod access at all, # so it leaves the scored population and travels as a fact. CmdbCloudAccessGrant( scope_id=prod_scope.id, person_id=plain.id, role_key="private-prod", status="active", expires_at=NOW - timedelta(days=5), ), ] ) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} governance = metrics["cloud_network.access_governance"] # 3 governed of 4 managed grants; the prod grant is out of population. assert governance.value == Decimal("75.00") # The only gap is the one a human can close: an active grant whose # holder has left. assert governance.evidence == ( EvidenceObject("grant", str(leaver_grant.id)), ) facts = result.aggregate_facts assert facts["active_grant_count"] == 5 assert facts["managed_grant_count"] == 4 assert facts["governed_grant_count"] == 3 assert facts["perpetual_grant_count"] == 1 assert facts["owner_grant_count"] == 1 assert facts["prod_grant_count"] == 1 assert facts["prod_overdue_grant_count"] == 1 assert_llm_safe(facts) @pytest.mark.asyncio async def test_cloud_network_owner_coverage_counts_inherited_subscription_owner(db_session): """Owner coverage counts the owner the product actually shows. A server with no stored ``owner_person_id`` inherits its subscription's owner, live, via ``resolve_inherited_server_owners`` — the server list, the efficiency page, the owner digest and savings attribution all render that. Reading only the stored column reported a gap the product does not have: production 2026-08-18 had 13 of 6503 servers with a stored owner while 6486 resolved to an active subscription owner. The provider stamp is the link (``_aws_account_id`` here); a server whose driver stamps nothing and which has no billing line stays a real gap. """ owner = person(85) integration = Integration( id=uuid4(), kind="cloud", provider="aws", name="private inherit cloud", slug=f"intel-inherit-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([owner, integration]) await db_session.flush() account_id = f"9{uuid4().int % 10**11:011d}" scope = CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=account_id, status="active", owner_person_id=owner.id, last_synced_at=NOW, last_seen_sync_job_id=str(uuid4()), ) db_session.add(scope) await db_session.flush() inheriting = CmdbServer( integration_id=integration.id, instance_id=f"i-private-{uuid4().hex[:12]}", status="running", owner_person_id=None, last_synced_at=NOW, raw_data={"_aws_account_id": account_id}, ) unattributable = CmdbServer( integration_id=integration.id, instance_id=f"i-private-{uuid4().hex[:12]}", status="running", owner_person_id=None, last_synced_at=NOW, raw_data={"InstanceId": "no account stamp at all"}, ) db_session.add_all([inheriting, unattributable]) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} coverage = metrics["cloud_network.owner_coverage"] # scope (owned) + inheriting server (covered) + unattributable (gap) = 2/3. assert coverage.value == Decimal("66.67") assert coverage.evidence == (EvidenceObject("server", str(unattributable.id)),) facts = result.aggregate_facts assert facts["owner_object_count"] == 3 assert facts["owner_covered_count"] == 2 assert facts["owner_coverage_basis"] == ( "effective_owner_direct_or_cloud_scope_inherited" ) assert facts["owner_gap_count"] == 1 assert facts["directly_owned_server_count"] == 0 assert facts["inherited_owner_server_count"] == 1 @pytest.mark.asyncio async def test_cloud_network_sync_freshness_separates_never_discovered_scope(db_session): """A Helix-provisioned scope that discovery has never adopted is not a broken connector. A root scope no connector has ever confirmed keeps the strict reading — nothing legitimate refreshes it, so it really is stale inventory. What must not happen is sending the reader to a connector console for it, so the never-discovered count and evidence order are pinned here. """ integration = Integration( id=uuid4(), kind="cloud", provider="azure", name="private freshness cloud", slug=f"intel-fresh-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() never_discovered = CmdbCloudScope( integration_id=integration.id, scope_type="azure_subscription", external_id=f"private-provisioned-{uuid4().hex[:8]}", status="active", last_synced_at=NOW - timedelta(days=20), last_seen_sync_job_id=None, ) connector_stale = CmdbCloudScope( integration_id=integration.id, scope_type="azure_subscription", external_id=f"private-stalled-{uuid4().hex[:8]}", status="active", last_synced_at=NOW - timedelta(days=4), last_seen_sync_job_id=str(uuid4()), ) fresh = CmdbCloudScope( integration_id=integration.id, scope_type="azure_subscription", external_id=f"private-fresh-{uuid4().hex[:8]}", status="active", last_synced_at=NOW, last_seen_sync_job_id=str(uuid4()), ) db_session.add_all([never_discovered, connector_stale, fresh]) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} freshness = metrics["cloud_network.sync_freshness"] assert freshness.value == 20 # Never-discovered rows lead the evidence so truncation cannot bury the # condition the reader must act on first. assert freshness.evidence[0] == EvidenceObject("scope", str(never_discovered.id)) assert set(freshness.evidence) == { EvidenceObject("scope", str(never_discovered.id)), EvidenceObject("scope", str(connector_stale.id)), } facts = result.aggregate_facts assert facts["stale_scope_count"] == 2 assert facts["never_discovered_root_scope_count"] == 1 assert facts["child_scope_inheriting_freshness_count"] == 0 @pytest.mark.asyncio async def test_cloud_network_sync_freshness_inherits_parent_for_child_scope(db_session): """A child scope inherits its parent's freshness. Helix provisions ``azure_resource_group`` scopes inside an already-tracked subscription (one per Azure OpenAI key request). The discovery sync walks SUBSCRIPTIONS, so it never enumerates a resource group and the child's ``last_synced_at`` is frozen at its creation stamp forever. Production 2026-08-18: all six such rows had ``last_seen_sync_job_id IS NULL``, and the one still active — key valid to 2026-12-31, resource group live, parent subscription synced hours earlier — reported 20 missed cycles and dragged the whole dimension to the worst band. """ integration = Integration( id=uuid4(), kind="cloud", provider="azure", name="private child cloud", slug=f"intel-child-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() parent = CmdbCloudScope( integration_id=integration.id, scope_type="azure_subscription", external_id=f"private-parent-{uuid4().hex[:8]}", status="active", last_synced_at=NOW, last_seen_sync_job_id=str(uuid4()), ) db_session.add(parent) await db_session.flush() child = CmdbCloudScope( integration_id=integration.id, scope_type="azure_resource_group", external_id=f"private-child-rg-{uuid4().hex[:8]}", status="active", last_synced_at=NOW - timedelta(days=20), last_seen_sync_job_id=None, parent_scope_id=str(parent.id), ) db_session.add(child) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} freshness = metrics["cloud_network.sync_freshness"] assert freshness.value == 0 assert freshness.evidence == () facts = result.aggregate_facts assert facts["stale_scope_count"] == 0 assert facts["child_scope_inheriting_freshness_count"] == 1 assert facts["never_discovered_root_scope_count"] == 0 @pytest.mark.asyncio async def test_cloud_network_waste_rate_drops_owner_triaged_recommendations(db_session): """Owner triage has to reach the score. The efficiency workspace counts 待处理 through ``_effective_triage``; if this metric ignores the triage table, an owner can justify every flagged box and the rate never moves — the loop the triage table exists to close would be invisible to the health score. A stale decision (the verdict drifted since it was made) still counts as open. """ owner = person(84) integration = Integration( id=uuid4(), kind="cloud", provider="aws", name="private triage cloud", slug=f"intel-triage-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([owner, integration]) await db_session.flush() servers = [ CmdbServer( integration_id=integration.id, instance_id=f"private-triage-{index}", status="running", owner_person_id=owner.id, last_synced_at=NOW, ) for index in range(4) ] db_session.add_all(servers) await db_session.flush() db_session.add_all( [ # Open, priced — the one row that should still count. CmdbServerEfficiency( server_id=servers[0].id, integration_id=integration.id, verdict="oversized", recommendation="downsize", computed_at=NOW, monthly_cost_usd=Decimal("400.00"), estimated_monthly_savings_usd=Decimal("120.00"), ), # Owner decided "keep" against the current verdict — suppressed. CmdbServerEfficiency( server_id=servers[1].id, integration_id=integration.id, verdict="oversized", recommendation="downsize", computed_at=NOW, ), # Triaged against a verdict that has since drifted — still open. CmdbServerEfficiency( server_id=servers[2].id, integration_id=integration.id, verdict="idle", recommendation="stop", computed_at=NOW, ), CmdbServerEfficiency( server_id=servers[3].id, integration_id=integration.id, verdict="healthy", recommendation="keep", computed_at=NOW, ), ] ) db_session.add_all( [ CmdbServerEfficiencyTriage( server_id=servers[1].id, status=EFFICIENCY_TRIAGE_KEPT, reason="pinned by a batch job", verdict_at_decision="oversized", decided_by_person_id=owner.id, decided_at=NOW, ), CmdbServerEfficiencyTriage( server_id=servers[2].id, status=EFFICIENCY_TRIAGE_KEPT, reason="was healthy when reviewed", verdict_at_decision="healthy", decided_by_person_id=owner.id, decided_at=NOW, ), ] ) await db_session.commit() result = await CloudNetworkFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cloud_network", "cloud_admin") ) metrics = {row.key: row for row in result.metrics} waste = metrics["cloud_network.resource_waste_rate"] # 2 open of 4 classified: the "keep" against the live verdict is out. assert waste.value == Decimal("50.00") assert set(waste.evidence) == { EvidenceObject("server", str(servers[0].id)), EvidenceObject("server", str(servers[2].id)), } facts = result.aggregate_facts assert facts["flagged_efficiency_count"] == 3 assert facts["actionable_efficiency_count"] == 2 assert facts["triaged_efficiency_count"] == 1 assert facts["priced_actionable_efficiency_count"] == 1 assert facts["actionable_monthly_savings_usd"] == Decimal("120.00") @pytest.mark.asyncio async def test_cost_provider_is_business_basis_only_and_reuses_efficiency(db_session): owner = person(70) integration = Integration( id=uuid4(), kind="cloud", provider="aws", name="private cost cloud", slug=f"intel-cost-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([owner, integration]) await db_session.flush() scopes = [ CmdbCloudScope( integration_id=integration.id, scope_type="aws_account", external_id=f"private-cost-scope-{index}", status="active", owner_person_id=owner.id, budget_usd_monthly=1000, budget_measure=measure, last_synced_at=NOW, ) for index, measure in enumerate(("list", "net"), start=1) ] servers = [ CmdbServer( integration_id=integration.id, instance_id=f"private-cost-server-{index}", status="running", owner_person_id=owner.id, last_synced_at=NOW, ) for index in range(2) ] db_session.add_all([*scopes, *servers]) await db_session.flush() def cost_fact( *, day: date, resource_id: str, list_cost: int | None, invoice_cost: int, owner_source: str, bucket: str = "usage", ) -> CmdbCostFact: return CmdbCostFact( fact_key=uuid4().hex, date=day, integration_id=str(integration.id), scope_id=str(scopes[0].id), resource_id=resource_id, resource_name="private resource", service="private service", amount_usd=invoice_cost, amount_native=invoice_cost, amount_usd_list=list_cost, amount_native_list=list_cost, cost_bucket=bucket, currency="USD", auto_owner_source=owner_source, ) last_day = NOW.date() - timedelta(days=1) current_days = [last_day - timedelta(days=value) for value in range(6, -1, -1)] rows = [ cost_fact( day=day, resource_id="private-resource-1", list_cost=200 if day == last_day else 10, invoice_cost=1, owner_source="rule", ) for day in current_days ] rows.extend( cost_fact( day=day, resource_id="private-resource-2", list_cost=40 if day == current_days[0] else 50, invoice_cost=1, owner_source="none", ) for day in current_days ) rows.extend( [ cost_fact( day=last_day - timedelta(days=30), resource_id="private-resource-1", list_cost=600, invoice_cost=1, owner_source="rule", ), # Invoice-only adjustment is outside the shared business basis. cost_fact( day=last_day, resource_id="private-resource-1", list_cost=None, invoice_cost=-999, owner_source="rule", bucket="adjustment", ), # Sparse current resource is counted as data but cannot be called # anomalous until it has the detector's seven-day baseline. cost_fact( day=last_day, resource_id="private-resource-3", list_cost=0, invoice_cost=0, owner_source="rule", ), ] ) db_session.add_all(rows) db_session.add_all( [ CmdbServerEfficiency( server_id=servers[0].id, integration_id=integration.id, verdict="idle", recommendation="stop", monthly_cost_usd=100, estimated_monthly_savings_usd=100, computed_at=NOW, ), CmdbServerEfficiency( server_id=servers[1].id, integration_id=integration.id, verdict="healthy", recommendation="keep", monthly_cost_usd=900, estimated_monthly_savings_usd=0, computed_at=NOW, ), ] ) await db_session.commit() result = await CostEfficiencyFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "cost_efficiency", "shared") ) metrics = {row.key: row for row in result.metrics} assert metrics["cost_efficiency.business_budget_coverage"].value == 50 assert metrics["cost_efficiency.anomaly_rate"].value == 50 assert metrics["cost_efficiency.anomaly_rate"].confidence == Decimal("66.67") assert metrics["cost_efficiency.optimizable_cost_rate"].value == 10 assert metrics["cost_efficiency.allocation_coverage"].value == Decimal("43.33") assert metrics["cost_efficiency.growth_rate"].value == 0 assert result.aggregate_facts["current_business_cost_usd"] == 600 assert result.aggregate_facts["previous_business_cost_usd"] == 600 assert result.aggregate_facts["estimated_monthly_savings_usd"] == 100 assert {row.kind for row in metrics["cost_efficiency.anomaly_rate"].evidence} == { "costresource", } assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_cost_provider_has_no_confident_facts_without_sources(db_session): result = await CostEfficiencyFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) assert all(row.confidence == 0 for row in result.metrics) assert result.data_as_of is None @pytest.mark.asyncio async def test_security_provider_reuses_auditor_ledgers_and_evidence_jobs(db_session): people = [person(80), person(81)] db_session.add_all(people) await db_session.flush() campaign = AccessCertification( label="private privileged review", campaign_type="privileged_role", status="in_progress", started_at=NOW - timedelta(days=10), due_at=NOW + timedelta(days=20), snapshot_item_count=2, decided_count=1, ) db_session.add(campaign) await db_session.flush() db_session.add_all( [ # Observability flags carry the values MfaSyncService really # writes per provider: Google answers both questions, GitHub # cannot report org 2FA *enforcement* at all. AuditorMfaStatus( person_id=people[0].id, provider="google", is_enrolled=True, is_enforced=True, enrollment_observed=True, enforcement_observed=True, last_synced_at=NOW, ), AuditorMfaStatus( person_id=people[1].id, provider="google", is_enrolled=False, is_enforced=True, enrollment_observed=True, enforcement_observed=True, last_synced_at=NOW, ), # Second row for the SAME person: must neither rescue the gap # above nor add a second slot to the person-level denominator. AuditorMfaStatus( person_id=people[1].id, provider="github", is_enrolled=True, is_enforced=False, enrollment_observed=True, enforcement_observed=False, last_synced_at=NOW, ), AccessCertificationItem( certification_id=campaign.id, person_id=people[0].id, access_type="privileged_role", access_label="private role", decision="keep", snapshot_at=NOW, ), AccessCertificationItem( certification_id=campaign.id, person_id=people[1].id, access_type="privileged_role", access_label="private role", decision="pending", snapshot_at=NOW, ), AuditorPublicExposureFinding( scan_run_id=uuid4(), source="aws_security_group", resource_id="private-exposure-1", exposure_type="sensitive_port", severity="critical", first_seen_at=NOW, last_seen_at=NOW, status="open", ), AuditorPublicExposureFinding( scan_run_id=uuid4(), source="aws_s3", resource_id="private-exposure-2", exposure_type="public_acl", severity="high", first_seen_at=NOW, last_seen_at=NOW, status="resolved", resolved_at=NOW, ), AuditorSecretRotationRecord( secret_kind="api_key", source="helix_apikey", resource_id="private-secret-1", status="overdue", last_synced_at=NOW, ), AuditorSecretRotationRecord( secret_kind="cloud_access_key", source="aws", resource_id="private-secret-2", status="current", last_synced_at=NOW, ), AuditorPatAudit( kind="helix_pat", source="helix_pat", token_external_id="private-pat-1", status="active", risk_flags=[], last_synced_at=NOW, ), AuditorPatAudit( kind="github_pat", source="github", token_external_id="private-pat-2", status="dormant", risk_flags=["dormant"], last_synced_at=NOW, ), ] ) for index, key in enumerate( ( "auditor.mfa_sync_daily", "auditor.public_exposure_scan_daily", "auditor.secret_rotation_check_daily", "auditor.pat_audit_daily", ) ): db_session.add( ScheduledJob( name=key, display_name="private audit job", handler_key=key, trigger_type="cron", cron_expr="0 0 * * *", enabled=True, last_run_at=NOW - timedelta(days=3) if index == 3 else NOW, last_status="failed" if index == 3 else "ok", ) ) await db_session.commit() result = await SecurityComplianceFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "security_compliance", "auditor") ) metrics = {row.key: row for row in result.metrics} # One of two persons covered — three rows, deduped to two persons, and # the enforcement-blind GitHub row excluded from both sides. assert metrics["security_compliance.mfa_enforcement_coverage"].value == 50 assert metrics["security_compliance.mfa_enforcement_coverage"].confidence == 100 assert result.aggregate_facts["mfa_person_count"] == 2 assert result.aggregate_facts["mfa_observable_person_count"] == 2 assert result.aggregate_facts["mfa_compliant_count"] == 1 assert metrics["security_compliance.privileged_review_coverage"].value == 50 assert metrics["security_compliance.public_exposure_rate"].value == 50 assert metrics["security_compliance.credential_risk_rate"].value == 50 assert metrics["security_compliance.evidence_freshness"].value == 3 assert metrics["security_compliance.evidence_freshness"].confidence == 100 assert {row.kind for row in metrics["security_compliance.credential_risk_rate"].evidence} == { "secretrecord", "pataudit", } assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_security_provider_has_no_confident_facts_without_sources(db_session): result = await SecurityComplianceFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) assert all(row.confidence == 0 for row in result.metrics) assert result.data_as_of is None @pytest.mark.asyncio async def test_security_provider_does_not_score_unbuilt_review_or_running_jobs_as_failures( db_session, ): db_session.add( AccessCertification( label="not materialized yet", campaign_type="privileged_role", status="in_progress", started_at=NOW - timedelta(hours=1), due_at=NOW + timedelta(days=30), snapshot_item_count=0, decided_count=0, ) ) for key in ( "auditor.mfa_sync_daily", "auditor.public_exposure_scan_daily", "auditor.secret_rotation_check_daily", "auditor.pat_audit_daily", ): db_session.add( ScheduledJob( name=key, display_name=key, handler_key=key, trigger_type="cron", cron_expr="0 0 * * *", enabled=True, last_run_at=NOW, last_status="running", ) ) await db_session.commit() result = await SecurityComplianceFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) metrics = {row.key: row for row in result.metrics} review = metrics["security_compliance.privileged_review_coverage"] assert review.value == 0 assert review.confidence == 0 freshness = metrics["security_compliance.evidence_freshness"] assert freshness.value == 0 assert freshness.confidence == 100 @pytest.mark.asyncio async def test_security_provider_accepts_completed_empty_privileged_review(db_session): db_session.add( AccessCertification( label="completed empty review", campaign_type="privileged_role", status="completed", started_at=NOW - timedelta(days=1), due_at=NOW, completed_at=NOW, snapshot_item_count=0, decided_count=0, ) ) await db_session.commit() result = await SecurityComplianceFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) review = next( row for row in result.metrics if row.key == "security_compliance.privileged_review_coverage" ) assert review.value == 100 assert review.confidence == 100 @pytest.mark.asyncio async def test_platform_operations_provider_reuses_durable_runtime_ledgers(db_session): people = [person(90), person(91)] db_session.add_all(people) job_ok = ScheduledJob( name="private.job.ok", display_name="private", handler_key="private.job.ok", trigger_type="interval", interval_seconds=3600, enabled=True, ) job_failed = ScheduledJob( name="private.job.failed", display_name="private", handler_key="private.job.failed", trigger_type="interval", interval_seconds=3600, enabled=True, ) db_session.add_all([job_ok, job_failed]) await db_session.flush() db_session.add_all([ ScheduledJobRun( job_id=job_ok.id, started_at=NOW - timedelta(hours=2), finished_at=NOW - timedelta(hours=2) + timedelta(seconds=1), status="ok", triggered_by="scheduler", ), ScheduledJobRun( job_id=job_failed.id, started_at=NOW - timedelta(hours=1), finished_at=NOW - timedelta(hours=1) + timedelta(seconds=1), status="failed", triggered_by="scheduler", error_message="private", ), ]) for index, status in enumerate((STATUS_COMPLETED, STATUS_FAILED)): db_session.add(ProvisioningTask( user_id=people[index].id, action="grant", target_type="seat", source="onboarding", execution_mode=EXEC_AUTO_API, status=status, )) db_session.add(ProvisioningTask( user_id=people[0].id, action="grant", target_type="seat", source="manual", execution_mode=EXEC_MANUAL_NOTIFY, status=STATUS_FAILED, )) db_session.add(ProvisioningTask( user_id=people[0].id, action="grant", target_type="seat", source="onboarding", execution_mode=EXEC_AUTO_API, status="pending", created_at=NOW - timedelta(days=2), updated_at=NOW - timedelta(days=2), )) for index, status in enumerate(("delivered", "failed")): db_session.add(NotificationOutbox( recipient_person_id=people[index].id, title="private", body_text="private", channels=["feishu"], status=status, )) db_session.add_all([ Integration( kind="hris", provider="successfactors", name="private healthy", slug=f"private-healthy-{uuid4().hex[:6]}", enabled=True, last_test_status="ok", last_tested_at=NOW - timedelta(days=1), ), Integration( kind="cloud", provider="aws", name="private stale", slug=f"private-stale-{uuid4().hex[:6]}", enabled=True, last_test_status="failed", last_tested_at=NOW - timedelta(days=1), ), Integration( kind="chat_approval", provider="feishu", name="private untested", slug=f"private-untested-{uuid4().hex[:6]}", enabled=True, ), ClientErrorDailyAggregate( day=NOW.date() - timedelta(days=1), surface="mobile", report_count=1, first_seen_at=NOW - timedelta(days=1), last_seen_at=NOW - timedelta(days=1), ), ClientErrorDailyAggregate( day=NOW.date() - timedelta(days=8), surface="mobile", report_count=100, first_seen_at=NOW - timedelta(days=8), last_seen_at=NOW - timedelta(days=8), ), PlatformSetting( key="observability.client_error_telemetry_started", value=True, description="private", created_at=NOW - timedelta(days=8), ), IdentityDataQualityRun( started_at=NOW - timedelta(hours=2), finished_at=NOW - timedelta(hours=1), status="ok", trigger_source="cron", open_total=1, open_high=1, ), IdentityDataQualityFinding( finding_type="orphan_login_identity", dedup_key="private-person", severity="high", person_id=people[0].id, title="private", status="open", ), IdentityDataQualityFinding( finding_type="missing_manager", dedup_key="private-person-2", severity="low", person_id=people[0].id, title="private", status="open", ), IdentityDataQualityFinding( finding_type="sync_drift_failed", dedup_key="private-person-3", severity="medium", person_id=people[1].id, title="private", status="open", ), ]) await db_session.commit() result = await PlatformOperationsFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "platform_ops", "global_admin") ) metrics = {row.key: row for row in result.metrics} assert metrics["platform_ops.scheduled_job_failure_rate"].value == 50 assert metrics["platform_ops.scheduled_job_failure_rate"].confidence == 100 assert metrics["platform_ops.provisioning_failure_rate"].value == Decimal("66.67") assert metrics["platform_ops.notification_failure_rate"].value == 50 connector = metrics["platform_ops.connector_health_coverage"] assert connector.value == Decimal("33.33") assert connector.confidence == Decimal("66.67") assert metrics["platform_ops.client_error_report_rate"].value == 50 assert metrics["platform_ops.client_error_report_rate"].confidence == 100 assert metrics["platform_ops.data_quality_finding_rate"].value == 100 assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_platform_operations_provider_has_no_confident_facts_without_sources(db_session): provider = PlatformOperationsFactProvider(now=lambda: NOW) result = await provider.collect( db_session, SnapshotSubject("platform", "platform", "shared") ) assert all(row.confidence == 0 for row in result.metrics) assert result.data_as_of is None built = await build_snapshot( db_session, subject=SnapshotSubject("platform", "platform", "platform_ops_test"), providers=[provider], ) platform_dimension = next( row for row in built.snapshot.dimensions if row["key"] == "platform_ops" ) assert platform_dimension["score"] is None @pytest.mark.asyncio async def test_software_api_provider_uses_observable_seats_and_governance_ledgers(db_session): owner = person(40) it_contact = person(41) key_owner = person(42) left_owner = person(43, employment_status="left") db_session.add_all([owner, it_contact, key_owner, left_owner]) await db_session.flush() complete_sw = AssetSoftwareCatalog( name="private software one", status="active", owner_person_id=owner.id, it_contact_person_id=it_contact.id, ) owner_gap_sw = AssetSoftwareCatalog( name="private software two", status="active", owner_person_id=owner.id, ) integration = Integration( id=uuid4(), kind="software_provisioning", provider="feishu", name="private integration", slug=f"intel-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) provider = ApiKeyProvider(name=f"private-provider-{uuid4().hex[:6]}", is_active=True) db_session.add_all([complete_sw, owner_gap_sw, integration, provider]) await db_session.flush() full_plan = SoftwareLicensePlan( software_id=complete_sw.id, name="private full plan", price=10, billing_cycle="monthly", max_seats=3, status="active", ) spare_plan = SoftwareLicensePlan( software_id=owner_gap_sw.id, name="private spare plan", price=10, billing_cycle="monthly", max_seats=1, status="active", ) db_session.add_all([full_plan, spare_plan]) await db_session.flush() db_session.add_all( [ SoftwareSeatAllocation( plan_id=full_plan.id, user_id=owner.id, status="active", idle_warned_at=NOW - timedelta(days=1), ), SoftwareSeatAllocation( plan_id=full_plan.id, user_id=it_contact.id, status="active", ), SoftwareSeatAllocation( plan_id=full_plan.id, user_id=key_owner.id, status="active", idle_kept_until=NOW + timedelta(days=30), ), SoftwareSeatSyncStat( integration_id=integration.id, software_id=complete_sw.id, plan_id=full_plan.id, provider="feishu", product_key=None, vendor_license_holders=2, holders_matched=1, holders_external=1, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=NOW, ), ApiKeyAssignment( provider_id=provider.id, assigned_to=key_owner.id, status="active", request_number=f"AK-{uuid4().hex[:10]}", project_name="private project", reason="private reason", monthly_budget_usd=100, valid_until=date(2026, 12, 31), ), ApiKeyAssignment( provider_id=provider.id, assigned_to=left_owner.id, status="active", request_number=f"AK-{uuid4().hex[:10]}", project_name="private project", reason="private reason", monthly_budget_usd=100, rotation_retire_at=NOW - timedelta(hours=1), ), ScheduledJob( name="software_seat.idle_reclaim", display_name="Idle reclaim", handler_key="software_seat.idle_reclaim", trigger_type="interval", interval_seconds=86_400, enabled=True, last_run_at=NOW, last_status="ok", last_result_summary={ "seats_scanned": 3, "undetermined": 0, "admin_skipped": 0, "leaver_skipped": 0, }, ), ] ) await db_session.flush() db_session.add( SoftwareExternalHolder( integration_id=integration.id, software_id=complete_sw.id, plan_id=full_plan.id, provider="feishu", holder_key="private-holder", category="external", resolved=False, first_observed_at=NOW, last_observed_at=NOW, ) ) await db_session.commit() result = await SoftwareApiFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "software_api", "asset_global") ) metrics = {row.key: row for row in result.metrics} # A future keep decision is observed/retained use, not a known-idle seat. assert metrics["software_api.seat_utilization"].value == Decimal("66.67") assert metrics["software_api.seat_utilization"].confidence == 100 assert metrics["software_api.capacity_headroom"].value == 50 assert metrics["software_api.owner_coverage"].value == 50 assert metrics["software_api.reconciliation_coverage"].value == 50 assert metrics["software_api.key_expiry_coverage"].value == 50 assert metrics["software_api.rotation_failure_rate"].value == 50 assert result.aggregate_facts["open_reconciliation_anomaly_count"] == 1 assert {row.kind for row in metrics["software_api.owner_coverage"].evidence} == { "software", "key", } assert_llm_safe(result.aggregate_facts) assert "private" not in str(result.aggregate_facts) @pytest.mark.asyncio async def test_software_api_provider_uses_one_complete_product_family_snapshot(db_session): owner = person(50) contact = person(51) db_session.add_all([owner, contact]) await db_session.flush() software = AssetSoftwareCatalog( name="private family software", status="active", owner_person_id=owner.id, it_contact_person_id=contact.id, ) integration = Integration( id=uuid4(), kind="software_provisioning", provider="atlassian", name="private family integration", slug=f"intel-family-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([software, integration]) await db_session.flush() plans = [ SoftwareLicensePlan( software_id=software.id, name=f"private plan {index}", price=10, billing_cycle="monthly", status="active", ) for index in range(2) ] db_session.add_all(plans) await db_session.flush() db_session.add_all( [ SoftwareSeatSyncStat( integration_id=integration.id, software_id=software.id, plan_id=plans[0].id, provider="atlassian", product_key="jira", vendor_license_holders=10, holders_matched=4, holders_external=6, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=NOW, ), SoftwareSeatSyncStat( integration_id=integration.id, software_id=software.id, plan_id=plans[1].id, provider="atlassian", product_key="jira", vendor_license_holders=8, holders_matched=8, holders_external=0, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=NOW, ), ] ) await db_session.commit() result = await SoftwareApiFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "software_api", "asset_global") ) metrics = {row.key: row for row in result.metrics} assert result.aggregate_facts["vendor_holder_count"] == 10 assert result.aggregate_facts["matched_holder_count"] == 4 assert metrics["software_api.reconciliation_coverage"].value == 40 @pytest.mark.asyncio async def test_software_api_provider_does_not_guess_utilization_without_telemetry(db_session): result = await SoftwareApiFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("platform", "platform", "shared") ) utilization = next( row for row in result.metrics if row.key == "software_api.seat_utilization" ) assert utilization.confidence == 0 assert result.data_as_of is None @pytest.mark.asyncio async def test_onboarding_sla_scores_shipped_72h_attainment_not_current_flag(db_session): """health-v3 口径 for ``lifecycle.onboarding_sla`` (2026-08-18). v2 read the CURRENT ``sla_breach_level``. ``OnboardingService. evaluate_sla`` returns ``none`` for every terminal process and resets to ``none`` as soon as the last blocker clears, so a hire whose hard gaps closed weeks past the SLA was recorded as an attainment. In production that reported 67.57% against a real 45.71%. v3 scores the shipped SLA (hard readiness within ``hire_date`` + 72h) over the hires whose window has already closed — a hire still inside the grace window is neither an attainment nor a miss. """ people = [person(index) for index in range(70, 74)] db_session.add_all(people) await db_session.flush() hire_date = date(2026, 8, 1) on_time = OnboardingProcess( person_id=people[0].id, hire_date=hire_date, status=PROC_COMPLETED, # Inside the 72h window — attained. hard_ready_at=datetime(2026, 8, 3, 2, tzinfo=UTC), ) # The v2 killer: finished 13 days late, yet ``sla_breach_level`` is # back to ``none`` because the process reached a terminal state. late_but_flag_clear = OnboardingProcess( person_id=people[1].id, hire_date=hire_date, status=PROC_COMPLETED, hard_ready_at=datetime(2026, 8, 14, 2, tzinfo=UTC), ) still_open_past_window = OnboardingProcess( person_id=people[2].id, hire_date=hire_date, status=PROC_PREPARING, sla_breach_level=ONBOARDING_BREACH, ) # Hired yesterday relative to NOW (2026-08-15): the SLA verdict is # not in yet, so this hire must leave the population entirely rather # than be scored as a miss. inside_grace = OnboardingProcess( person_id=people[3].id, hire_date=date(2026, 8, 14), status=PROC_PREPARING, ) db_session.add_all( [on_time, late_but_flag_clear, still_open_past_window, inside_grace] ) await db_session.commit() result = await LifecycleFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "lifecycle", "hr") ) metrics = {row.key: row for row in result.metrics} # 1 attainment out of the 3 hires whose window closed. Reading the # current flag instead would score 2 of 4 (50.00). assert metrics["lifecycle.onboarding_sla"].value == Decimal("33.33") assert result.aggregate_facts["onboarding_sla_total_count"] == 3 assert result.aggregate_facts["onboarding_sla_ok_count"] == 1 assert result.aggregate_facts["onboarding_sla_in_window_count"] == 1 gap_ids = {row.object_id for row in metrics["lifecycle.onboarding_sla"].evidence} assert str(late_but_flag_clear.id) in gap_ids assert str(inside_grace.id) not in gap_ids @pytest.mark.asyncio async def test_key_expiry_coverage_scores_only_helix_issued_credentials(db_session): """Adopted console keys and unissued approvals are not expiry gaps. Helix never held the plaintext of an adopted key (the reconciler leaves ``key_value`` NULL by design) so it cannot stamp a validity window on one; an ``approved`` row has no credential at all until an admin issues it. Counting both put the 95% band out of arithmetic reach — 44 of production's 61 "live" rows could never carry an expiry. The numerator is "has a bounded validity", not "expires more than 30 days out": applications are capped at one year, so the old rule dropped every correctly-governed key for its final month. """ holder = person(80) db_session.add(holder) await db_session.flush() provider = ApiKeyProvider(name=f"private-provider-{uuid4().hex[:6]}", is_active=True) db_session.add(provider) await db_session.flush() def assignment(status, *, valid_until=None, detail=None): return ApiKeyAssignment( provider_id=provider.id, assigned_to=holder.id, status=status, request_number=f"AK-{uuid4().hex[:10]}", project_name="private project", reason="private reason", monthly_budget_usd=100, valid_until=valid_until, detail=detail or {}, ) governed = assignment("active", valid_until=date(2027, 1, 31)) # Governed too: 20 days out is the expiry-alert cron doing its job. expiring_soon = assignment("active", valid_until=date(2026, 9, 4)) ungoverned = assignment("active") adopted = assignment("active", detail={"source": "adopted"}) adopted_manual = assignment("active", detail={"source": "adopted_manual"}) unissued = assignment("approved") db_session.add_all( [governed, expiring_soon, ungoverned, adopted, adopted_manual, unissued] ) await db_session.commit() result = await SoftwareApiFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "software_api", "asset_global") ) metrics = {row.key: row for row in result.metrics} # 2 of the 3 Helix-issued live credentials carry a bounded validity. assert metrics["software_api.key_expiry_coverage"].value == Decimal("66.67") assert result.aggregate_facts["issued_key_count"] == 3 assert result.aggregate_facts["expiry_governed_key_count"] == 2 assert result.aggregate_facts["adopted_key_count"] == 2 assert result.aggregate_facts["pending_issuance_key_count"] == 1 assert result.aggregate_facts["expiring_within_30d_key_count"] == 1 assert result.aggregate_facts["live_key_count"] == 6 gap_ids = {row.object_id for row in metrics["software_api.key_expiry_coverage"].evidence} assert gap_ids == {str(ungoverned.id)} @pytest.mark.asyncio async def test_reconciliation_coverage_nets_registered_external_holders(db_session): """备案保留 external accounts are reconciled, not anomalies. ``asset.router``'s seat-reconciliation summary subtracts KNOWN classifications (``shared_service`` / ``external_guest`` / ``contractor``) from the anomaly count. Scoring them as unmatched made the health metric contradict the panel that owns the same rows and put 95% out of reach — 105 of production's 326 unmatched vendor seats are registered reservations. """ owner = person(90) contact = person(91) db_session.add_all([owner, contact]) await db_session.flush() software = AssetSoftwareCatalog( name="private recon software", status="active", owner_person_id=owner.id, it_contact_person_id=contact.id, ) integration = Integration( id=uuid4(), kind="software_provisioning", provider="feishu", name="private integration", slug=f"intel-{uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add_all([software, integration]) await db_session.flush() plan = SoftwareLicensePlan( software_id=software.id, name="private recon plan", price=10, billing_cycle="monthly", status="active", ) db_session.add(plan) await db_session.flush() registered_email = f"partner-{uuid4().hex[:8]}@example.invalid" untriaged_email = f"ghost-{uuid4().hex[:8]}@example.invalid" db_session.add_all([ SoftwareSeatSyncStat( integration_id=integration.id, software_id=software.id, plan_id=plan.id, provider="feishu", product_key=None, vendor_license_holders=4, holders_matched=2, holders_external=2, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=NOW, ), SoftwareExternalHolder( integration_id=integration.id, software_id=software.id, plan_id=plan.id, provider="feishu", holder_key=registered_email, email=registered_email, category="external", resolved=False, first_observed_at=NOW, last_observed_at=NOW, ), SoftwareExternalHolder( integration_id=integration.id, software_id=software.id, plan_id=plan.id, provider="feishu", holder_key=untriaged_email, email=untriaged_email, category="external", resolved=False, first_observed_at=NOW, last_observed_at=NOW, ), ExternalAccountRegistry( email=registered_email, classification="external_guest", ), ]) await db_session.commit() result = await SoftwareApiFactProvider(now=lambda: NOW).collect( db_session, SnapshotSubject("domain", "software_api", "asset_global") ) metrics = {row.key: row for row in result.metrics} # 2 matched employees + 1 registered reservation of 4 vendor seats. assert metrics["software_api.reconciliation_coverage"].value == 75 assert result.aggregate_facts["known_reserved_holder_count"] == 1 assert result.aggregate_facts["attributed_holder_count"] == 3 # Evidence must exclude what the metric already counted as handled. assert result.aggregate_facts["open_reconciliation_anomaly_count"] == 1 anomaly_ids = { row.object_id for row in metrics["software_api.reconciliation_coverage"].evidence } assert len(anomaly_ids) == 1