tegwick
3b3e1a1ff0
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f
700 lines
21 KiB
Python
700 lines
21 KiB
Python
"""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,
|
|
apply_bounded_ingest,
|
|
)
|
|
|
|
|
|
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, dict[str, str]]] = []
|
|
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,
|
|
headers: dict[str, str] | None = None,
|
|
) -> DummyResponse:
|
|
self.posts.append((url, json, headers or {}))
|
|
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_declared_apply_remains_read_only_during_selection(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))
|
|
result = SbomNexusContextResolver().resolve(
|
|
"catch_up", None, {"limit": 2, "apply": True}
|
|
)
|
|
|
|
assert result["selected_count"] == 2
|
|
assert "attempted_count" not in result
|
|
assert client.posts == []
|
|
|
|
|
|
def test_apply_processes_only_fixed_targets_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,
|
|
},
|
|
]
|
|
client = _install(monkeypatch, _payload([]))
|
|
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 = apply_bounded_ingest(
|
|
repos,
|
|
operation_id="run-1",
|
|
)
|
|
|
|
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
|
|
first_key = client.posts[0][2]["Idempotency-Key"]
|
|
assert first_key == client.posts[0][2]["X-Activity-Core-Operation-ID"]
|
|
assert first_key != client.posts[1][2]["Idempotency-Key"]
|
|
|
|
|
|
def test_ambiguous_ingest_failure_does_not_create_synthetic_skip(monkeypatch) -> None:
|
|
client = _install(
|
|
monkeypatch,
|
|
_payload([]),
|
|
)
|
|
client.post_results = {
|
|
"/sbom/broken/ingest": [DummyResponse({}, status_code=503)],
|
|
}
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
apply_bounded_ingest(
|
|
[{"repo_slug": "broken", "checkout_available": True}],
|
|
operation_id="run-ambiguous",
|
|
)
|
|
|
|
assert len(client.posts) == 1
|
|
assert client.posts[0][0].endswith("/sbom/broken/ingest")
|
|
|
|
|
|
def test_retry_resumes_heartbeat_outcomes(monkeypatch) -> None:
|
|
repos = [
|
|
{"repo_slug": "first", "checkout_available": True},
|
|
{"repo_slug": "second", "checkout_available": True},
|
|
]
|
|
first_client = _install(monkeypatch, _payload([]))
|
|
first_client.post_results = {
|
|
"/sbom/first/ingest": [
|
|
DummyResponse(
|
|
{
|
|
"repo_slug": "first",
|
|
"status": "ingested",
|
|
"snapshot_id": "snapshot-first",
|
|
}
|
|
)
|
|
],
|
|
"/sbom/second/ingest": [DummyResponse({}, status_code=503)],
|
|
}
|
|
heartbeats: list[list[dict[str, Any]]] = []
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
apply_bounded_ingest(
|
|
repos,
|
|
operation_id="run-retry",
|
|
on_progress=lambda outcomes: heartbeats.append(outcomes),
|
|
)
|
|
|
|
completed = heartbeats[-1]
|
|
retry_client = _install(monkeypatch, _payload([]))
|
|
retry_client.post_results = {
|
|
"/sbom/second/ingest": [
|
|
DummyResponse(
|
|
{
|
|
"repo_slug": "second",
|
|
"status": "ingested",
|
|
"snapshot_id": "snapshot-second",
|
|
}
|
|
)
|
|
]
|
|
}
|
|
|
|
result = apply_bounded_ingest(
|
|
repos,
|
|
operation_id="run-retry",
|
|
completed=completed,
|
|
)
|
|
|
|
assert [outcome["repo_slug"] for outcome in result["updated"]] == [
|
|
"first",
|
|
"second",
|
|
]
|
|
assert len(retry_client.posts) == 1
|
|
assert retry_client.posts[0][0].endswith("/sbom/second/ingest")
|
|
|
|
|
|
def test_retry_reuses_stable_identity_for_same_run_and_repo(monkeypatch) -> None:
|
|
repo = {"repo_slug": "ambiguous", "checkout_available": True}
|
|
first_client = _install(monkeypatch, _payload([]))
|
|
first_client.post_results = {
|
|
"/sbom/ambiguous/ingest": [DummyResponse({}, status_code=503)],
|
|
}
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
apply_bounded_ingest([repo], operation_id="run-stable")
|
|
|
|
retry_client = _install(monkeypatch, _payload([]))
|
|
retry_client.post_results = {
|
|
"/sbom/ambiguous/ingest": [
|
|
DummyResponse(
|
|
{
|
|
"repo_slug": "ambiguous",
|
|
"status": "ingested",
|
|
"snapshot_id": "snapshot-ambiguous",
|
|
}
|
|
)
|
|
],
|
|
}
|
|
apply_bounded_ingest([repo], operation_id="run-stable")
|
|
|
|
first_headers = first_client.posts[0][2]
|
|
retry_headers = retry_client.posts[0][2]
|
|
assert retry_headers["Idempotency-Key"] == first_headers["Idempotency-Key"]
|
|
assert retry_headers["X-Activity-Core-Operation-ID"] == first_headers[
|
|
"X-Activity-Core-Operation-ID"
|
|
]
|
|
|
|
|
|
def test_apply_collapses_duplicate_targets(monkeypatch) -> None:
|
|
client = _install(monkeypatch, _payload([]))
|
|
client.post_results = {
|
|
"/sbom/duplicate/skip": [
|
|
DummyResponse(
|
|
{
|
|
"repo_slug": "duplicate",
|
|
"status": "skipped",
|
|
"reason": "no-checkout",
|
|
}
|
|
)
|
|
]
|
|
}
|
|
|
|
result = apply_bounded_ingest(
|
|
[
|
|
{"repo_slug": "duplicate", "checkout_available": False},
|
|
{"repo_slug": "duplicate", "checkout_available": False},
|
|
],
|
|
operation_id="run-duplicates",
|
|
)
|
|
|
|
assert result["attempted_count"] == 1
|
|
assert len(client.posts) == 1
|
|
|
|
|
|
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 == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedicated_activity_applies_truncated_workflow_selection(monkeypatch) -> None:
|
|
from activity_core import activities
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_apply(repos, *, operation_id, completed, on_progress):
|
|
captured.update(
|
|
repos=repos,
|
|
operation_id=operation_id,
|
|
completed=completed,
|
|
)
|
|
return {"attempted_count": len(repos), "updated": [], "skipped": []}
|
|
|
|
monkeypatch.setattr(
|
|
"activity_core.context_resolvers.sbom_nexus.apply_bounded_ingest",
|
|
fake_apply,
|
|
)
|
|
patch = await activities.apply_sbom_catchup(
|
|
{
|
|
"run_id": "run-fixed",
|
|
"context_sources": [
|
|
{
|
|
"type": "sbom-nexus",
|
|
"query": "catch_up",
|
|
"bind_to": "context.catchup",
|
|
"params": {"limit": 2, "apply": True},
|
|
}
|
|
],
|
|
"context": {
|
|
"catchup": {
|
|
"limit": 2,
|
|
"repos": [
|
|
{"repo_slug": "a"},
|
|
{"repo_slug": "b"},
|
|
{"repo_slug": "must-not-run"},
|
|
],
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert [repo["repo_slug"] for repo in captured["repos"]] == ["a", "b"]
|
|
assert captured["operation_id"] == "run-fixed"
|
|
assert patch["catchup"]["attempted_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedicated_activity_forwards_heartbeat_outcomes(monkeypatch) -> None:
|
|
from activity_core import activities
|
|
|
|
completed = [
|
|
{
|
|
"repo_slug": "already-done",
|
|
"status": "skipped",
|
|
"reason": "no-checkout",
|
|
}
|
|
]
|
|
captured: dict[str, Any] = {}
|
|
|
|
monkeypatch.setattr(
|
|
activities,
|
|
"_sbom_heartbeat_state",
|
|
lambda run_id: {
|
|
"run_id": run_id,
|
|
"outcomes_by_bind": {"catchup": completed},
|
|
},
|
|
)
|
|
|
|
def fake_apply(repos, *, operation_id, completed, on_progress):
|
|
captured["completed"] = completed
|
|
return {"attempted_count": 1, "updated": [], "skipped": completed}
|
|
|
|
monkeypatch.setattr(
|
|
"activity_core.context_resolvers.sbom_nexus.apply_bounded_ingest",
|
|
fake_apply,
|
|
)
|
|
await activities.apply_sbom_catchup(
|
|
{
|
|
"run_id": "run-resumed",
|
|
"context_sources": [
|
|
{
|
|
"type": "sbom-nexus",
|
|
"query": "catch_up",
|
|
"bind_to": "context.catchup",
|
|
"params": {"limit": 1, "apply": True},
|
|
}
|
|
],
|
|
"context": {
|
|
"catchup": {
|
|
"limit": 1,
|
|
"repos": [{"repo_slug": "already-done"}],
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
assert captured["completed"] == completed
|
|
|
|
|
|
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}
|