Fix WfFormat 1.6 compatibility

.gitignore

@@ -60,3 +60,6 @@ bin/gpu_benchmark
 coverage.xml
 
 test/
+
+# WfFormat schema (fetched from the WfFormat repository, not versioned here)
+wfcommons-schema.json

pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 ]
 license = {text = "GNU Lesser General Public License v3 (LGPLv3)"}
 dependencies = [
-    "jsonschema",
+    "jsonschema>=4.18",
     "matplotlib",
     "networkx",
     "numpy",

tests/recipes/test_recipes.py

@@ -25,6 +25,7 @@
 from wfcommons.wfchef.recipes import SrasearchRecipe
 
 from wfcommons import WorkflowGenerator
+from wfcommons.wfinstances import SchemaValidator
 
 class TestRecipes:
 
@@ -53,6 +54,10 @@ def test_recipes(self, recipe_class) -> None:
         workflows = WorkflowGenerator(recipe).build_workflows(1)
         assert len(workflows) == 1
 
+        # generated instances (including their metrics) must be valid WfFormat instances
+        workflows[0].generate_json()
+        SchemaValidator().validate_instance(workflows[0].workflow_json)
+
 
 
     @pytest.mark.unit

tests/unit/common/test_task.py

@@ -76,7 +76,7 @@ def test_task_execution(self, task: Task) -> None:
             "writtenBytes": 54321,
             "memoryInBytes": 10,
             "energyInKWh": 100,
-            "avgPowerInKWh": 1.0,
+            "avgPowerInW": 1.0,
             "priority": 100,
             "machines": ["machine_1"],
             "executedAt": "2024-09-15T08:59:33.699321-04:00",

tests/unit/common/test_workflow.py

@@ -17,8 +17,8 @@
 from typing import Any
 
 from datetime import datetime
-from wfcommons.common import Task, Workflow
-from wfcommons.wfinstances import Instance
+from wfcommons.common import File, Task, Workflow
+from wfcommons.wfinstances import Instance, SchemaValidator
 from wfcommons.version import __version__, __schema_version__
 
 
@@ -274,5 +274,98 @@ def test_workflow_dot_file(self):
         assert(num_dependencies == len(instance.workflow.edges))
 
 
+def _build_workflow(tasks: list[tuple[str, str]], dependencies: list[tuple[str, str]], **kwargs) -> Workflow:
+    """Build a workflow from (name, id) pairs and (parent id, child id) pairs."""
+    workflow = Workflow(name="wfformat_test", makespan=10.0, **kwargs)
+    for name, task_id in tasks:
+        workflow.add_task(Task(name=name, task_id=task_id, runtime=1.0,
+                               output_files=[File(f"{task_id}_out", 10)]))
+    for parent, child in dependencies:
+        workflow.add_dependency(parent, child)
+    workflow.generate_json()
+    return workflow
 
 
+class TestWorkflowWfFormat:
+
+    @pytest.mark.unit
+    def test_metrics_with_shared_task_names(self) -> None:
+        # four tasks sharing the same name fan into a fifth one
+        workflow = _build_workflow(
+            [("split", f"split_ID000000{i}") for i in range(1, 5)] + [("merge", "merge_ID0000005")],
+            [(f"split_ID000000{i}", "merge_ID0000005") for i in range(1, 5)])
+        metrics = workflow.workflow_json["workflow"]["specification"]["metrics"]
+        assert metrics["numberOfLevels"] == 2
+        assert metrics["minimumWidth"] == 1
+        assert metrics["maximumWidth"] == 4
+        SchemaValidator().validate_instance(workflow.workflow_json)
+
+    @pytest.mark.unit
+    def test_metrics_of_a_chain_of_same_name_tasks(self) -> None:
+        ids = [f"step_ID000000{i}" for i in range(1, 5)]
+        workflow = _build_workflow([("step", task_id) for task_id in ids], list(zip(ids, ids[1:])))
+        metrics = workflow.workflow_json["workflow"]["specification"]["metrics"]
+        assert (metrics["numberOfLevels"], metrics["minimumWidth"], metrics["maximumWidth"]) == (4, 1, 1)
+        SchemaValidator().validate_instance(workflow.workflow_json)
+
+    @pytest.mark.unit
+    @pytest.mark.parametrize(
+        "task_ids",
+        [
+            ["ID0000001", "ID0000002"],  # as in the WfFormat documentation
+            ["ingest:raw", "proc/a"],  # ':' and '/' are allowed from WfFormat 1.6
+            ["main.cwl#step-1", "main.cwl#step-2"],
+            ["stage_in_local_local_0_0", "get_IDs_ID0000003"],
+        ]
+    )
+    def test_task_ids_allowed_by_wfformat(self, task_ids: list[str]) -> None:
+        workflow = _build_workflow([("task", task_id) for task_id in task_ids], [(task_ids[0], task_ids[1])])
+        SchemaValidator().validate_instance(workflow.workflow_json)
+
+    @pytest.mark.unit
+    def test_byte_metrics_are_integers(self) -> None:
+        workflow = Workflow(name="bytes_test")
+        workflow.add_task(Task(name="a", task_id="a", runtime=0, bytes_read=2, bytes_written=3))
+        workflow.add_task(Task(name="b", task_id="b", runtime=0, bytes_read=4, bytes_written=0.5))
+        workflow.generate_json()
+        metrics = workflow.workflow_json["workflow"]["execution"]["metrics"]
+        # totals do not depend on runtimes, and fractional totals are omitted
+        assert metrics == {"totalNumBytesRead": 6}
+        assert isinstance(metrics["totalNumBytesRead"], int)
+        SchemaValidator().validate_instance(workflow.workflow_json)
+
+    @pytest.mark.unit
+    def test_execution_fields(self) -> None:
+        task = Task(name="a", task_id="a", runtime=1.0, cores=0, args=["-x", "", 3],
+                    executed_at="2025-03-01 10:00:01.000000")
+        execution = task.execution_as_dict()
+        assert "coreCount" not in execution
+        assert execution["command"] == {"arguments": ["-x", "3"]}
+        assert datetime.fromisoformat(execution["executedAt"]).tzinfo is not None
+        assert "command" not in Task(name="b", task_id="b", runtime=1.0).execution_as_dict()
+
+    @pytest.mark.unit
+    def test_specification_only_workflow(self) -> None:
+        workflow = _build_workflow([("a", "a"), ("b", "b")], [("a", "b")], include_execution=False)
+        assert "execution" not in workflow.workflow_json["workflow"]
+        SchemaValidator().validate_instance(workflow.workflow_json)
+
+    @pytest.mark.unit
+    def test_instance_properties(self) -> None:
+        workflow = _build_workflow([("a", "a")], [], author_name="Jane Doe", author_email="jane@example.org",
+                                   author_institution="Example University", author_country="US",
+                                   created_at="2026-08-04T15:19:28-08:00", executed_at="2026-08-04T15:20:00-08:00")
+        assert workflow.workflow_json["author"] == {"name": "Jane Doe", "email": "jane@example.org",
+                                                   "institution": "Example University", "country": "US"}
+        assert workflow.workflow_json["createdAt"] == "2026-08-04T15:19:28-08:00"
+        assert workflow.workflow_json["workflow"]["execution"]["executedAt"] == "2026-08-04T15:20:00-08:00"
+
+    @pytest.mark.unit
+    def test_duplicate_file_ids(self) -> None:
+        workflow = Workflow(name="files_test")
+        workflow.add_task(Task(name="a", task_id="a", runtime=1.0, output_files=[File("f.txt", 10)]))
+        workflow.add_task(Task(name="b", task_id="b", runtime=1.0, input_files=[File("f.txt", 20)]))
+        workflow.add_dependency("a", "b")
+        workflow.generate_json()
+        assert workflow.workflow_json["workflow"]["specification"]["files"] == [{"id": "f.txt", "sizeInBytes": 10}]
+        SchemaValidator().validate_instance(workflow.workflow_json)

tests/unit/wfinstances/test_instance_analyzer.py

@@ -34,7 +34,7 @@ def instance(self) -> pathlib.Path:
                         {
                             "name": "task_01",
                             "id": "task_01",
-                            "children": ["task_02", "task_03"],
+                            "children": [],
                             "inputFiles": ["input_01.txt", "input_02.txt"],
                             "outputFiles": ["output_01.txt"],
                             "parents": []

tests/unit/wfinstances/test_instances.py

@@ -12,7 +12,9 @@
 import pathlib
 import pytest
 
-from wfcommons.wfinstances import Instance
+from wfcommons.version import __schema_version__
+from wfcommons.wfinstances import Instance, SchemaValidator
+from wfcommons.wfinstances import schema as schema_module
 
 
 class TestInstance:
@@ -34,7 +36,7 @@ def instance(self) -> pathlib.Path:
                         {
                             "name": "task_01",
                             "id": "task_01",
-                            "children": ["task_02", "task_03"],
+                            "children": [],
                             "inputFiles": ["input_01.txt", "input_02.txt"],
                             "outputFiles": ["output_01.txt"],
                             "parents": []
@@ -79,3 +81,158 @@ def instance(self) -> pathlib.Path:
     def test_instance_creation(self, instance : pathlib.Path) -> None:
         instance = Instance(instance)
         assert(instance.leaves() == instance.roots())
+
+
+def _wfformat_example() -> dict:
+    """The example instance of the WfFormat 1.6 documentation."""
+    return {
+        "name": "my-workflow-instance",
+        "description": "A two-task workflow executed on a single node",
+        "createdAt": "2026-08-04T15:19:28-08:00",
+        "schemaVersion": "1.6",
+        "author": {"name": "Jane Doe", "email": "jane.doe@example.org",
+                   "institution": "Example University", "country": "US"},
+        "runtimeSystem": {"name": "pegasus", "version": "5.0.6", "url": "https://pegasus.isi.edu"},
+        "workflow": {
+            "specification": {
+                "tasks": [
+                    {"name": "split", "id": "ID0000001", "parents": [], "children": ["ID0000002"],
+                     "inputFiles": ["input.txt"], "outputFiles": ["chunk.txt"]},
+                    {"name": "analyze", "id": "ID0000002", "parents": ["ID0000001"], "children": [],
+                     "inputFiles": ["chunk.txt"], "outputFiles": ["result.txt"]}
+                ],
+                "files": [
+                    {"id": "input.txt", "sizeInBytes": 1024}, {"id": "chunk.txt", "sizeInBytes": 512},
+                    {"id": "result.txt", "sizeInBytes": 128}
+                ],
+                "metrics": {"numberOfTasks": 2, "numberOfFiles": 3, "sumOfFileSizesInBytes": 1664,
+                            "numberOfLevels": 2, "minimumWidth": 1, "maximumWidth": 1}
+            },
+            "execution": {
+                "makespanInSeconds": 14.2,
+                "executedAt": "2026-08-04T15:20:00-08:00",
+                "tasks": [
+                    {"id": "ID0000001", "runtimeInSeconds": 4.2, "executedAt": "2026-08-04T15:20:00-08:00",
+                     "command": {"program": "split", "arguments": ["-n", "2", "input.txt"]},
+                     "coreCount": 1, "avgCPU": 93.78, "readBytes": 1024, "writtenBytes": 512,
+                     "memoryInBytes": 4194304, "energyInKWh": 0.25, "avgPowerInW": 120.0,
+                     "machines": ["node01.example.org"]},
+                    {"id": "ID0000002", "runtimeInSeconds": 10.0, "executedAt": "2026-08-04T15:20:04-08:00",
+                     "coreCount": 4, "readBytes": 512, "writtenBytes": 128, "machines": ["node01.example.org"]}
+                ],
+                "machines": [
+                    {"system": "linux", "architecture": "x86_64", "nodeName": "node01.example.org",
+                     "release": "5.15.0", "memoryInBytes": 34359738368,
+                     "cpu": {"coreCount": 8, "speedInMHz": 3600, "vendor": "AMD"}}
+                ],
+                "metrics": {"sumTaskRuntimesInSeconds": 14.2, "totalNumBytesRead": 1536,
+                            "totalNumBytesWritten": 640}
+            }
+        }
+    }
+
+
+def _load(instance: dict, tmp_path: pathlib.Path) -> Instance:
+    instance_path = tmp_path / "instance.json"
+    instance_path.write_text(json.dumps(instance))
+    return Instance(instance_path)
+
+
+class TestInstanceWfFormat:
+
+    @pytest.mark.unit
+    def test_round_trip(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        output_path = tmp_path / "written.json"
+        _load(original, tmp_path).workflow.write_json(output_path)
+        written = json.loads(output_path.read_text())
+        assert written == original
+        SchemaValidator().validate_instance(written)
+
+    @pytest.mark.unit
+    def test_specification_only(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        del original["workflow"]["execution"]
+        instance = _load(original, tmp_path)
+        assert len(instance.workflow.tasks) == 2
+        instance.workflow.generate_json()
+        assert "execution" not in instance.workflow.workflow_json["workflow"]
+        SchemaValidator().validate_instance(instance.workflow.workflow_json)
+
+    @pytest.mark.unit
+    def test_minimal_instance(self, tmp_path: pathlib.Path) -> None:
+        # only name, schemaVersion, and workflow are required, and task IDs may hold ':' and '/'
+        minimal = {
+            "name": "minimal",
+            "schemaVersion": "1.6",
+            "workflow": {
+                "specification": {"tasks": [
+                    {"name": "t", "id": "t:1", "parents": [], "children": ["t/2"]},
+                    {"name": "t", "id": "t/2", "parents": ["t:1"], "children": []}
+                ]},
+                "execution": {"makespanInSeconds": 1.0, "executedAt": "2026-08-04T15:20:00-08:00",
+                              "tasks": [{"id": "t:1", "runtimeInSeconds": 0.5}]}
+            }
+        }
+        instance = _load(minimal, tmp_path)
+        assert instance.roots() == ["t:1"] and instance.leaves() == ["t/2"]
+        instance.workflow.generate_json()
+        written = instance.workflow.workflow_json
+        SchemaValidator().validate_instance(written)
+        # optional properties that the instance does not declare are not made up
+        assert not {"description", "createdAt", "author", "runtimeSystem"} & written.keys()
+        assert "executedAt" not in written["workflow"]["execution"]["tasks"][1]
+
+    @pytest.mark.unit
+    def test_unused_machine(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        original["workflow"]["execution"]["machines"].append({"nodeName": "node02.example.org"})
+        instance = _load(original, tmp_path)
+        instance.workflow.generate_json()
+        machines = instance.workflow.workflow_json["workflow"]["execution"]["machines"]
+        assert machines == original["workflow"]["execution"]["machines"]
+
+    @pytest.mark.unit
+    def test_machine_without_cpu(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        del original["workflow"]["execution"]["machines"][0]["cpu"]
+        instance = _load(original, tmp_path)
+        assert instance.machines["node01.example.org"].cpu_cores is None
+
+    @pytest.mark.unit
+    def test_integral_float_file_size(self, tmp_path: pathlib.Path) -> None:
+        # JSON Schema draft 2020-12 considers 1024.0 to be an integer
+        original = _wfformat_example()
+        original["workflow"]["specification"]["files"][0]["sizeInBytes"] = 1024.0
+        _load(original, tmp_path)
+
+    @pytest.mark.unit
+    def test_invalid_metrics(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        original["workflow"]["specification"]["metrics"]["maximumWidth"] = 2
+        with pytest.raises(RuntimeError, match="semantic"):
+            _load(original, tmp_path)
+
+    @pytest.mark.unit
+    def test_previous_schema_version(self, tmp_path: pathlib.Path) -> None:
+        original = _wfformat_example()
+        original["schemaVersion"] = "1.5"
+        with pytest.raises(RuntimeError, match="syntax"):
+            _load(original, tmp_path)
+
+    @pytest.mark.unit
+    def test_schema_of_wfformat_release(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        # a schema file in the working directory (e.g., left by an older WfCommons) is ignored
+        (tmp_path / "wfcommons-schema.json").write_text(json.dumps({"type": "object", "required": ["bogus"]}))
+        monkeypatch.chdir(tmp_path)
+        # a cached schema that does not accept the current schema version is replaced
+        monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache"))
+        monkeypatch.setattr(schema_module, "_schema_cache", {})
+        cache_file = tmp_path / "cache" / "wfcommons" / f"wfformat-v{__schema_version__}-wfcommons-schema.json"
+        cache_file.parent.mkdir(parents=True)
+        cache_file.write_text(json.dumps({"properties": {"schemaVersion": {"enum": ["1.5"]}}}))
+
+        validator = SchemaValidator()
+        assert validator.schema["properties"]["schemaVersion"]["enum"] == [__schema_version__]
+        assert json.loads(cache_file.read_text()) == validator.schema
+        validator.validate_instance(_wfformat_example())