awoooi/apps/api/tests/test_ai_automation_asset_capability_matrix.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from pathlib import Path

from src.jobs.asset_capability_reconciliation_job import (
    build_reconciliation_summary_payload,
    build_reconciliation_work_item_input,
)
from src.services.ai_automation_asset_capability_matrix import (
    REQUIRED_CONTROL_IDS,
    STAGE_IDS,
    build_asset_capability_matrix,
    build_reconciliation_candidates,
)
from src.services.awoooi_priority_work_order_readback import (
    apply_ai_automation_asset_capability_matrix,
    get_ai_automation_program_scope_classes,
)

_REPO_ROOT = Path(__file__).resolve().parents[3]


def _single_host_scope() -> list[dict[str, object]]:
    return [
        {
            "id": "hosts",
            "label": "Hosts",
            "asset_count": 1,
            "asset_ids": ["99"],
            "inventory_sources": ["host-inventory.json"],
        }
    ]


def _fully_covered_host(now: datetime) -> dict[str, object]:
    dimensions = (
        "auto_monitoring",
        "auto_alerting",
        "auto_rule_creation",
        "auto_rule_matching",
        "auto_playbook",
        "auto_remediation",
        "auto_km_creation",
    )
    return {
        "asset_id": 42,
        "asset_key": "host:99",
        "asset_type": "host",
        "environment": "prod",
        "namespace": "",
        "name": "99",
        "owner_team": "platform_sre",
        "source_repo": "wooo/awoooi",
        "lifecycle_state": "active",
        "last_seen_at": now,
        "coverage": {
            dimension: {
                "coverage_status": "green",
                "created_at": now,
            }
            for dimension in dimensions
        },
        "compliance": {
            "backup_tested": {
                "status": "compliant",
                "detected_at": now,
            }
        },
        "change_types": [],
    }


def test_declared_catalog_expands_to_193_unique_assets() -> None:
    matrix = build_asset_capability_matrix(
        get_ai_automation_program_scope_classes(),
        repo_root=_REPO_ROOT,
    )

    assert matrix["summary"]["declared_asset_count"] == 193
    assert matrix["summary"]["matrix_asset_count"] == 193
    assert len({asset["canonical_id"] for asset in matrix["assets"]}) == 193
    assert {
        item["scope_id"]: item["declared_asset_count"]
        for item in matrix["scope_rollups"]
    } == {
        "hosts": 7,
        "services": 31,
        "products": 12,
        "websites": 31,
        "tools": 23,
        "packages": 61,
        "data_and_backup": 18,
        "observability_and_security": 10,
    }
    assert matrix["controls"]["full_scope_class_coverage"] is True
    assert matrix["operation_boundaries"]["raw_asset_key_exposed"] is False


def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> None:
    now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
    initial = build_asset_capability_matrix(
        _single_host_scope(),
        live_assets=[_fully_covered_host(now)],
        latest_run={"run_id": "run-1", "status": "success", "ended_at": now},
        repo_root=_REPO_ROOT,
        generated_at=now,
    )
    receipt = {
        "op_id": "receipt-1",
        "status": "success",
        "created_at": now,
        "input": {
            "trace_id": "trace-1",
            "automation_run_id": "automation-run-1",
            "work_item_id": "AIA-P0-006",
        },
        "output": {
            "matrix_fingerprint": initial["matrix_fingerprint"],
            "candidate_count": 0,
            "repository_readback_count": 0,
            "repository_readback_verified": True,
        },
    }

    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        live_assets=[_fully_covered_host(now)],
        latest_run={"run_id": "run-1", "status": "success", "ended_at": now},
        reconciliation_receipt=receipt,
        repo_root=_REPO_ROOT,
        generated_at=now,
    )

    asset = matrix["assets"][0]
    assert set(asset["stages"]) == set(STAGE_IDS)
    assert {stage["status"] for stage in asset["stages"].values()} == {"covered"}
    assert all(matrix["controls"][control_id] for control_id in REQUIRED_CONTROL_IDS)
    assert matrix["closed"] is True
    assert matrix["summary"]["gap_asset_count"] == 0


def test_outdated_live_receipts_are_classified_stale_per_stage() -> None:
    now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC)
    old = now - timedelta(hours=40)
    live = _fully_covered_host(old)

    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        live_assets=[live],
        latest_run={"run_id": "run-old", "status": "success", "ended_at": old},
        repo_root=_REPO_ROOT,
        generated_at=now,
    )

    asset = matrix["assets"][0]
    assert asset["stages"]["runtime_identity"]["status"] == "stale"
    assert asset["stages"]["sensor"]["status"] == "stale"
    assert asset["stages"]["policy"]["status"] == "stale"
    assert asset["stages"]["executor"]["status"] == "stale"
    assert asset["stages"]["verifier"]["status"] == "stale"
    assert asset["stages"]["backup_restore"]["status"] == "stale"
    assert asset["stages"]["learning_writeback"]["status"] == "stale"
    assert asset["stale_stage_ids"] == [
        "runtime_identity",
        "sensor",
        "policy",
        "executor",
        "verifier",
        "backup_restore",
        "learning_writeback",
    ]


def test_reconciliation_candidates_are_stable_and_public_safe() -> None:
    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        repo_root=_REPO_ROOT,
        generated_at=datetime(2026, 7, 11, 12, 0, tzinfo=UTC),
    )

    first = build_reconciliation_candidates(matrix)
    second = build_reconciliation_candidates(matrix)

    assert first == second
    assert len(first) == 1
    assert first[0]["work_item_id"].startswith("AIA-P0-006-WI-")
    assert first[0]["candidate_kind"] == "capability_gap"
    assert first[0]["controlled_apply_allowed"] is True
    assert "asset_key" not in first[0]
    assert "host" not in first[0]


def test_reconciliation_summary_requires_repository_readback() -> None:
    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        latest_run={"run_id": "discovery-run-1", "status": "success"},
        repo_root=_REPO_ROOT,
    )
    candidates = build_reconciliation_candidates(matrix)

    failed = build_reconciliation_summary_payload(
        matrix,
        candidates,
        trace_id="trace-1",
        automation_run_id="run-1",
        work_item_id="AIA-P0-006",
        repository_readback_count=0,
        created_count=1,
        existing_count=0,
    )
    passed = build_reconciliation_summary_payload(
        matrix,
        candidates,
        trace_id="trace-1",
        automation_run_id="run-1",
        work_item_id="AIA-P0-006",
        repository_readback_count=1,
        created_count=1,
        existing_count=0,
    )

    assert failed["repository_readback_verified"] is False
    assert failed["closed"] is False
    assert passed["repository_readback_verified"] is True
    assert passed["closed"] is True


def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None:
    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        repo_root=_REPO_ROOT,
    )
    candidate = build_reconciliation_candidates(matrix)[0]

    receipt_input = build_reconciliation_work_item_input(
        candidate,
        trace_id="trace-1",
        automation_run_id="reconciliation-run-1",
        program_work_item_id="AIA-P0-006",
        matrix_fingerprint=matrix["matrix_fingerprint"],
    )

    assert receipt_input["trace_id"] == "trace-1"
    assert receipt_input["automation_run_id"] == "reconciliation-run-1"
    assert receipt_input["program_work_item_id"] == "AIA-P0-006"
    assert (
        receipt_input["source_diff"]["matrix_fingerprint"]
        == (matrix["matrix_fingerprint"])
    )


def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None:
    matrix = build_asset_capability_matrix(
        _single_host_scope(),
        repo_root=_REPO_ROOT,
    )
    payload: dict[str, object] = {"summary": {}, "rollups": {}}

    apply_ai_automation_asset_capability_matrix(payload, matrix)

    assert payload["ai_automation_asset_capability_matrix"]["schema_version"] == (
        "ai_automation_asset_capability_matrix_v1"
    )
    assert payload["ai_automation_asset_capability_matrix"]["assets"] == []
    assert payload["ai_automation_asset_capability_matrix"]["assets_embedded"] is False
    assert payload["ai_automation_asset_capability_matrix"]["assets_endpoint"] == (
        "/api/v1/agents/ai-automation-asset-capability-matrix"
    )
    assert (
        payload["summary"]["ai_automation_asset_capability_declared_asset_count"] == 1
    )
    work_item = next(
        item
        for item in payload["ai_automation_program_ledger"]["work_items"]
        if item["id"] == "AIA-P0-006"
    )
    assert work_item["status"] == "in_progress"
    assert work_item["runtime_progress"]["required_control_count"] == len(
        REQUIRED_CONTROL_IDS
    )