196 lines
7.3 KiB
Python
196 lines
7.3 KiB
Python
"""Structural contracts for the handoff acceptance catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_handoff_support import load_handoff_module
|
|
|
|
catalog = load_handoff_module("hermes_handoff_catalog")
|
|
harness_run = load_handoff_module("hermes_handoff_run")
|
|
evaluators = load_handoff_module("hermes_handoff_evaluators")
|
|
model = load_handoff_module("hermes_handoff_model")
|
|
policy = load_handoff_module("hermes_handoff_policy")
|
|
exec_module = load_handoff_module("hermes_handoff_exec")
|
|
|
|
NOW = dt.datetime(2026, 8, 17, 12, 0, 0, tzinfo=dt.timezone.utc)
|
|
TARGETS = catalog.Targets(now=NOW)
|
|
SPECS = harness_run.build_catalog(TARGETS)
|
|
BY_ID = {check.id: check for check in SPECS}
|
|
|
|
VANTAGES = {
|
|
catalog.OPERATOR: exec_module.operator_vantage(context="atlas-operator"),
|
|
catalog.SELF: exec_module.pod_vantage("hermes", "hermes-agent-1", "hermes"),
|
|
catalog.SWITCHYARD: exec_module.pod_vantage(
|
|
"hermes", "hermes-switchyard-1", "switchyard"
|
|
),
|
|
catalog.NODE: exec_module.pod_vantage(
|
|
"hermes", "hermes-node-ssh-access-1", "key-reconciler"
|
|
),
|
|
catalog.CHAT: exec_module.pod_vantage("hermes", "hermes-chat-tenant-0", "hermes"),
|
|
}
|
|
|
|
|
|
def test_the_catalog_is_structurally_valid() -> None:
|
|
assert harness_run.validate_catalog(SPECS) == []
|
|
assert len(SPECS) >= 60
|
|
|
|
|
|
def test_every_step_is_accepted_by_the_read_only_policy_from_its_vantage() -> None:
|
|
"""A catalog entry that cannot run under policy is a bug found before a run."""
|
|
for check in SPECS:
|
|
for step in check.steps:
|
|
addressed = VANTAGES[step.vantage].wrap(step.argv)
|
|
policy.check_argv(addressed, policy.READ_ONLY)
|
|
|
|
|
|
def test_no_step_constructs_a_mutation_or_dry_run() -> None:
|
|
"""Authorization review may name a verb; no executed subcommand mutates."""
|
|
for check in SPECS:
|
|
for step in check.steps:
|
|
subcommand = next(policy.positionals(step.argv), "")
|
|
assert subcommand not in {"apply", "create", "delete", "patch", "replace"}
|
|
assert not any(argument.startswith("--dry-run") for argument in step.argv)
|
|
|
|
|
|
def test_no_catalog_step_constructs_impersonation() -> None:
|
|
operator_steps = [
|
|
step
|
|
for check in SPECS
|
|
for step in check.steps
|
|
if step.vantage == catalog.OPERATOR
|
|
]
|
|
assert operator_steps
|
|
assert not any(
|
|
set(step.argv) & set(policy.IMPERSONATION_ARGS) for step in operator_steps
|
|
)
|
|
|
|
assert not any(
|
|
set(step.argv) & set(policy.IMPERSONATION_ARGS)
|
|
for check in SPECS
|
|
for step in check.steps
|
|
)
|
|
|
|
|
|
def test_credential_or_mutation_denials_are_review_only() -> None:
|
|
deny_checks = [check for check in SPECS if check.rule == "review_denied"]
|
|
assert len(deny_checks) >= 5
|
|
for check in deny_checks:
|
|
assert check.steps and all(step.kind == model.REVIEW for step in check.steps), (
|
|
check.id
|
|
)
|
|
|
|
|
|
def test_both_vantages_are_represented_across_the_catalog() -> None:
|
|
used = {step.vantage for check in SPECS for step in check.steps}
|
|
assert catalog.OPERATOR in used and catalog.SELF in used
|
|
assert used <= set(catalog.VANTAGE_NAMES)
|
|
|
|
|
|
def test_every_rule_named_by_a_check_is_registered() -> None:
|
|
for check in SPECS:
|
|
if check.scope == model.EPHEMERAL:
|
|
assert check.rule == "not_armed"
|
|
continue
|
|
assert check.rule in evaluators.EVALUATORS, check.id
|
|
|
|
|
|
def test_every_evidence_step_referenced_by_expect_exists() -> None:
|
|
for check in SPECS:
|
|
keys = {step.key for step in check.steps}
|
|
named = [check.expect.get("step")] + list(check.expect.get("steps", ()))
|
|
for key in filter(None, named):
|
|
assert key in keys, f"{check.id} names missing step {key}"
|
|
|
|
|
|
def test_the_default_catalog_reports_every_mutating_check_as_not_armed() -> None:
|
|
ephemeral = [check for check in SPECS if check.scope == model.EPHEMERAL]
|
|
assert len(ephemeral) == 4
|
|
for check in ephemeral:
|
|
assert check.mandatory is False
|
|
assert evaluators.evaluate(check, {}).status == model.NOT_RUN
|
|
|
|
|
|
def test_dependency_pull_requests_each_get_their_own_merge_gate() -> None:
|
|
for number in TARGETS.dependency_pull_requests:
|
|
check = BY_ID[f"baseline.dependency-pr-{number}-merged"]
|
|
assert check.mandatory
|
|
assert check.expect["fields"]["merged"] is True
|
|
assert check.expect["fields"]["base.ref"] == "main"
|
|
assert "head.sha" in check.expect["fields"]
|
|
|
|
|
|
def test_the_baseline_commit_gate_names_the_configured_commit() -> None:
|
|
check = BY_ID["baseline.origin-main-descends-merged-work"]
|
|
assert TARGETS.baseline_commit in check.steps[0].argv
|
|
assert "--is-ancestor" in check.steps[0].argv
|
|
|
|
|
|
def test_pool_assignment_and_chat_continuity_are_unconditional() -> None:
|
|
assert "pool.assignment-safety-knobs-are-configured" in BY_ID
|
|
assert "pool.assignment-fieldrefs-are-bound" in BY_ID
|
|
assert "surfaces.chat-runs-the-telegram-topic-revision" in BY_ID
|
|
assert "surfaces.chat-telegram-sessions-are-continuous" in BY_ID
|
|
assert TARGETS.node_count == TARGETS.pool_replicas == 3
|
|
|
|
|
|
def test_bulk_evidence_is_parsed_but_not_pasted_into_the_report() -> None:
|
|
routing = BY_ID["routing.provider-effort-and-fallback-evidence"]
|
|
assert routing.steps[0].record is False
|
|
assert routing.steps[0].max_bytes == TARGETS.routing_tail_bytes
|
|
|
|
kanban = BY_ID["surfaces.kanban-activity-is-visible"]
|
|
assert kanban.steps[0].record is False
|
|
|
|
|
|
def test_freshness_gates_carry_the_run_clock() -> None:
|
|
for identifier in (
|
|
"identity.codex-evidence-is-fresh",
|
|
"identity.claude-evidence-is-fresh",
|
|
):
|
|
assert BY_ID[identifier].expect["now"] == NOW
|
|
routing = BY_ID["routing.provider-effort-and-fallback-evidence"]
|
|
assert routing.expect["now"] == NOW
|
|
assert routing.expect["max_age_seconds"] == TARGETS.max_evidence_age_seconds
|
|
|
|
|
|
def test_check_identifiers_are_unique_and_grouped() -> None:
|
|
identifiers = [check.id for check in SPECS]
|
|
assert len(identifiers) == len(set(identifiers))
|
|
for check in SPECS:
|
|
assert check.id.split(".")[0]
|
|
assert check.group
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"identifier",
|
|
[
|
|
"identity.no-provider-api-key-in-pod",
|
|
"identity.no-provider-api-key-in-manifest",
|
|
"forge.worker-holds-no-git-credential-variable",
|
|
"access.secrets-are-denied",
|
|
"access.service-account-tokens-are-denied",
|
|
"access.workload-mutation-is-denied",
|
|
"access.pod-exec-is-denied",
|
|
"access.required-reads-are-allowed",
|
|
"access.flux-and-helm-status-are-allowed",
|
|
"access.namespace-view-agrees-across-vantages",
|
|
"nodes.hardening-covers-every-node",
|
|
"build.builder-service-account-is-tokenless",
|
|
"build.harbor-immutability-rule-is-applied",
|
|
"release.exact-lineage-is-running",
|
|
"reliability.finalization-and-replay-patches-are-deployed",
|
|
"pool.three-workers-have-distinct-nodes-and-volumes",
|
|
"pool.coordinator-retains-sole-state-ownership",
|
|
"surfaces.kanban-activity-is-visible",
|
|
"surfaces.chat-telegram-topic-state-is-durable",
|
|
"gitops.kustomizations-reconcile-cleanly",
|
|
"routing.provider-effort-and-fallback-evidence",
|
|
"scopes.agent-profile-is-distinct",
|
|
],
|
|
)
|
|
def test_the_handoff_requirements_each_have_a_mandatory_check(identifier: str) -> None:
|
|
assert BY_ID[identifier].mandatory is True
|