fix: harden workflow contract tooling
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
from .generate import generate_manifest
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .io import (
|
||||
DEFAULT_MANIFEST_PATH,
|
||||
ManifestDriftError,
|
||||
@@ -9,6 +10,18 @@ from .io import (
|
||||
from .model import ContractManifest, JsonSchema, JsonValue, ManifestError
|
||||
from .normalize import manifest_from_openrpc
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .generate import generate_manifest
|
||||
|
||||
|
||||
def __getattr__(name: str) -> object:
|
||||
"""Load the server-backed generator only when a caller requests it."""
|
||||
if name == "generate_manifest":
|
||||
from .generate import generate_manifest
|
||||
|
||||
return generate_manifest
|
||||
raise AttributeError(name)
|
||||
|
||||
__all__ = [
|
||||
"ContractManifest",
|
||||
"JsonSchema",
|
||||
|
||||
@@ -6,6 +6,7 @@ from pathlib import Path
|
||||
from .model import ContractManifest
|
||||
|
||||
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
||||
REPOSITORY_MARKER = REPOSITORY_ROOT / "pyproject.toml"
|
||||
DEFAULT_MANIFEST_PATH = REPOSITORY_ROOT / "contracts" / "workflow-api.manifest.json"
|
||||
|
||||
|
||||
@@ -34,6 +35,11 @@ def canonical_manifest_json(manifest: ContractManifest) -> str:
|
||||
def write_manifest(
|
||||
manifest: ContractManifest, path: Path = DEFAULT_MANIFEST_PATH
|
||||
) -> Path:
|
||||
if path == DEFAULT_MANIFEST_PATH and not REPOSITORY_MARKER.is_file():
|
||||
raise RuntimeError(
|
||||
"cannot write the default manifest outside the repository checkout: "
|
||||
f"missing {REPOSITORY_MARKER}"
|
||||
)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(canonical_manifest_json(manifest), encoding="utf-8", newline="\n")
|
||||
return path
|
||||
|
||||
@@ -251,11 +251,12 @@ def manifest_from_openrpc(document: Mapping[str, object]) -> ContractManifest:
|
||||
):
|
||||
parameter_path = f"{method_path}.params[{parameter_index}]"
|
||||
parameter = _mapping(parameter_value, parameter_path)
|
||||
raw_required = parameter.get("required", False)
|
||||
params.append(
|
||||
{
|
||||
"name": _string(parameter.get("name"), f"{parameter_path}.name"),
|
||||
"required": _boolean(
|
||||
parameter.get("required"), f"{parameter_path}.required"
|
||||
raw_required, f"{parameter_path}.required"
|
||||
),
|
||||
"schema": _schema(
|
||||
parameter.get("schema"), f"{parameter_path}.schema"
|
||||
@@ -266,9 +267,11 @@ def manifest_from_openrpc(document: Mapping[str, object]) -> ContractManifest:
|
||||
result_path = f"{method_path}.result"
|
||||
result = _mapping(method.get("result"), result_path)
|
||||
if "schema" not in result:
|
||||
raise ManifestError(f"{result_path}.schema", "expected an object")
|
||||
raise ManifestError(
|
||||
f"{result_path}.schema", "missing success result schema"
|
||||
)
|
||||
raw_result_schema = result["schema"]
|
||||
if isinstance(raw_result_schema, Mapping) and (
|
||||
if not isinstance(raw_result_schema, Mapping) or (
|
||||
set(raw_result_schema) != {"$ref"}
|
||||
or not (
|
||||
isinstance(raw_result_schema.get("$ref"), str)
|
||||
|
||||
Reference in New Issue
Block a user