lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve named title contract schemas

Browse Source
This commit is contained in:
lda
2026-08-03 08:38:19 +07:00 Verified
parent 859d251755
commit 231e023a3c
8 changed files with 371 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/wf_contract_manifest/fixtures.py
+8 -1
View File
@@ -63,8 +63,15 @@ def synthetic_openrpc_document() -> dict[str, Any]:
"properties": {
"mode": {"title": "Mode", "const": "alpha"},
"payload": {},
"title": {"title": "Display Title", "type": "string"},
},
"required": ["mode", "payload", "title"],
"$defs": {
"title": {
"title": "Reusable Title",
"type": "string",
}
},
"required": ["mode", "payload"],
"if": {"properties": {"mode": {"const": "alpha"}}},
"then": {"required": ["payload"]},
"not": {"required": ["forbidden"]},
tests/wf_contract_manifest/test_generate.py
+99 -6
View File
@@ -1,5 +1,8 @@
from __future__ import annotations
from collections.abc import Iterator, Mapping
from typing import Any
from wf_contract_manifest import generate_manifest
UNION_RESULTS = {
@@ -18,6 +21,69 @@ AUTH_SECURITY_COMPONENTS = {
"SourceDiagnosisResult",
}
AUTH_METHODS = {
"workflow.admin.auth.delete",
"workflow.admin.auth.inspect",
"workflow.admin.auth.list",
"workflow.admin.auth.save",
}
def _schema_references(value: Any) -> Iterator[str]:
if isinstance(value, Mapping):
reference = value.get("$ref")
if isinstance(reference, str):
yield reference
for child in value.values():
yield from _schema_references(child)
elif isinstance(value, list):
for child in value:
yield from _schema_references(child)
def _reachable_schema_names(schemas: Mapping[str, Any], roots: set[str]) -> set[str]:
"""Return schema components reachable through local schema references."""
reachable: set[str] = set()
pending = list(roots)
while pending:
name = pending.pop()
if name in reachable:
continue
reachable.add(name)
prefix = "#/components/schemas/"
for reference in _schema_references(schemas[name]):
if reference.startswith(prefix):
pending.append(reference.removeprefix(prefix))
return reachable
def _structured_strings(value: Any) -> Iterator[str]:
if isinstance(value, str):
yield value
elif isinstance(value, Mapping):
for key, child in value.items():
yield str(key)
yield from _structured_strings(child)
elif isinstance(value, list):
for child in value:
yield from _structured_strings(child)
def _schema_objects(value: Any) -> Iterator[Mapping[str, Any]]:
if isinstance(value, Mapping):
yield value
for child in value.values():
yield from _schema_objects(child)
elif isinstance(value, list):
for child in value:
yield from _schema_objects(child)
def _result_component_name(operation: Any) -> str:
reference = operation["result"]["schema"]["$ref"]
assert isinstance(reference, str)
return reference.removeprefix("#/components/schemas/")
def test_generates_the_complete_real_workflow_contract() -> None:
manifest = generate_manifest()
@@ -35,7 +101,8 @@ def test_generates_the_complete_real_workflow_contract() -> None:
def test_generated_contract_preserves_security_and_extension_boundaries() -> None:
schemas = generate_manifest()["components"]["schemas"]
manifest = generate_manifest()
schemas = manifest["components"]["schemas"]
assert AUTH_SECURITY_COMPONENTS <= schemas.keys()
for name in AUTH_SECURITY_COMPONENTS:
@@ -43,14 +110,40 @@ def test_generated_contract_preserves_security_and_extension_boundaries() -> Non
assert isinstance(properties, dict)
assert "payload" not in properties
result_components = {
_result_component_name(operation)
for operation in manifest["operations"]
if operation["method"] in AUTH_METHODS
}
assert {
operation["method"] for operation in manifest["operations"]
} & AUTH_METHODS == AUTH_METHODS
reachable = _reachable_schema_names(schemas, result_components)
for name in reachable:
for schema in _schema_objects(schemas[name]):
properties = schema.get("properties", {})
if isinstance(properties, Mapping):
assert "payload" not in properties, name
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())
strings = set(_structured_strings(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
assert not any("TemporaryDirectory" in value for value in strings)
assert not any("\\Temp\\" in value for value in strings)
assert "127.0.0.1" not in strings
assert "/rpc" not in strings
def test_generated_required_properties_are_declared() -> None:
schemas = generate_manifest()["components"]["schemas"]
for component_name, component in schemas.items():
for schema in _schema_objects(component):
required = schema.get("required")
properties = schema.get("properties")
if isinstance(required, list) and isinstance(properties, Mapping):
assert set(required) <= properties.keys(), component_name
tests/wf_contract_manifest/test_io.py
+10
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import math
from collections.abc import Mapping
from pathlib import Path
from typing import Any
@@ -51,6 +52,15 @@ def test_canonical_json_ignores_recursive_mapping_insertion_order() -> None:
)
@pytest.mark.parametrize("value", [math.nan, math.inf, -math.inf])
def test_canonical_json_rejects_non_finite_numbers(value: float) -> None:
manifest = _manifest()
manifest["components"]["schemas"]["FreeJson"] = {"const": value}
with pytest.raises(ValueError, match="manifest is not canonically serializable"):
canonical_manifest_json(manifest)
def test_write_and_check_round_trip(tmp_path: Path) -> None:
path = tmp_path / "workflow-api.manifest.json"
tests/wf_contract_manifest/test_normalize.py
+60 -14
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import math
from copy import deepcopy
from typing import Any, Protocol
@@ -14,9 +15,7 @@ class DocumentMutation(Protocol):
def __call__(self, document: dict[str, Any]) -> object: ...
def assert_manifest_error(
document: dict[str, Any], path: str, message: str
) -> None:
def assert_manifest_error(document: dict[str, Any], path: str, message: str) -> None:
with pytest.raises(ManifestError) as error_info:
manifest_from_openrpc(document)
@@ -44,9 +43,7 @@ def test_normalizes_operations_and_components_deterministically() -> None:
assert manifest["operations"][0]["result"] == {
"schema": {"$ref": "#/components/schemas/AlphaResult"}
}
assert manifest["operations"][0]["errors"] == [
{"$ref": "#/components/errors/5000"}
]
assert manifest["operations"][0]["errors"] == [{"$ref": "#/components/errors/5000"}]
assert list(manifest["components"]["schemas"]) == [
"AlphaResult",
"FreeJson",
@@ -81,17 +78,34 @@ def test_removes_only_titles_and_preserves_unknown_schema_keywords() -> None:
optional_schema = manifest["operations"][0]["params"][0]["schema"]
assert "title" not in optional_schema
assert optional_schema["x-future-keyword"] == {"value": 1}
assert optional_schema["x-future-keyword"] == {
"title": "removed recursively",
"value": 1,
}
assert manifest["components"]["schemas"]["FreeJson"] == {}
assert manifest["components"]["schemas"]["ZetaResult"]["properties"] == {
"extension": {"additionalProperties": True}
}
def test_preserves_named_title_entries_in_schema_maps() -> None:
alpha = manifest_from_openrpc(synthetic_openrpc_document())["components"][
"schemas"
]["AlphaResult"]
assert alpha["required"] == ["mode", "payload", "title"]
properties = alpha["properties"]
definitions = alpha["$defs"]
assert isinstance(properties, dict)
assert isinstance(definitions, dict)
assert properties["title"] == {"type": "string"}
assert definitions["title"] == {"type": "string"}
def test_preserves_conditional_schema_keywords() -> None:
alpha = manifest_from_openrpc(synthetic_openrpc_document())["components"]["schemas"][
"AlphaResult"
]
alpha = manifest_from_openrpc(synthetic_openrpc_document())["components"][
"schemas"
]["AlphaResult"]
assert alpha["if"] == {"properties": {"mode": {"const": "alpha"}}}
assert alpha["then"] == {"required": ["payload"]}
@@ -240,6 +254,36 @@ def test_rejects_invalid_method_error_schema_shape() -> None:
)
@pytest.mark.parametrize("value", [math.nan, math.inf, -math.inf])
def test_rejects_non_finite_schema_numbers_with_exact_path(value: float) -> None:
document = synthetic_openrpc_document()
document["components"]["schemas"]["AlphaResult"]["properties"]["mode"]["const"] = (
value
)
assert_manifest_error(
document,
"$.components.schemas.AlphaResult.properties.mode.const",
"expected a finite JSON number",
)
@pytest.mark.parametrize(
("namespace", "path"),
[
("schemas", "$.components.schemas"),
("errors", "$.components.errors"),
],
)
def test_rejects_non_string_component_keys_before_sorting(
namespace: str, path: str
) -> None:
document = synthetic_openrpc_document()
document["components"][namespace][1] = {}
assert_manifest_error(document, path, "expected string object keys")
@pytest.mark.parametrize(
("mutate", "path", "message"),
[
@@ -248,7 +292,11 @@ def test_rejects_invalid_method_error_schema_shape() -> None:
"$.openrpc",
"unsupported OpenRPC version '2.0.0'; expected '1.2.6'",
),
(lambda document: document.update({"methods": {}}), "$.methods", "expected an array"),
(
lambda document: document.update({"methods": {}}),
"$.methods",
"expected an array",
),
(
lambda document: document["methods"].append(
deepcopy(document["methods"][0])
@@ -338,9 +386,7 @@ def test_reports_reference_errors_at_original_method_path() -> None:
"workflow.zeta.run",
"workflow.alpha.inspect",
]
document["methods"][0]["errors"][0] = {
"$ref": "#/components/errors/Missing"
}
document["methods"][0]["errors"][0] = {"$ref": "#/components/errors/Missing"}
with pytest.raises(ManifestError) as exc_info:
manifest_from_openrpc(document)