tegwick
598f6418e7
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
import hashlib
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from tests.conftest import (
|
|
create_test_domain,
|
|
create_test_repo,
|
|
create_test_topic,
|
|
create_test_workplan,
|
|
)
|
|
|
|
|
|
def _digest(value):
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode()
|
|
).hexdigest()
|
|
|
|
|
|
def _contract(artifact_digest="a" * 64):
|
|
return {
|
|
"schema_version": "review-contract/v1",
|
|
"contract_key": "TEST-REVIEW-1",
|
|
"subject": {"kind": "task", "id": "TEST-WP-1-T01"},
|
|
"allowed_dispositions": ["approve", "request_changes"],
|
|
"artifacts": {
|
|
"docs/runbook.md": {"algorithm": "sha256", "digest": artifact_digest}
|
|
},
|
|
"owners": [
|
|
{
|
|
"id": "railiance-infra",
|
|
"artifact_ids": ["docs/runbook.md"],
|
|
"assertions": [
|
|
{
|
|
"id": "safe-boundary",
|
|
"statement": "No live action is authorized.",
|
|
}
|
|
],
|
|
"check_ids": ["runbook-lint"],
|
|
}
|
|
],
|
|
"gates": [
|
|
{
|
|
"id": "TEST-WP-1-T01",
|
|
"policy": "all_required",
|
|
"owners": ["railiance-infra"],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _source(repo, path, revision):
|
|
return {"repo": repo, "path": path, "revision": revision}
|
|
|
|
|
|
def _approval(contract, revision="2" * 40):
|
|
return {
|
|
"owner_id": "railiance-infra",
|
|
"actor": "operator:test",
|
|
"disposition": "approve",
|
|
"contract_digest": _digest(contract),
|
|
"artifact_hashes": {
|
|
"docs/runbook.md": contract["artifacts"]["docs/runbook.md"]["digest"]
|
|
},
|
|
"assertion_ids": ["safe-boundary"],
|
|
"checks": [{"id": "runbook-lint", "passed": True, "read_only": True}],
|
|
"source": _source("railiance-infra", "reviews/receipt.json", revision),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_review_round_trip_and_idempotency(client):
|
|
contract = _contract()
|
|
projected = await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": contract,
|
|
"source": _source("test-owner", "interfaces/review.json", "1" * 40),
|
|
},
|
|
)
|
|
assert projected.status_code == 201, projected.text
|
|
assert projected.json()["contract_digest"] == _digest(contract)
|
|
|
|
missing = await client.get("/review-contracts/TEST-REVIEW-1/aggregate")
|
|
assert missing.json()["satisfied"] is False
|
|
assert missing.json()["authorizes_execution"] is False
|
|
assert missing.json()["owners"][0]["status"] == "missing"
|
|
|
|
approval = _approval(contract)
|
|
first = await client.post("/review-contracts/TEST-REVIEW-1/receipts", json=approval)
|
|
second = await client.post(
|
|
"/review-contracts/TEST-REVIEW-1/receipts", json=approval
|
|
)
|
|
assert first.status_code == second.status_code == 201
|
|
assert first.json()["id"] == second.json()["id"]
|
|
assert first.json()["submitted_at"].endswith("+00:00") or first.json()[
|
|
"submitted_at"
|
|
].endswith("Z")
|
|
|
|
result = (await client.get("/review-contracts/TEST-REVIEW-1/aggregate")).json()
|
|
assert result["satisfied"] is True
|
|
assert result["owners"][0]["status"] == "approved"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_receipt_rejects_wrong_owner_and_artifact_drift(client):
|
|
contract = _contract()
|
|
await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": contract,
|
|
"source": _source("test-owner", "interfaces/review.json", "1" * 40),
|
|
},
|
|
)
|
|
wrong_owner = _approval(contract)
|
|
wrong_owner["owner_id"] = "somebody-else"
|
|
assert (
|
|
await client.post("/review-contracts/TEST-REVIEW-1/receipts", json=wrong_owner)
|
|
).status_code == 422
|
|
|
|
drift = _approval(contract)
|
|
drift["artifact_hashes"]["docs/runbook.md"] = "f" * 64
|
|
assert (
|
|
await client.post("/review-contracts/TEST-REVIEW-1/receipts", json=drift)
|
|
).status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_contract_revision_marks_prior_receipt_stale(client):
|
|
old = _contract()
|
|
await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": old,
|
|
"source": _source("test-owner", "interfaces/review.json", "1" * 40),
|
|
},
|
|
)
|
|
await client.post("/review-contracts/TEST-REVIEW-1/receipts", json=_approval(old))
|
|
|
|
new = _contract("b" * 64)
|
|
replaced = await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": new,
|
|
"source": _source("test-owner", "interfaces/review.json", "3" * 40),
|
|
},
|
|
)
|
|
assert replaced.status_code == 201, replaced.text
|
|
state = (await client.get("/review-contracts/TEST-REVIEW-1/aggregate")).json()
|
|
assert state["owners"][0]["status"] == "stale"
|
|
assert state["satisfied"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_changes_supersedes_approval(client):
|
|
contract = _contract()
|
|
await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": contract,
|
|
"source": _source("test-owner", "interfaces/review.json", "1" * 40),
|
|
},
|
|
)
|
|
await client.post(
|
|
"/review-contracts/TEST-REVIEW-1/receipts", json=_approval(contract)
|
|
)
|
|
change = {
|
|
"owner_id": "railiance-infra",
|
|
"actor": "operator:test",
|
|
"disposition": "request_changes",
|
|
"contract_digest": _digest(contract),
|
|
"note": "Clarify the recovery owner.",
|
|
"source": _source("railiance-infra", "reviews/request-changes.json", "4" * 40),
|
|
}
|
|
assert (
|
|
await client.post("/review-contracts/TEST-REVIEW-1/receipts", json=change)
|
|
).status_code == 201
|
|
state = (await client.get("/review-contracts/TEST-REVIEW-1/aggregate")).json()
|
|
assert state["owners"][0]["status"] == "request_changes"
|
|
assert state["satisfied"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_required_review_blocks_legacy_decision_resolution(client):
|
|
domain = await create_test_domain(client)
|
|
topic = await create_test_topic(client, domain_slug=domain["slug"])
|
|
repo = await create_test_repo(client, domain_slug=domain["slug"])
|
|
workplan = await create_test_workplan(client, repo["id"], topic["id"])
|
|
decision = (
|
|
await client.post(
|
|
"/decisions/",
|
|
json={"title": "Run attended procedure", "workplan_id": workplan["id"]},
|
|
)
|
|
).json()
|
|
contract = _contract()
|
|
projected = await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": contract,
|
|
"source": _source("test-owner", "interfaces/review.json", "1" * 40),
|
|
"decision_id": decision["id"],
|
|
"required_for_decision": True,
|
|
},
|
|
)
|
|
assert projected.status_code == 201, projected.text
|
|
blocked = await client.post(
|
|
f"/decisions/{decision['id']}/resolve",
|
|
json={
|
|
"rationale": "Proceed",
|
|
"decided_by": "operator:test",
|
|
"write_log": False,
|
|
},
|
|
)
|
|
assert blocked.status_code == 409
|
|
|
|
await client.post(
|
|
"/review-contracts/TEST-REVIEW-1/receipts", json=_approval(contract)
|
|
)
|
|
resolved = await client.post(
|
|
f"/decisions/{decision['id']}/resolve",
|
|
json={
|
|
"rationale": "Reviewed",
|
|
"decided_by": "operator:test",
|
|
"write_log": False,
|
|
},
|
|
)
|
|
assert resolved.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_wp0024_contract_adapter(client):
|
|
legacy = {
|
|
"interface": "railiance.owner-review",
|
|
"version": 1,
|
|
"workplan_id": "RAILIANCE-WP-0024",
|
|
"decisions": ["approve", "request-changes"],
|
|
"task_owners": {"RAILIANCE-WP-0024-T03": ["railiance-infra"]},
|
|
"owners": {
|
|
"railiance-infra": {
|
|
"artifacts": ["docs/runbook.md"],
|
|
"assertions": ["No live execution is authorized."],
|
|
"checks": ["node-reboot-preflight"],
|
|
}
|
|
},
|
|
"artifact_sha256": {"docs/runbook.md": "a" * 64},
|
|
}
|
|
result = await client.post(
|
|
"/review-contracts/projections",
|
|
json={
|
|
"contract": legacy,
|
|
"source": _source(
|
|
"railiance-platform",
|
|
"interfaces/RAILIANCE-WP-0024-owner-reviews.json",
|
|
"5" * 40,
|
|
),
|
|
},
|
|
)
|
|
assert result.status_code == 201, result.text
|
|
body = result.json()
|
|
assert body["contract_key"] == "RAILIANCE-WP-0024"
|
|
assert body["contract_digest"] == _digest(legacy)
|
|
assert body["document"]["owners"][0]["assertions"][0]["id"].startswith("assertion:")
|