be sure that it saves the new shape

This commit is contained in:
lda
2026-05-20 21:45:52 +07:00 Verified
parent e29e093ad1
commit f5af98fb95
3 changed files with 131 additions and 0 deletions
@@ -8,6 +8,16 @@
**Tech Stack:** Python 3.14, Pydantic v2, pytest, jsonschema, basedpyright, ruff. **Tech Stack:** Python 3.14, Pydantic v2, pytest, jsonschema, basedpyright, ruff.
**Current status as of 2026-05-20:** The implementation has moved past the
original checklist. Typed path values, canonical node bindings, canonical
runtime input/output handling, atomic state patches, canonical validation,
JSON-Schema-native state reducers, schema validation, and authoring canonical
emission are present in the tree. The remaining checklist item is the full
compatibility/regression pass in Task 9. If future code changes touch this area,
prefer adding focused tests to the existing `tests/core/test_*path*`,
`tests/core/test_*mapping*`, and `tests/authoring/test_builder.py` coverage
rather than reimplementing the earlier tasks.
--- ---
## File Structure ## File Structure
+53
View File
@@ -63,6 +63,40 @@ def test_workflow_artifact_can_be_marked_as_wrapper_intent() -> None:
assert dumped["kind"] == "wrapper" assert dumped["kind"] == "wrapper"
def test_workflow_artifact_accepts_legacy_required_capability_map_and_dumps_list() -> (
None
):
artifact = WorkflowArtifact.model_validate(
{
"id": "legacy_capabilities",
"version": 1,
"title": "Legacy Capabilities",
"input_schema": {"type": "object", "properties": {}},
"output_schema": {"type": "object", "properties": {}},
"outcomes": ["done"],
"plan": {"name": "legacy_capabilities", "nodes": [], "edges": []},
"required_capabilities": {
"demo.echo": {
"kind": "tool",
"input_schema_hash": "sha256:input",
"output_schema_hash": "sha256:output",
"observed_concrete_source": "demo.personal",
}
},
}
)
dumped = artifact.model_dump(mode="json")
required = dumped["required_capabilities"][0]
assert isinstance(dumped["required_capabilities"], list)
assert required["ref"] == "demo.echo"
assert required["observed_concrete_source"] == "demo.personal"
assert "logical_source" not in required
assert "capability_name" not in required
assert artifact.required_capability_map()["demo.echo"].logical_source == "demo"
def test_workflow_deployment_binds_logical_sources_to_concrete_sources() -> None: def test_workflow_deployment_binds_logical_sources_to_concrete_sources() -> None:
deployment = WorkflowDeployment( deployment = WorkflowDeployment(
id="summarize_docs.personal", id="summarize_docs.personal",
@@ -85,6 +119,25 @@ def test_workflow_deployment_binds_logical_sources_to_concrete_sources() -> None
assert deployment.binding_map()["context7"] == "context7.personal" assert deployment.binding_map()["context7"] == "context7.personal"
def test_workflow_deployment_accepts_legacy_binding_map_and_dumps_list() -> None:
deployment = WorkflowDeployment.model_validate(
{
"id": "legacy_bindings.personal",
"artifact_id": "legacy_bindings",
"artifact_version": 1,
"bindings": {"demo": "demo.personal"},
}
)
dumped = deployment.model_dump(mode="json")
binding = dumped["bindings"][0]
assert isinstance(dumped["bindings"], list)
assert binding["logical_source"] == "demo"
assert binding["concrete_source"] == "demo.personal"
assert deployment.binding_map()["demo"] == "demo.personal"
def test_dependency_diagnostic_is_structured() -> None: def test_dependency_diagnostic_is_structured() -> None:
diagnostic = DependencyDiagnostic( diagnostic = DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR, severity=DiagnosticSeverity.ERROR,
+68
View File
@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
import json
from wf_artifacts import ( from wf_artifacts import (
FileWorkflowArtifactStore, FileWorkflowArtifactStore,
WorkflowArtifact, WorkflowArtifact,
@@ -63,6 +65,72 @@ def test_file_store_round_trips_deployment(tmp_path) -> None:
assert loaded.binding_map()["context7"] == "context7.personal" assert loaded.binding_map()["context7"] == "context7.personal"
def test_file_store_loads_legacy_artifact_and_rewrites_canonical_shape(
tmp_path,
) -> None:
store = FileWorkflowArtifactStore(tmp_path)
artifact_dir = store.artifacts_dir / "legacy_capabilities"
artifact_dir.mkdir(parents=True)
artifact_path = artifact_dir / "1.json"
artifact_path.write_text(
json.dumps(
{
"id": "legacy_capabilities",
"version": 1,
"title": "Legacy Capabilities",
"input_schema": {"type": "object", "properties": {}},
"output_schema": {"type": "object", "properties": {}},
"outcomes": ["done"],
"plan": {"name": "legacy_capabilities", "nodes": [], "edges": []},
"required_capabilities": {
"demo.echo": {
"kind": "tool",
"input_schema_hash": "sha256:input",
}
},
}
),
encoding="utf-8",
)
loaded = store.get_artifact("legacy_capabilities", 1)
store.save_artifact(loaded)
rewritten = json.loads(artifact_path.read_text(encoding="utf-8"))
required = rewritten["required_capabilities"][0]
assert required["ref"] == "demo.echo"
assert required["kind"] == "tool"
assert "logical_source" not in required
assert "capability_name" not in required
def test_file_store_loads_legacy_deployment_and_rewrites_canonical_shape(
tmp_path,
) -> None:
store = FileWorkflowArtifactStore(tmp_path)
deployment_path = store.deployments_dir / "legacy_bindings.personal.json"
deployment_path.write_text(
json.dumps(
{
"id": "legacy_bindings.personal",
"artifact_id": "legacy_bindings",
"artifact_version": 1,
"bindings": {"demo": "demo.personal"},
}
),
encoding="utf-8",
)
loaded = store.get_deployment("legacy_bindings.personal")
store.save_deployment(loaded)
rewritten = json.loads(deployment_path.read_text(encoding="utf-8"))
binding = rewritten["bindings"][0]
assert binding["logical_source"] == "demo"
assert binding["concrete_source"] == "demo.personal"
assert loaded.binding_map()["demo"] == "demo.personal"
def test_file_store_lists_deployments_in_id_order(tmp_path) -> None: def test_file_store_lists_deployments_in_id_order(tmp_path) -> None:
store = FileWorkflowArtifactStore(tmp_path) store = FileWorkflowArtifactStore(tmp_path)
store.save_deployment( store.save_deployment(