feat: generate canonical workflow contract

This commit is contained in:
lda
2026-08-03 07:30:31 +07:00 Verified
parent 55cfddd781
commit 99f0671771
5 changed files with 181 additions and 0 deletions
@@ -0,0 +1,51 @@
from __future__ import annotations
from wf_contract_manifest import generate_manifest
UNION_RESULTS = {
"InspectCapabilityResult",
"PatchDraftResult",
"ValidateDraftResult",
"CompileDraftWorkspaceResult",
"CreateArtifactFromWorkspaceResult",
}
def test_generates_the_complete_real_workflow_contract() -> None:
manifest = generate_manifest()
schemas = manifest["components"]["schemas"]
assert len(manifest["operations"]) == 70
assert len({operation["method"] for operation in manifest["operations"]}) == 70
assert len(schemas) == 126
assert len(manifest["components"]["errors"]) == 1
assert all(
set(operation["result"]["schema"]) == {"$ref"}
for operation in manifest["operations"]
)
assert {name for name in UNION_RESULTS if "anyOf" in schemas[name]} == UNION_RESULTS
def test_generated_contract_preserves_security_and_extension_boundaries() -> None:
schemas = generate_manifest()["components"]["schemas"]
auth_result_names = [
name for name in schemas if "Auth" in name and name.endswith("Result")
]
assert auth_result_names
for name in auth_result_names:
properties = schemas[name].get("properties", {})
assert isinstance(properties, dict)
assert "payload" not in properties
assert schemas["SourceDiagnosisResult"]["additionalProperties"] is True
assert schemas["RegistryEntryPayload"]["additionalProperties"] is True
def test_generated_contract_contains_no_temporary_or_transport_state() -> None:
serialized = str(generate_manifest())
assert "TemporaryDirectory" not in serialized
assert "\\\\Temp\\\\" not in serialized
assert "127.0.0.1" not in serialized
assert '"/rpc"' not in serialized
+51
View File
@@ -0,0 +1,51 @@
from __future__ import annotations
from pathlib import Path
import pytest
from wf_contract_manifest import (
ManifestDriftError,
canonical_manifest_json,
check_manifest,
manifest_from_openrpc,
write_manifest,
)
from .fixtures import synthetic_openrpc_document
def _manifest():
return manifest_from_openrpc(synthetic_openrpc_document())
def test_canonical_json_is_stable_utf8_text_with_trailing_newline() -> None:
first = canonical_manifest_json(_manifest())
second = canonical_manifest_json(_manifest())
assert first == second
assert first.endswith("\n")
assert ' "manifest_version": 1' in first
assert "\\u" not in first
def test_write_and_check_round_trip(tmp_path: Path) -> None:
path = tmp_path / "workflow-api.manifest.json"
assert write_manifest(_manifest(), path) == path
check_manifest(_manifest(), path)
assert path.read_bytes() == canonical_manifest_json(_manifest()).encode("utf-8")
def test_check_reports_drift_without_mutating_the_file(tmp_path: Path) -> None:
path = tmp_path / "workflow-api.manifest.json"
path.write_text("stale\n", encoding="utf-8")
before = path.read_bytes()
with pytest.raises(
ManifestDriftError, match="python -m wf_contract_manifest write"
):
check_manifest(_manifest(), path)
assert path.read_bytes() == before