From 4d18f3373b835fe204775bfa35e0f0b07a47cc16 Mon Sep 17 00:00:00 2001 From: Ryunosuke O'Neil Date: Fri, 23 Oct 2026 16:18:15 +0200 Subject: [PATCH] feat(core): canonical identity of CWL processes A process is identified by the SHA-256 of a canonical form defined over the parsed JSON document, not over any CWL library's output, so that it is deterministic, stable across library upgrades and reproducible by any tool. Ids, doc and label where they are metadata, $namespaces and dirac: hints are dropped; prefixes are expanded to full URIs; lists become maps; fully scoped identifiers are shortened against their scope; type shorthands are expanded and named types written by short name. Each child run body is replaced by its own identity, so a workflow's identity covers its children's (a Merkle tree). pack and unpack convert to and from a self-contained $graph whose components are named by their identity. Documents must be resolved before canonicalisation: $import, $include and any other schema-salad directive is refused, since the identity would not cover what it names and a runner would fetch it. Standard library only. Closes #1106 Part of #1080 --- diracx-core/src/diracx/core/cwl/__init__.py | 25 + diracx-core/src/diracx/core/cwl/canonical.py | 650 +++++++++++++++++++ diracx-core/tests/test_cwl_canonical.py | 442 +++++++++++++ 3 files changed, 1117 insertions(+) create mode 100644 diracx-core/src/diracx/core/cwl/__init__.py create mode 100644 diracx-core/src/diracx/core/cwl/canonical.py create mode 100644 diracx-core/tests/test_cwl_canonical.py diff --git a/diracx-core/src/diracx/core/cwl/__init__.py b/diracx-core/src/diracx/core/cwl/__init__.py new file mode 100644 index 000000000..adcf1b43d --- /dev/null +++ b/diracx-core/src/diracx/core/cwl/__init__.py @@ -0,0 +1,25 @@ +"""Common Workflow Language utilities shared by the client and the services.""" + +from __future__ import annotations + +__all__ = [ + "CANONICAL_VERSION", + "CanonicalProcess", + "canonical_bytes", + "decompose", + "format_reference", + "pack", + "parse_reference", + "unpack", +] + +from .canonical import ( + CANONICAL_VERSION, + CanonicalProcess, + canonical_bytes, + decompose, + format_reference, + pack, + parse_reference, + unpack, +) diff --git a/diracx-core/src/diracx/core/cwl/canonical.py b/diracx-core/src/diracx/core/cwl/canonical.py new file mode 100644 index 000000000..9941209d2 --- /dev/null +++ b/diracx-core/src/diracx/core/cwl/canonical.py @@ -0,0 +1,650 @@ +"""Canonical form and content identity of CWL processes (DX-ADR-004, DX-ADR-007). + +A process is identified by the SHA-256 of its canonical byte form. The form is +defined over the parsed document (a plain JSON value), not over the output of +any CWL library, so that it is deterministic, stable across library upgrades +and reproducible by any tool. Version 1 of the rules: + +1. Only CWL v1.2 is accepted. +2. On every process, ``id``, ``$namespaces`` and ``$schemas`` are dropped: the + identity of a process is its content, not its name or where it was loaded + from. Before that, every key, ``class`` and ``format`` written with a bound + prefix (``edam:format_2572``) is expanded to its full URI, using the + bindings of the process and of the processes enclosing it, so the identity + covers what a prefix means and not which prefix was chosen. The ``dirac`` + prefix is never expanded. +3. ``doc`` and ``label`` are dropped from processes, parameters, steps and + schemas, but kept as names a user chose (a record field, an environment + variable), inside ``default`` values and inside requirements and hints + other than ``SchemaDefRequirement``, where they are data. ``dirac:`` fields, + ``dirac:`` hints and anonymous schema names (``name: _:``) are dropped. + ``dirac:`` data is extracted into transformation columns at submission, so + it is not part of what a parcel runs. +4. ``inputs``, ``outputs``, ``steps`` and a step's ``in`` are written in map + form keyed by their short id; ``requirements`` and ``hints`` in map form + keyed by class. Shorthand values (``x: File``, ``in: {x: source}``) are + expanded to objects. A step's ``out`` is a sorted list of short ids. +5. A fully scoped identifier or reference (``file:///p/x.cwl#main/x``, + ``#main/x``) is shortened to the part after ``#``, without the enclosing + scope, so that absolute ``file://`` URIs never enter the identity. A + relative one is already relative to that scope, as a CWL runner reads it, + and is kept as written. +6. A child ``run`` body is replaced by a reference to its own identity, so the + identity of a workflow covers those of its children (a Merkle tree). Child + bodies are therefore canonicalised before their parents. +7. Type shorthands are expanded, as a CWL runner reads them: ``File[]`` is + ``{type: array, items: File}`` and ``File?`` is ``[null, File]``. A named + type (``SchemaDefRequirement``) and every reference to it are written by + its short name, as are enum symbols and record field names. A + process may only use named types it defines itself, so that its identity + covers their definitions; :func:`pack` scopes them by component + (``#main/Seed``), as a CWL runner requires. +8. The result is serialised as JSON with sorted keys and no whitespace, then + hashed. Numbers use Python's ``json`` rendering, which matches RFC 8785 + for the integers and short decimals CWL documents contain in practice. + +Documents must be resolved before canonicalisation: a schema-salad directive +left in one (``$import``, ``$include``, ``$mixin``, or any other ``$`` key +except a process's ``$namespaces`` and ``$schemas``) is refused, since the +identity would not cover what it refers to and a CWL runner would fetch it. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +from dataclasses import dataclass +from typing import Any + +CANONICAL_VERSION = 1 + +REFERENCE_PREFIX = "proc:sha256-" +PACKED_PREFIX = "#sha256-" + +PROCESS_CLASSES = frozenset( + {"CommandLineTool", "Workflow", "ExpressionTool", "Operation"} +) +PRIMITIVE_TYPES = frozenset( + { + "null", + "boolean", + "int", + "long", + "float", + "double", + "string", + "File", + "Directory", + "Any", + "stdin", + "stdout", + "stderr", + } +) +PROCESS_ONLY_FIELDS = frozenset({"id", "$namespaces", "$schemas"}) +METADATA_FIELDS = frozenset({"doc", "label"}) +DIRAC_PREFIX = "dirac:" +# Fields whose value is a reference to a parameter or step output +REFERENCE_FIELDS = frozenset({"source", "outputSource"}) + + +@dataclass(frozen=True) +class CanonicalProcess: + """One process of a canonicalised tree. + + Attributes: + process_id: SHA-256 of ``body``. + body: Canonical bytes, child ``run`` bodies replaced by references. + children: Identities of the processes ``body`` references, sorted. + """ + + process_id: bytes + body: bytes + children: tuple[bytes, ...] + + +def canonical_bytes(value: Any) -> bytes: + """Serialise a JSON value in canonical form (rule 7). + + Args: + value: Plain JSON value. + + Returns: + UTF-8 JSON with sorted keys and no insignificant whitespace. + """ + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode() + + +def format_reference(process_id: bytes) -> str: + """Return the ``run`` reference to a stored process. + + Args: + process_id: Identity of the process. + + Returns: + The reference, e.g. ``proc:sha256-``. + """ + return REFERENCE_PREFIX + process_id.hex() + + +def parse_reference(reference: str) -> bytes: + """Return the identity a ``run`` reference points to. + + Args: + reference: A reference made by :func:`format_reference`. + + Returns: + The 32-byte identity. + + Raises: + ValueError: If ``reference`` is not a process reference. + """ + if not reference.startswith(REFERENCE_PREFIX): + raise ValueError(f"Not a process reference: {reference!r}") + process_id = bytes.fromhex(reference.removeprefix(REFERENCE_PREFIX)) + if len(process_id) != 32: + raise ValueError(f"Not a SHA-256 process reference: {reference!r}") + return process_id + + +def decompose(process: dict[str, Any]) -> tuple[bytes, dict[bytes, CanonicalProcess]]: + """Canonicalise a process and every process nested in it. + + Args: + process: A CWL process as a plain JSON value, with every ``run`` either + an inline body or a process reference. + + Returns: + The identity of ``process`` and every distinct process of the tree, + keyed by identity. + + Raises: + ValueError: If the document is not a CWL v1.2 process, or a ``run`` + path or schema-salad directive has not been resolved. + """ + _check_resolved(process) + nodes: dict[bytes, CanonicalProcess] = {} + root = _decompose(process, nodes, process.get("cwlVersion"), {}) + return root, nodes + + +def pack(root: bytes, nodes: dict[bytes, CanonicalProcess]) -> dict[str, Any]: + """Assemble a tree into one self-contained packed document. + + The root is ``#main`` and every other process is ``#sha256-``, so the + document runs with a stock CWL runner and still names each component by + its identity. + + Args: + root: Identity of the root process. + nodes: Processes of the tree, at least every one reachable from root. + + Returns: + A CWL document with a ``$graph``. + """ + graph = [] + for process_id in _reachable(root, nodes): + body = json.loads(nodes[process_id].body) + body.pop("cwlVersion", None) + body["id"] = "#main" if process_id == root else PACKED_PREFIX + process_id.hex() + _scope_types(body, body["id"]) + for step in (body.get("steps") or {}).values(): + if isinstance(step.get("run"), str): + step["run"] = PACKED_PREFIX + parse_reference(step["run"]).hex() + graph.append(body) + return {"cwlVersion": "v1.2", "$graph": graph} + + +def unpack(packed: dict[str, Any]) -> tuple[bytes, dict[bytes, CanonicalProcess]]: + """Recover the canonical tree from a document made by :func:`pack`. + + Args: + packed: The packed document. + + Returns: + The identity of the root and the processes of the tree. + + Raises: + ValueError: If the document was not made by :func:`pack` or a + component does not match the identity it is filed under. + """ + by_id = {entry["id"]: entry for entry in packed["$graph"]} + nodes: dict[bytes, CanonicalProcess] = {} + # Each entry is verified once, however many steps share it + resolved: dict[str, bytes] = {} + in_progress: set[str] = set() + + def resolve(entry_id: str) -> bytes: + if entry_id in resolved: + return resolved[entry_id] + if entry_id in in_progress: + raise ValueError(f"{entry_id} runs itself") + in_progress.add(entry_id) + body = copy.deepcopy(by_id[entry_id]) + body.pop("id") + _unscope_types(body, entry_id) + body["cwlVersion"] = packed["cwlVersion"] + children = set() + steps = {} + for name, step in (body.get("steps") or {}).items(): + step = dict(step) + if isinstance(step.get("run"), str): + child = resolve(step["run"]) + children.add(child) + step["run"] = format_reference(child) + steps[name] = step + if steps: + body["steps"] = steps + data = canonical_bytes(body) + process_id = hashlib.sha256(data).digest() + if entry_id != "#main" and entry_id != PACKED_PREFIX + process_id.hex(): + raise ValueError(f"{entry_id} does not match its content") + nodes[process_id] = CanonicalProcess(process_id, data, tuple(sorted(children))) + in_progress.discard(entry_id) + resolved[entry_id] = process_id + return process_id + + return resolve("#main"), nodes + + +def _reachable(root: bytes, nodes: dict[bytes, CanonicalProcess]) -> list[bytes]: + order, seen, stack = [], set(), [root] + while stack: + process_id = stack.pop() + if process_id in seen: + continue + seen.add(process_id) + order.append(process_id) + stack.extend(reversed(nodes[process_id].children)) + return order + + +def _check_resolved(value: Any) -> None: + """Refuse schema-salad directives, which would make a runner fetch files.""" + if isinstance(value, dict): + is_process = value.get("class") in PROCESS_CLASSES + for key, item in value.items(): + if str(key).startswith("$") and not ( + is_process and key in PROCESS_ONLY_FIELDS + ): + raise ValueError( + f"Unresolved {key} directive: resolve the document before " + "submitting it" + ) + _check_resolved(item) + elif isinstance(value, list): + for item in value: + _check_resolved(item) + + +def _decompose( + process: dict[str, Any], + nodes: dict[bytes, CanonicalProcess], + cwl_version: Any, + namespaces: dict[str, str], +) -> bytes: + if process.get("class") not in PROCESS_CLASSES: + raise ValueError(f"Not a CWL process: class={process.get('class')!r}") + version = process.get("cwlVersion", cwl_version) + if version != "v1.2": + raise ValueError( + f"Only CWL v1.2 is supported, got {version!r}: upgrade the document " + "with cwl-upgrader" + ) + + namespaces = {**namespaces, **(process.get("$namespaces") or {})} + namespaces.pop(DIRAC_PREFIX.rstrip(":"), None) + process = _qualify(process, namespaces, top=True) + + own_name = _short(str(process.get("id", "")), None) or None + body: dict[str, Any] = {} + children: set[bytes] = set() + for key, value in process.items(): + if key in PROCESS_ONLY_FIELDS or _is_dropped(key): + continue + if key in ("inputs", "outputs"): + body[key] = _parameters(value, own_name) + elif key in ("requirements", "hints"): + classes = _by_class(value, drop_dirac=key == "hints") + if classes: + body[key] = classes + elif key == "steps": + body[key] = { + name: _step( + step, + own_name, + f"{own_name}/{name}" if own_name else name, + nodes, + version, + namespaces, + children, + ) + for name, step in _by_id(value, own_name).items() + } + else: + body[key] = _strip(value) + _name_types(body, own_name) + body["cwlVersion"] = version + + data = canonical_bytes(body) + process_id = hashlib.sha256(data).digest() + nodes[process_id] = CanonicalProcess(process_id, data, tuple(sorted(children))) + return process_id + + +def _step( + step: dict[str, Any], + own_name: str | None, + step_scope: str, + nodes: dict[bytes, CanonicalProcess], + version: Any, + namespaces: dict[str, str], + children: set[bytes], +) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, value in step.items(): + if _is_dropped(key): + continue + if key == "run": + if isinstance(value, dict): + child = _decompose(value, nodes, version, namespaces) + elif isinstance(value, str) and value.startswith(REFERENCE_PREFIX): + child = parse_reference(value) + else: + raise ValueError(f"Unresolved run reference: {value!r}") + children.add(child) + out[key] = format_reference(child) + elif key == "in": + out[key] = { + name: _strip(_refs(spec, own_name)) + for name, spec in _by_id(value, step_scope, shorthand="source").items() + } + elif key == "out": + out[key] = sorted( + _short(item if isinstance(item, str) else item["id"], step_scope) + for item in value + ) + elif key in ("requirements", "hints"): + classes = _by_class(value, drop_dirac=key == "hints") + if classes: + out[key] = classes + else: + out[key] = _strip( + _refs(value, own_name) if key in REFERENCE_FIELDS else value + ) + return out + + +def _type_definitions(body: dict[str, Any]) -> list[dict[str, Any]]: + schema = (body.get("requirements") or {}).get("SchemaDefRequirement") or {} + return [d for d in schema.get("types") or [] if isinstance(d, dict)] + + +def _map_types(body: dict[str, Any], rename) -> None: + """Apply ``rename`` to every type name a process's types refer to.""" + schema = (body.get("requirements") or {}).get("SchemaDefRequirement") + if schema and schema.get("types"): + schema["types"] = [_map_type(d, rename) for d in schema["types"]] + for side in ("inputs", "outputs"): + for spec in (body.get(side) or {}).values(): + if isinstance(spec, dict) and "type" in spec: + spec["type"] = _map_type(spec["type"], rename) + + +def _map_type(value: Any, rename) -> Any: + """Expand shorthands and rename type names in a type or record field.""" + if isinstance(value, str): + return _expand_type(value, rename) + if isinstance(value, list): + union: list[Any] = [] + for member in value: + mapped = _map_type(member, rename) + # File? in a union is null plus File: a union cannot nest + for item in ( + mapped + if isinstance(member, str) and isinstance(mapped, list) + else [mapped] + ): + if item not in union: + union.append(item) + return union + if not isinstance(value, dict): + return value + value = dict(value) + if value.get("type") not in ("record", "enum", "array") and "type" in value: + value["type"] = _map_type(value["type"], rename) + if "items" in value: + value["items"] = _map_type(value["items"], rename) + fields = value.get("fields") + if isinstance(fields, dict): + value["fields"] = {n: _map_type(f, rename) for n, f in fields.items()} + elif isinstance(fields, list): + value["fields"] = [ + {**_map_type(f, rename), "name": _leaf(f["name"])} + if isinstance(f, dict) and "name" in f + else _map_type(f, rename) + for f in fields + ] + if isinstance(value.get("symbols"), list): + value["symbols"] = [_leaf(s) for s in value["symbols"]] + return value + + +def _expand_type(value: str, rename) -> Any: + """``X?`` is ``[null, X]`` and ``X[]`` is ``{type: array, items: X}``.""" + if value.endswith("?"): + return ["null", _expand_type(value.removesuffix("?"), rename)] + if value.endswith("[]"): + return { + "type": "array", + "items": _expand_type(value.removesuffix("[]"), rename), + } + return rename(value) + + +def _leaf(name: Any) -> Any: + """``red`` for an enum symbol or field written ``#main/Colour/red``.""" + if isinstance(name, str) and "#" in name: + return name.split("#", 1)[1].rsplit("/", 1)[-1] + return name + + +def _name_types(body: dict[str, Any], own_name: str | None) -> None: + """Write named types and references to them by short name (rule 7).""" + definitions = _type_definitions(body) + names = set() + for definition in definitions: + definition["name"] = _short(str(definition["name"]), own_name) + names.add(definition["name"]) + + def rename(name: str) -> str: + if name in PRIMITIVE_TYPES: + return name + short = _short(name, own_name) + if short not in names: + raise ValueError( + f"Type {name!r} is not defined by the process using it: define " + "it in that process's own SchemaDefRequirement" + ) + return short + + _map_types(body, rename) + + +def _scope_types(body: dict[str, Any], entry_id: str) -> None: + """Scope a packed component's named types by its id, as runners require.""" + definitions = _type_definitions(body) + names = {definition["name"] for definition in definitions} + if not names: + return + for definition in definitions: + definition["name"] = f"{entry_id}/{definition['name']}" + _map_types(body, lambda n: f"{entry_id}/{n}" if n in names else n) + + +def _unscope_types(body: dict[str, Any], entry_id: str) -> None: + prefix = f"{entry_id}/" + for definition in _type_definitions(body): + definition["name"] = str(definition["name"]).removeprefix(prefix) + _map_types(body, lambda n: n.removeprefix(prefix)) + + +def _qualify(value: Any, namespaces: dict[str, str], *, top: bool = False) -> Any: + """Expand prefixed keys, classes and formats to full URIs (rule 2). + + Nested processes are left alone: they are expanded with their own + bindings, which may rebind a prefix, when they are canonicalised. + """ + if not namespaces: + return value + if isinstance(value, list): + return [_qualify(v, namespaces) for v in value] + if not isinstance(value, dict): + return value + if not top and value.get("class") in PROCESS_CLASSES: + return value + result = {} + for key, item in value.items(): + if key in ("default", "$namespaces"): + result[key] = item + elif key in ("class", "format"): + result[key] = ( + [_expand_name(v, namespaces) for v in item] + if isinstance(item, list) + else _expand_name(item, namespaces) + ) + else: + result[_expand_name(key, namespaces)] = _qualify(item, namespaces) + return result + + +def _expand_name(name: Any, namespaces: dict[str, str]) -> Any: + if not isinstance(name, str): + return name + prefix, sep, rest = name.partition(":") + if sep and prefix in namespaces and not rest.startswith("//"): + return namespaces[prefix] + rest + return name + + +def _parameters(value: Any, own_name: str | None) -> dict[str, Any]: + return { + name: _strip(_refs(spec, own_name)) + for name, spec in _by_id(value, own_name, shorthand="type").items() + } + + +def _by_id( + value: Any, own_name: str | None, shorthand: str | None = None +) -> dict[str, Any]: + """Map form of an id-keyed list (rule 4), with shorthand values expanded.""" + if isinstance(value, dict): + items = [{"id": k, **_expand(v, shorthand)} for k, v in value.items()] + else: + items = [{"id": v} if isinstance(v, str) else v for v in value] + result = {} + for item in items: + item = dict(item) + name = _short(str(item.pop("id")), own_name) + if name in result: + raise ValueError(f"Duplicate id {name!r}") + result[name] = item + return result + + +def _expand(value: Any, shorthand: str | None) -> dict[str, Any]: + # A map-form value that is an object is the parameter (or step input) itself + if isinstance(value, dict) or shorthand is None: + return value + return {shorthand: value} + + +def _by_class(value: Any, *, drop_dirac: bool) -> dict[str, Any]: + """Map form of a requirements or hints list keyed by class (rule 4). + + ``dirac:`` hints are dropped (rule 3). A ``dirac:`` requirement is kept, + so that the policy refuses it: a runner must honour a requirement, and no + CWL runner knows that class. + """ + if isinstance(value, dict): + items = [{"class": k, **(v or {})} for k, v in value.items()] + else: + items = list(value) + result = {} + for item in items: + item = dict(item) + cls = item.pop("class") + if drop_dirac and str(cls).startswith(DIRAC_PREFIX): + continue + result[cls] = _strip(item, metadata=cls == "SchemaDefRequirement") + return result + + +def _refs(value: Any, own_name: str | None) -> Any: + """Shorten the references inside a parameter or step input (rule 5).""" + if isinstance(value, dict): + return { + k: (_shorten_refs(v, own_name) if k in REFERENCE_FIELDS else v) + for k, v in value.items() + } + return _shorten_refs(value, own_name) + + +def _shorten_refs(value: Any, own_name: str | None) -> Any: + if isinstance(value, str): + return _short(value, own_name) + if isinstance(value, list): + return [_short(v, own_name) if isinstance(v, str) else v for v in value] + return value + + +def _short(identifier: str, scope: str | None) -> str: + """Shorten an identifier relative to its enclosing scope (rule 5). + + Only a fully scoped identifier loses the scope: ``s0/out`` in a workflow + whose own name is ``s0`` names the output of its step ``s0``. + """ + if "#" in identifier: + identifier = identifier.split("#", 1)[1] + if scope and identifier.startswith(scope + "/"): + identifier = identifier[len(scope) + 1 :] + elif "://" in identifier or identifier.startswith("_:"): + identifier = identifier.rsplit("/", 1)[-1] + return identifier + + +def _is_dropped(key: str) -> bool: + return key in METADATA_FIELDS or key.startswith(DIRAC_PREFIX) + + +def _strip(value: Any, *, metadata: bool = True) -> Any: + """Drop ``dirac:`` fields, and metadata if asked, at every level (rule 3). + + ``default`` values are user data and kept verbatim, as are the names of a + record's fields in map form. Anonymous schema names (``name: _:``), + which some CWL libraries mint afresh on every load, are dropped. + """ + if isinstance(value, dict): + result = {} + for k, v in value.items(): + if k.startswith(DIRAC_PREFIX) or (metadata and k in METADATA_FIELDS): + continue + if k == "name" and isinstance(v, str) and v.startswith("_:"): + continue + if k == "default": + result[k] = v + elif k == "fields" and isinstance(v, dict): + result[k] = {n: _strip(f, metadata=metadata) for n, f in v.items()} + else: + result[k] = _strip(v, metadata=metadata) + return result + if isinstance(value, list): + return [_strip(v, metadata=metadata) for v in value] + return value diff --git a/diracx-core/tests/test_cwl_canonical.py b/diracx-core/tests/test_cwl_canonical.py new file mode 100644 index 000000000..b9c210b68 --- /dev/null +++ b/diracx-core/tests/test_cwl_canonical.py @@ -0,0 +1,442 @@ +from __future__ import annotations + +import copy +import json + +import pytest + +from diracx.core.cwl import ( + canonical_bytes, + decompose, + format_reference, + pack, + parse_reference, + unpack, +) + + +def merge_tool(command: str = "hadd") -> dict: + return { + "cwlVersion": "v1.2", + "class": "CommandLineTool", + "id": "file:///home/someone/prod/merge.cwl#main", + "doc": "Merge files", + "baseCommand": [command], + "inputs": {"input-data": {"type": "File[]", "inputBinding": {"position": 1}}}, + "outputs": {"merged": {"type": "File", "outputBinding": {"glob": "*.root"}}}, + } + + +def workflow(*tools: dict) -> dict: + return { + "cwlVersion": "v1.2", + "class": "Workflow", + "id": "transformation_3", + "label": "MCMerge", + "hints": [{"class": "dirac:Transformation", "group_size": 4}], + "inputs": [{"id": "input-data", "type": "File[]"}], + "outputs": [ + { + "id": "merged", + "type": "File", + "outputSource": f"step{len(tools) - 1}/merged", + } + ], + "steps": [ + { + "id": f"#transformation_3/step{i}", + "run": tool, + "in": {"input-data": "input-data"}, + "out": ["merged"], + } + for i, tool in enumerate(tools) + ], + } + + +def root_id(process: dict) -> bytes: + return decompose(process)[0] + + +def test_same_document_loaded_from_two_places_has_one_identity(): + a, b = merge_tool(), merge_tool() + b["id"] = "file:///tmp/elsewhere/merge.cwl#main" + assert root_id(a) == root_id(b) + + +def test_identity_ignores_formatting_key_order_and_map_vs_list_form(): + a = merge_tool() + b = json.loads(json.dumps(a, sort_keys=True)) + b["inputs"] = [ + {"id": "input-data", "inputBinding": {"position": 1}, "type": "File[]"} + ] + b["outputs"] = [ + {"id": "merged", "type": "File", "outputBinding": {"glob": "*.root"}} + ] + assert root_id(a) == root_id(b) + + +def test_identity_ignores_metadata_and_dirac_hints(): + a = workflow(merge_tool()) + b = copy.deepcopy(a) + b["label"] = "something else" + b["hints"] = [{"class": "dirac:Transformation", "group_size": 8}] + b["inputs"][0]["dirac:Feeder"] = {"name": "Seeds"} + assert root_id(a) == root_id(b) + + +def formatted(prefix: str, uri: str) -> dict: + tool = merge_tool() + tool["$namespaces"] = {prefix: uri} + tool["inputs"]["input-data"]["format"] = f"{prefix}:format_2572" + return tool + + +def test_identity_covers_what_a_prefix_means_not_which_prefix(): + edam = "http://edamontology.org/" + assert root_id(formatted("edam", edam)) == root_id(formatted("e", edam)) + assert root_id(formatted("edam", edam)) != root_id( + formatted("edam", "http://example.org/") + ) + + +def test_nested_processes_inherit_and_may_rebind_prefixes(): + inherited = workflow(formatted("edam", "http://edamontology.org/")) + child = inherited["steps"][0]["run"] + inherited["$namespaces"] = child.pop("$namespaces") + root, nodes = decompose(inherited) + assert root_id(formatted("edam", "http://edamontology.org/")) in nodes + rebound = copy.deepcopy(inherited) + rebound["steps"][0]["run"]["$namespaces"] = {"edam": "http://example.org/"} + assert root_id(rebound) != root + + +def test_fully_scoped_step_ports_match_the_short_form(): + short = workflow(merge_tool()) + short["steps"][0]["out"] = ["merged"] + scoped = copy.deepcopy(short) + scoped["id"] = "file:///home/someone/prod/merge.cwl#transformation_3" + scope = "file:///home/someone/prod/merge.cwl#transformation_3" + scoped["inputs"][0]["id"] = f"{scope}/input-data" + scoped["outputs"][0]["id"] = f"{scope}/merged" + scoped["outputs"][0]["outputSource"] = f"{scope}/step0/merged" + step = scoped["steps"][0] + step["id"] = f"{scope}/step0" + step["in"] = [{"id": f"{scope}/step0/input-data", "source": f"{scope}/input-data"}] + step["out"] = [{"id": f"{scope}/step0/merged"}] + assert root_id(scoped) == root_id(short) + + +def test_a_relative_reference_keeps_a_step_named_like_its_workflow(): + def named(process_id: str) -> dict: + wf = workflow(merge_tool()) + wf["id"] = process_id + wf["steps"][0]["id"] = "s0" + wf["outputs"][0]["outputSource"] = "s0/merged" + return wf + + root, nodes = decompose(named("s0")) + assert json.loads(nodes[root].body)["outputs"]["merged"]["outputSource"] == ( + "s0/merged" + ) + assert root == root_id(named("something-else")) + + +def test_dirac_hints_are_dropped_but_dirac_requirements_are_kept(): + hinted = merge_tool() + hinted["hints"] = {"dirac:Transformation": {"group_size": 4}} + assert root_id(hinted) == root_id(merge_tool()) + required = merge_tool() + required["requirements"] = {"dirac:Unsupported": {}} + root, nodes = decompose(required) + assert "dirac:Unsupported" in json.loads(nodes[root].body)["requirements"] + + on_step = workflow(merge_tool()) + on_step["steps"][0]["hints"] = {"dirac:Transformation": {"group_size": 4}} + on_step["steps"][0]["requirements"] = {"dirac:Unsupported": {}} + root, nodes = decompose(on_step) + step = json.loads(nodes[root].body)["steps"]["step0"] + assert "hints" not in step + assert "dirac:Unsupported" in step["requirements"] + + +def test_anonymous_schema_names_do_not_change_the_identity(): + a, b = merge_tool(), merge_tool() + for tool, name in ((a, "_:16606bec"), (b, "_:ad4e1bb5")): + tool["inputs"]["input-data"]["type"] = { + "type": "array", + "items": "File", + "name": name, + } + assert root_id(a) == root_id(b) + + +def test_what_runs_changes_the_identity(): + assert root_id(merge_tool("hadd")) != root_id(merge_tool("hadd -f")) + renamed = merge_tool() + renamed["inputs"] = {"files": renamed["inputs"]["input-data"]} + assert root_id(renamed) != root_id(merge_tool()) + + +def test_default_values_are_kept_verbatim(): + a, b = merge_tool(), merge_tool() + a["inputs"]["conf"] = {"type": "Any", "default": {"label": "A"}} + b["inputs"]["conf"] = {"type": "Any", "default": {"label": "B"}} + assert root_id(a) != root_id(b) + + +@pytest.mark.parametrize( + "where", + [ + lambda t, v: t.update( + requirements={"EnvVarRequirement": {"envDef": {"label": v, "doc": v}}} + ), + lambda t, v: t["inputs"].update( + seed={"type": {"type": "record", "fields": {"label": v, "doc": "int"}}} + ), + lambda t, v: t.update(hints={"SoftwareRequirement": {"packages": {"doc": v}}}), + ], +) +def test_user_chosen_names_called_doc_or_label_are_kept(where): + a, b = merge_tool(), merge_tool() + where(a, "string") + where(b, "File") + assert root_id(a) != root_id(b) + + +def typed(name: str, reference: str, value_type: str = "int") -> dict: + tool = merge_tool() + tool["requirements"] = { + "SchemaDefRequirement": { + "types": [ + {"name": "#Colour", "type": "enum", "symbols": ["#Colour/red"]}, + { + "name": name, + "type": "record", + "fields": [{"name": "value", "type": value_type}], + }, + ] + } + } + tool["inputs"]["seed"] = {"type": reference} + tool["inputs"]["seeds"] = {"type": f"{reference}[]"} + return tool + + +def test_named_types_are_written_by_short_name(): + root = root_id(typed("Seed", "Seed")) + assert root == root_id(typed("#Seed", "#Seed")) + assert root == root_id(typed("file:///p/merge.cwl#main/Seed", "#main/Seed")) + assert root != root_id(typed("Seed", "Seed", value_type="string")) + body = json.loads(decompose(typed("#main/Seed", "#main/Seed"))[1][root].body) + assert body["inputs"]["seeds"]["type"] == {"type": "array", "items": "Seed"} + assert body["requirements"]["SchemaDefRequirement"]["types"][0]["symbols"] == [ + "red" + ] + + +def test_named_types_must_be_defined_by_the_process_using_them(): + tool = merge_tool() + tool["inputs"]["seed"] = {"type": "Seed"} + with pytest.raises(ValueError, match="not defined by the process"): + decompose(workflow(tool)) + + +def test_pack_scopes_named_types_by_component(): + root, nodes = decompose(workflow(typed("Seed", "Seed"))) + packed = pack(root, nodes) + tool = packed["$graph"][1] + schema = tool["requirements"]["SchemaDefRequirement"]["types"] + assert schema[1]["name"] == tool["id"] + "/Seed" + assert tool["inputs"]["seeds"]["type"]["items"] == tool["id"] + "/Seed" + assert unpack(packed) == (root, nodes) + + +def test_child_change_propagates_to_parents_but_not_siblings(): + before_root, before = decompose(workflow(merge_tool("a"), merge_tool("b"))) + after_root, after = decompose(workflow(merge_tool("a"), merge_tool("c"))) + assert before_root != after_root + assert root_id(merge_tool("a")) in before and root_id(merge_tool("a")) in after + + +def test_shared_child_is_one_node(): + root, nodes = decompose(workflow(merge_tool(), merge_tool())) + assert len(nodes) == 2 + assert nodes[root].children == (root_id(merge_tool()),) + + +def test_child_body_is_a_reference_to_its_identity(): + root, nodes = decompose(workflow(merge_tool())) + body = json.loads(nodes[root].body) + child = parse_reference(body["steps"]["step0"]["run"]) + assert child == root_id(merge_tool()) + assert body["outputs"]["merged"]["outputSource"] == "step0/merged" + + +def test_existing_reference_is_kept_without_its_body(): + child = root_id(merge_tool()) + wf = workflow(merge_tool()) + wf["steps"][0]["run"] = format_reference(child) + assert root_id(wf) == root_id(workflow(merge_tool())) + + +@pytest.mark.parametrize( + "mutate, message", + [ + (lambda p: p.update(cwlVersion="v1.0"), "upgrade"), + (lambda p: p["steps"][0].update(run="tools/merge.cwl"), "Unresolved"), + (lambda p: p.update({"class": "Nope"}), "Not a CWL process"), + ( + lambda p: p.update( + requirements={ + "InitialWorkDirRequirement": { + "listing": [{"entryname": "f", "entry": {"$include": "x"}}] + } + } + ), + r"Unresolved \$include", + ), + ( + lambda p: p["steps"][0]["run"]["inputs"].update( + conf={"type": "Any", "default": {"$import": "conf.yml"}} + ), + r"Unresolved \$import", + ), + (lambda p: p.update({"$base": "http://example.org/"}), r"Unresolved \$base"), + ], +) +def test_rejected_documents(mutate, message): + wf = workflow(merge_tool()) + mutate(wf) + with pytest.raises(ValueError, match=message): + decompose(wf) + + +def test_pack_round_trips_to_the_same_tree(): + root, nodes = decompose(workflow(merge_tool("a"), merge_tool("b"))) + packed = pack(root, nodes) + assert packed["$graph"][0]["id"] == "#main" + assert unpack(packed) == (root, nodes) + + +def test_unpack_rejects_a_tampered_component(): + root, nodes = decompose(workflow(merge_tool())) + packed = pack(root, nodes) + packed["$graph"][1]["baseCommand"] = ["rm", "-rf"] + with pytest.raises(ValueError, match="does not match"): + unpack(packed) + + +def test_unpack_verifies_a_shared_component_once(): + # Each level runs the level below twice: 2**40 paths, 41 components + nodes = decompose(merge_tool())[1] + root = next(iter(nodes)) + for _ in range(40): + level = workflow(merge_tool(), merge_tool()) + for step in level["steps"]: + step["run"] = format_reference(root) + root, own = decompose(level) + nodes.update(own) + assert unpack(pack(root, nodes)) == (root, nodes) + + +def test_unpack_rejects_a_component_that_runs_itself(): + packed = pack(*decompose(workflow(merge_tool()))) + packed["$graph"][0]["steps"]["step0"]["run"] = "#main" + with pytest.raises(ValueError, match="runs itself"): + unpack(packed) + + +def test_source_lists_and_scatter_are_shortened_like_single_sources(): + def scattered(scope: str) -> dict: + wf = workflow(merge_tool()) + wf["requirements"] = { + "ScatterFeatureRequirement": {}, + "MultipleInputFeatureRequirement": {}, + } + wf["inputs"].append({"id": f"{scope}extra", "type": "File[]"}) + step = wf["steps"][0] + step["in"] = { + "input-data": { + "source": [f"{scope}input-data", f"{scope}extra"], + "linkMerge": "merge_flattened", + } + } + step["scatter"] = "input-data" + step["dirac:Group"] = {"size": 4} + return wf + + root, nodes = decompose(scattered("#transformation_3/")) + step = json.loads(nodes[root].body)["steps"]["step0"] + assert step["in"]["input-data"]["source"] == ["input-data", "extra"] + assert step["scatter"] == "input-data" + assert "dirac:Group" not in step + assert root == root_id(scattered("")) + + +@pytest.mark.parametrize( + "short, expanded", + [ + ("File[]", {"type": "array", "items": "File"}), + ("File?", ["null", "File"]), + ("File[]?", ["null", {"type": "array", "items": "File"}]), + ("File[][]", {"type": "array", "items": {"type": "array", "items": "File"}}), + (["File?", "int"], ["null", "File", "int"]), + ], +) +def test_type_shorthands_are_expanded(short, expanded): + a, b = merge_tool(), merge_tool() + a["inputs"]["input-data"]["type"] = short + b["inputs"]["input-data"]["type"] = expanded + root, nodes = decompose(a) + assert root == root_id(b) + body = json.loads(nodes[root].body) + assert body["inputs"]["input-data"]["type"] == expanded + + +def test_union_types_name_their_members(): + a, b = typed("Seed", "Seed"), typed("Seed", "Seed") + b["inputs"]["seed"] = {"type": ["null", "#main/Seed"]} + a["inputs"]["seed"] = {"type": ["null", "Seed"]} + root, nodes = decompose(a) + assert root == root_id(b) + assert json.loads(nodes[root].body)["inputs"]["seed"]["type"] == ["null", "Seed"] + + +def test_a_process_id_without_a_fragment_is_ignored(): + a = merge_tool() + a["id"] = "file:///home/someone/prod/merge.cwl" + assert root_id(a) == root_id(merge_tool()) + + +@pytest.mark.parametrize( + "reference, message", + [ + ("tools/merge.cwl", "Not a process reference"), + ("proc:sha256-abcd", "Not a SHA-256 process reference"), + ], +) +def test_malformed_process_references_are_refused(reference, message): + with pytest.raises(ValueError, match=message): + parse_reference(reference) + + +def test_duplicate_ids_are_refused(): + tool = merge_tool() + tool["inputs"] = [{"id": "x", "type": "File"}, {"id": "#main/x", "type": "File"}] + tool["id"] = "#main" + with pytest.raises(ValueError, match="Duplicate id 'x'"): + decompose(tool) + + +def test_canonical_bytes_are_compact_and_sorted(): + assert canonical_bytes({"b": 1, "a": "é"}) == '{"a":"é","b":1}'.encode() + + +def test_golden_identity(): + # Changing this value means every stored identity changes: bump + # CANONICAL_VERSION and plan the migration instead of editing it. + assert ( + root_id(workflow(merge_tool())).hex() + == "b06bc1e02806b735f5100eff9d5e9b00fb0ea154c0856d3e03b5d500689344a9" + )