repo-manager/tests/test_workloads.py
tegwick b36b68bc57 feat(workloads): define authoritative reference contract
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
2026-08-22 14:03:07 +02:00

104 lines
3 KiB
Python

from __future__ import annotations
from pathlib import Path
from repo_manager.workloads import index_workloads, resolve_workload
def _declaration(
root: Path,
rapp_id: str,
name: str,
deployables: list[str],
) -> None:
path = root / rapp_id / "declarations" / "rapp.yaml"
path.parent.mkdir(parents=True)
rendered = "\n".join(f" - {item}" for item in deployables)
path.write_text(
f"""kind: managed-workload-package
rapp_id: {rapp_id}
ownership_repo: {name}
readiness_state: declared
workload_identity:
name: {name}
data_classification: confidential
criticality: high
composition:
member_repos:
- repo: {rapp_id}
deployables:
{rendered}
""",
encoding="utf-8",
)
def test_indexes_authoritative_workload_identity_and_deployables(tmp_path: Path) -> None:
_declaration(tmp_path, "rapp-issue-core", "issue-core", ["issue-core"])
result = index_workloads(tmp_path)
assert result["ok"] is True
assert result["declaration_count"] == 1
assert result["workloads"][0]["reference"] == {
"rapp_id": "rapp-issue-core",
"name": "issue-core",
}
assert result["workloads"][0]["deployables"] == ["issue-core"]
def test_reports_missing_rapp_declaration_without_inventing_identity(tmp_path: Path) -> None:
(tmp_path / "rapp-undeclared").mkdir()
result = index_workloads(tmp_path)
assert result["ok"] is False
assert result["workloads"] == []
assert result["errors"][0]["code"] == "declaration_missing"
def test_reports_deployable_owned_by_more_than_one_rapp(tmp_path: Path) -> None:
_declaration(tmp_path, "rapp-one", "one", ["shared-api"])
_declaration(tmp_path, "rapp-two", "two", ["shared-api"])
result = index_workloads(tmp_path)
assert result["ok"] is False
assert any(error["code"] == "deployable_duplicate" for error in result["errors"])
def test_resolves_only_explicit_rapp_and_workload_pair(tmp_path: Path) -> None:
_declaration(tmp_path, "rapp-qonto", "qonto", ["qonto-assistant"])
resolved = resolve_workload(
tmp_path,
rapp_id="rapp-qonto",
name="qonto",
deployable="qonto-assistant",
)
unknown = resolve_workload(tmp_path, rapp_id="rapp-qonto", name="rapp-qonto")
assert resolved["status"] == "resolved"
assert resolved["workload"]["name"] == "qonto"
assert unknown == {
"ok": True,
"contract": "helixforge.workload-reference/v1",
"status": "unknown",
"reference": {"rapp_id": "rapp-qonto", "name": "rapp-qonto"},
"reason": "not_found",
"index_errors": [],
}
def test_unknown_deployable_is_not_inferred_from_repository_name(tmp_path: Path) -> None:
_declaration(tmp_path, "rapp-openbao", "openbao", ["openbao-api"])
result = resolve_workload(
tmp_path,
rapp_id="rapp-openbao",
name="openbao",
deployable="rapp-openbao",
)
assert result["status"] == "unknown"
assert result["reason"] == "deployable_not_declared"