fix: stabilize workflow manifest canonicalization
This commit is contained in:
@@ -15,7 +15,7 @@ class ManifestDriftError(RuntimeError):
|
|||||||
|
|
||||||
def canonical_manifest_json(manifest: ContractManifest) -> str:
|
def canonical_manifest_json(manifest: ContractManifest) -> str:
|
||||||
try:
|
try:
|
||||||
return json.dumps(manifest, ensure_ascii=False, indent=2) + "\n"
|
return json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||||
except (TypeError, ValueError) as error:
|
except (TypeError, ValueError) as error:
|
||||||
raise ValueError(f"manifest is not canonically serializable: {error}") from error
|
raise ValueError(f"manifest is not canonically serializable: {error}") from error
|
||||||
|
|
||||||
|
|||||||
@@ -10,6 +10,14 @@ UNION_RESULTS = {
|
|||||||
"CreateArtifactFromWorkspaceResult",
|
"CreateArtifactFromWorkspaceResult",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
AUTH_SECURITY_COMPONENTS = {
|
||||||
|
"AuthRecordSummaryPayload",
|
||||||
|
"ListAuthRecordsResult",
|
||||||
|
"DeleteAuthRecordResult",
|
||||||
|
"SourceAuthDiagnosisPayload",
|
||||||
|
"SourceDiagnosisResult",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_generates_the_complete_real_workflow_contract() -> None:
|
def test_generates_the_complete_real_workflow_contract() -> None:
|
||||||
manifest = generate_manifest()
|
manifest = generate_manifest()
|
||||||
@@ -29,11 +37,8 @@ def test_generates_the_complete_real_workflow_contract() -> None:
|
|||||||
def test_generated_contract_preserves_security_and_extension_boundaries() -> None:
|
def test_generated_contract_preserves_security_and_extension_boundaries() -> None:
|
||||||
schemas = generate_manifest()["components"]["schemas"]
|
schemas = generate_manifest()["components"]["schemas"]
|
||||||
|
|
||||||
auth_result_names = [
|
assert AUTH_SECURITY_COMPONENTS <= schemas.keys()
|
||||||
name for name in schemas if "Auth" in name and name.endswith("Result")
|
for name in AUTH_SECURITY_COMPONENTS:
|
||||||
]
|
|
||||||
assert auth_result_names
|
|
||||||
for name in auth_result_names:
|
|
||||||
properties = schemas[name].get("properties", {})
|
properties = schemas[name].get("properties", {})
|
||||||
assert isinstance(properties, dict)
|
assert isinstance(properties, dict)
|
||||||
assert "payload" not in properties
|
assert "payload" not in properties
|
||||||
|
|||||||
@@ -1,6 +1,8 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Mapping
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -19,6 +21,17 @@ def _manifest():
|
|||||||
return manifest_from_openrpc(synthetic_openrpc_document())
|
return manifest_from_openrpc(synthetic_openrpc_document())
|
||||||
|
|
||||||
|
|
||||||
|
def _reverse_mapping_insertions(value: Any) -> Any:
|
||||||
|
if isinstance(value, Mapping):
|
||||||
|
return {
|
||||||
|
key: _reverse_mapping_insertions(child)
|
||||||
|
for key, child in reversed(list(value.items()))
|
||||||
|
}
|
||||||
|
if isinstance(value, list):
|
||||||
|
return [_reverse_mapping_insertions(item) for item in value]
|
||||||
|
return value
|
||||||
|
|
||||||
|
|
||||||
def test_canonical_json_is_stable_utf8_text_with_trailing_newline() -> None:
|
def test_canonical_json_is_stable_utf8_text_with_trailing_newline() -> None:
|
||||||
first = canonical_manifest_json(_manifest())
|
first = canonical_manifest_json(_manifest())
|
||||||
second = canonical_manifest_json(_manifest())
|
second = canonical_manifest_json(_manifest())
|
||||||
@@ -29,6 +42,15 @@ def test_canonical_json_is_stable_utf8_text_with_trailing_newline() -> None:
|
|||||||
assert "\\u" not in first
|
assert "\\u" not in first
|
||||||
|
|
||||||
|
|
||||||
|
def test_canonical_json_ignores_recursive_mapping_insertion_order() -> None:
|
||||||
|
manifest = _manifest()
|
||||||
|
reordered = _reverse_mapping_insertions(manifest)
|
||||||
|
|
||||||
|
assert canonical_manifest_json(manifest).encode("utf-8") == (
|
||||||
|
canonical_manifest_json(reordered).encode("utf-8")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_write_and_check_round_trip(tmp_path: Path) -> None:
|
def test_write_and_check_round_trip(tmp_path: Path) -> None:
|
||||||
path = tmp_path / "workflow-api.manifest.json"
|
path = tmp_path / "workflow-api.manifest.json"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user