activity-core/tests/test_sbom_nexus_context_resolver.py

"""sbom-nexus bounded catch-up resolver (ACTIVITY-WP-0030-T01/T02)."""

from __future__ import annotations

from typing import Any

import httpx
import pytest

from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY
from activity_core.context_resolvers.sbom_nexus import SbomNexusContextResolver


class DummyResponse:
    def __init__(self, payload: Any, status_code: int = 200) -> None:
        self._payload = payload
        self.status_code = status_code
        self.text = str(payload)

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise httpx.HTTPStatusError(
                f"HTTP {self.status_code}",
                request=httpx.Request("GET", "http://sbom-nexus.test"),
                response=None,  # type: ignore[arg-type]
            )

    def json(self) -> Any:
        return self._payload


class DummyClient:
    """Records the single ranked call the contract allows."""

    def __init__(self, payload: Any, status_code: int = 200) -> None:
        self._payload = payload
        self._status_code = status_code
        self.calls: list[tuple[str, dict[str, Any] | None]] = []
        self.posts: list[tuple[str, dict[str, Any] | None]] = []
        self.post_results: dict[str, list[DummyResponse]] = {}

    def __enter__(self) -> "DummyClient":
        return self

    def __exit__(self, *args: Any) -> None:
        return None

    def get(self, url: str, params: dict[str, Any] | None = None) -> DummyResponse:
        self.calls.append((url, params))
        return DummyResponse(self._payload, self._status_code)

    def post(self, url: str, json: dict[str, Any] | None = None) -> DummyResponse:
        self.posts.append((url, json))
        for suffix, results in self.post_results.items():
            if url.endswith(suffix) and results:
                return results.pop(0)
        raise AssertionError(f"unexpected POST {url}")


def _install(monkeypatch, payload: Any, status_code: int = 200) -> DummyClient:
    client = DummyClient(payload, status_code)
    monkeypatch.setattr(httpx, "Client", lambda **kwargs: client)
    return client


def _payload(repos: list[dict[str, Any]], **counts: Any) -> dict[str, Any]:
    base = {"stale_count": 111, "never_count": 93, "total_count": 111}
    base.update(counts)
    base["repos"] = repos
    return base


def test_registered_as_sbom_nexus_source_type() -> None:
    assert CONTEXT_RESOLVER_REGISTRY["sbom-nexus"] is SbomNexusContextResolver


def test_catch_up_returns_ranked_targets_and_fleet_counts(monkeypatch) -> None:
    client = _install(
        monkeypatch,
        _payload(
            [
                {
                    "repo_slug": "never-scanned",
                    "last_sbom_at": None,
                    "sbom_age_days": 9999,
                    "has_sbom": False,
                    "checkout_available": True,
                },
                {
                    "repo_slug": "activity-core",
                    "last_sbom_at": "2026-04-26T00:00:00Z",
                    "sbom_age_days": 117,
                    "has_sbom": True,
                    "checkout_available": True,
                },
                {
                    "repo_slug": "no-checkout",
                    "last_sbom_at": "2026-05-01T00:00:00Z",
                    "sbom_age_days": 112,
                    "has_sbom": True,
                    "checkout_available": False,
                },
            ]
        ),
    )

    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 3})

    # One ranked call, not a per-repo walk.
    assert len(client.calls) == 1
    url, params = client.calls[0]
    assert url.endswith("/sbom/catch-up")
    assert params == {"limit": 3}

    assert [repo["repo_slug"] for repo in result["repos"]] == [
        "never-scanned",
        "activity-core",
        "no-checkout",
    ]
    assert result["selected_count"] == 3
    assert result["limit"] == 3
    assert result["stale_count"] == 111
    assert result["never_count"] == 93
    assert result["total_count"] == 111
    assert result["repos"][0]["has_sbom"] is False
    assert result["repos"][2]["checkout_available"] is False


def test_catch_up_defaults_to_three(monkeypatch) -> None:
    client = _install(monkeypatch, _payload([]))

    result = SbomNexusContextResolver().resolve("catch_up", None, {})

    assert client.calls[0][1] == {"limit": 3}
    assert result["limit"] == 3
    assert result["repos"] == []
    assert result["selected_count"] == 0


@pytest.mark.parametrize("requested", [0, -5, 500, "seven", None])
def test_catch_up_limit_stays_bounded(monkeypatch, requested: Any) -> None:
    _install(monkeypatch, _payload([]))

    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": requested})

    assert 1 <= result["limit"] <= 25


def test_catch_up_truncates_an_over_long_response(monkeypatch) -> None:
    """The bounded side-effect in T02 must never exceed the declared N."""
    repos = [
        {"repo_slug": f"repo-{index}", "last_sbom_at": None, "has_sbom": False}
        for index in range(10)
    ]
    _install(monkeypatch, _payload(repos))

    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 2})

    assert len(result["repos"]) == 2
    assert result["selected_count"] == 2


def test_apply_processes_at_most_limit_and_records_terminal_outcomes(monkeypatch) -> None:
    repos = [
        {
            "repo_slug": "no-checkout",
            "last_sbom_at": None,
            "has_sbom": False,
            "checkout_available": False,
        },
        {
            "repo_slug": "scan-me",
            "last_sbom_at": None,
            "has_sbom": False,
            "checkout_available": True,
        },
        {
            "repo_slug": "not-selected",
            "last_sbom_at": None,
            "has_sbom": False,
            "checkout_available": True,
        },
    ]
    client = _install(monkeypatch, _payload(repos))
    client.post_results = {
        "/sbom/no-checkout/skip": [
            DummyResponse(
                {
                    "repo_slug": "no-checkout",
                    "status": "skipped",
                    "reason": "no-checkout",
                    "snapshot_id": "skip-1",
                }
            )
        ],
        "/sbom/scan-me/ingest": [
            DummyResponse(
                {
                    "repo_slug": "scan-me",
                    "status": "ingested",
                    "snapshot_id": "scan-1",
                    "entry_count": 12,
                }
            )
        ],
    }

    result = SbomNexusContextResolver().resolve(
        "catch_up", None, {"limit": 2, "apply": True}
    )

    assert result["attempted_count"] == 2
    assert result["updated"] == [
        {
            "repo_slug": "scan-me",
            "status": "ingested",
            "snapshot_id": "scan-1",
            "entry_count": 12,
        }
    ]
    assert result["skipped"] == [
        {
            "repo_slug": "no-checkout",
            "status": "skipped",
            "reason": "no-checkout",
            "snapshot_id": "skip-1",
        }
    ]
    assert len(client.posts) == 2
    assert all("not-selected" not in url for url, _body in client.posts)


def test_apply_records_ingest_error_when_ingest_fails(monkeypatch) -> None:
    client = _install(
        monkeypatch,
        _payload(
            [
                {
                    "repo_slug": "broken",
                    "last_sbom_at": None,
                    "has_sbom": False,
                    "checkout_available": True,
                }
            ]
        ),
    )
    client.post_results = {
        "/sbom/broken/ingest": [DummyResponse({}, status_code=503)],
        "/sbom/broken/skip": [
            DummyResponse(
                {
                    "repo_slug": "broken",
                    "status": "skipped",
                    "reason": "ingest-error",
                    "snapshot_id": "skip-error-1",
                }
            )
        ],
    }

    result = SbomNexusContextResolver().resolve(
        "catch_up", None, {"limit": 3, "apply": True}
    )

    assert result["attempted_count"] == 1
    assert result["updated"] == []
    assert result["skipped"][0]["reason"] == "ingest-error"
    assert client.posts[1][1] == {
        "reason": "ingest-error",
        "detail": "HTTPStatusError",
    }


def test_read_only_default_never_posts(monkeypatch) -> None:
    client = _install(monkeypatch, _payload([{"repo_slug": "selected"}]))

    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 1})

    assert result["selected_count"] == 1
    assert "attempted_count" not in result
    assert client.posts == []


def test_catch_up_normalises_partial_entries(monkeypatch) -> None:
    _install(
        monkeypatch,
        _payload(
            [
                {"repo_slug": "sparse"},
                {"repo_slug": ""},
                "not-an-object",
                {"last_sbom_at": "2026-01-01T00:00:00Z"},
            ]
        ),
    )

    result = SbomNexusContextResolver().resolve("catch_up", None, {"limit": 10})

    assert [repo["repo_slug"] for repo in result["repos"]] == ["sparse"]
    sparse = result["repos"][0]
    assert sparse["has_sbom"] is False
    assert sparse["last_sbom_at"] is None
    assert sparse["sbom_age_days"] == 9999
    assert sparse["checkout_available"] is None


def test_catch_up_rejects_a_response_without_repos(monkeypatch) -> None:
    _install(monkeypatch, {"stale_count": 111})

    with pytest.raises(RuntimeError, match="missing required key: repos"):
        SbomNexusContextResolver().resolve("catch_up", None, {})


def test_catch_up_rejects_a_non_object_response(monkeypatch) -> None:
    _install(monkeypatch, ["activity-core"])

    with pytest.raises(RuntimeError, match="non-object response"):
        SbomNexusContextResolver().resolve("catch_up", None, {})


def test_unknown_query_returns_empty(monkeypatch) -> None:
    _install(monkeypatch, _payload([]))

    assert SbomNexusContextResolver().resolve("ingest", None, {}) == {}


def test_base_url_is_configurable(monkeypatch) -> None:
    client = _install(monkeypatch, _payload([]))
    monkeypatch.setenv("SBOM_NEXUS_URL", "http://sbom-nexus.test:9000/")

    SbomNexusContextResolver().resolve("catch_up", None, {})

    assert client.calls[0][0] == "http://sbom-nexus.test:9000/sbom/catch-up"


def _instruction():
    from activity_core.models import InstructionDef

    return InstructionDef.model_validate(
        {
            "id": "daily-sbom-catchup-report",
            "trusted_fields": [],
            "model": "deterministic",
            "temperature": 0,
            "max_tokens": 1,
            "prompt": "Deterministic SBOM catch-up report.",
            "output_schema": "",
            "review_required": False,
            "report_sinks": [
                {
                    "type": "state-hub-progress",
                    "event_type": "sbom_catchup",
                    "author": "activity-core",
                }
            ],
        }
    )


def test_deterministic_report_names_selected_repos() -> None:
    from activity_core.rules.executor import _deterministic_context_report

    context = {
        "catchup": {
            "repos": [
                {
                    "repo_slug": "never-scanned",
                    "last_sbom_at": None,
                    "sbom_age_days": 9999,
                    "has_sbom": False,
                    "checkout_available": True,
                },
                {
                    "repo_slug": "no-checkout",
                    "last_sbom_at": "2026-05-01T00:00:00Z",
                    "sbom_age_days": 112,
                    "has_sbom": True,
                    "checkout_available": False,
                },
            ],
            "selected_count": 2,
            "stale_count": 111,
            "never_count": 93,
            "total_count": 111,
            "limit": 3,
        }
    }

    result = _deterministic_context_report(_instruction(), context)

    assert result.tasks == []
    report = result.report
    assert report["deterministic"] is True
    assert report["never_count"] == 93
    assert [r["repo_slug"] for r in report["selected_repos"]] == [
        "never-scanned",
        "no-checkout",
    ]
    assert "93 never scanned of 111 repos" in report["summary"]
    assert report["updated_repos"] == []
    assert report["skipped_repos"] == []


def test_deterministic_report_carries_t02_ingest_outcomes() -> None:
    """Forward-compatible with the bounded ingest side-effect (T02)."""
    from activity_core.rules.executor import _deterministic_context_report

    context = {
        "catchup": {
            "repos": [{"repo_slug": "a"}, {"repo_slug": "b"}],
            "updated": [{"repo_slug": "a"}],
            "skipped": [{"repo_slug": "b", "reason": "no-checkout"}],
            "stale_count": 110,
            "never_count": 92,
            "total_count": 111,
            "limit": 3,
        }
    }

    report = _deterministic_context_report(_instruction(), context).report

    assert report["updated_repos"] == ["a"]
    assert report["skipped_repos"] == [{"repo_slug": "b", "reason": "no-checkout"}]
    assert "1 updated, 1 skipped" in report["summary"]


def test_daily_definition_is_bounded_and_enabled() -> None:
    from pathlib import Path

    from activity_core.definition_parser import parse_file

    definition = parse_file(Path("activity-definitions/daily-sbom-catchup.md"))

    assert definition.enabled is True
    # No rule block: the weekly flood came from `for_each` over every stale repo.
    assert definition.rules == []
    source = definition.context_sources[0]
    assert source["type"] == "sbom-nexus"
    assert source["query"] == "catch_up"
    assert source["params"]["limit"] == 3
    assert source["params"]["apply"] is True
    assert source["bind_to"] == "context.catchup"
    instruction = definition.instructions[0]
    assert instruction["model"] == "deterministic"
    assert instruction["report_sinks"][0]["event_type"] == "sbom_catchup"


def test_railiance_projection_enables_the_same_bounded_contract(tmp_path) -> None:
    from pathlib import Path

    import yaml
    from activity_core.definition_parser import parse_file

    documents = list(
        yaml.safe_load_all(Path("k8s/railiance/20-runtime.yaml").read_text())
    )
    config = next(
        item
        for item in documents
        if isinstance(item, dict)
        and item.get("kind") == "ConfigMap"
        and item.get("metadata", {}).get("name")
        == "actcore-external-activity-definitions"
    )
    projected = tmp_path / "daily-sbom-catchup.md"
    projected.write_text(config["data"]["daily-sbom-catchup.md"])

    definition = parse_file(projected)

    assert definition.enabled is True
    assert definition.rules == []
    source = definition.context_sources[0]
    assert source["params"] == {"limit": 3, "apply": True}