From 3e4d98bf9684385594091d61541a3c2fcb183140 Mon Sep 17 00:00:00 2001 From: Liam Huber Date: Tue, 11 Aug 2026 12:24:42 -0700 Subject: [PATCH 1/4] Only convert single-output atomic nodes Due to axiomatic differences in how output of python functions is to be interpreted. Co-authored-by: Claude Signed-off-by: Liam Huber --- src/flowrep/api/tools.py | 3 + .../converters/python_workflow_definition.py | 75 ++++- .../mono-workflow.json | 23 ++ .../test_python_workflow_definition.py | 298 +++++++++++++----- 4 files changed, 309 insertions(+), 90 deletions(-) create mode 100644 tests/flowrep_static/python-workflow-definition/mono-workflow.json diff --git a/src/flowrep/api/tools.py b/src/flowrep/api/tools.py index cfd11b36..96d2cf35 100644 --- a/src/flowrep/api/tools.py +++ b/src/flowrep/api/tools.py @@ -6,6 +6,9 @@ """ from flowrep.compiler.source import flowrep2python as flowrep2python +from flowrep.converters.python_workflow_definition import ( + OutputContractError as OutputContractError, +) from flowrep.converters.python_workflow_definition import flowrep2pwd as flowrep2pwd from flowrep.converters.python_workflow_definition import pwd2flowrep as pwd2flowrep from flowrep.parsers.atomic_parser import atomic as atomic diff --git a/src/flowrep/converters/python_workflow_definition.py b/src/flowrep/converters/python_workflow_definition.py index 99923c37..4d0c7a8a 100644 --- a/src/flowrep/converters/python_workflow_definition.py +++ b/src/flowrep/converters/python_workflow_definition.py @@ -36,6 +36,23 @@ _PORT_SANITIZE_PREFIX: str = "flowrep_sanitized_" +class OutputContractError(ValueError): + """ + A node's output contract is not representable in the target format. + + flowrep and pwd disagree about how a function's return value maps onto + output ports. pwd says ``sourcePort: null`` is the entire return value and + a named ``sourcePort`` is a key of a returned dict; flowrep says a single + output port is the entire return value and N>1 ports unpack an N-tuple. + The two overlap only on *exactly one output, meaning the whole return + value*. Converting a node outside that overlap would produce a recipe that + executes incorrectly, so we refuse rather than mistranslate. + + Subclasses :class:`ValueError` so that callers already handling the + converter's other validation failures keep working. + """ + + def _needs_sanitization(port: str) -> bool: """Return ``True`` if *port* is not a valid flowrep :class:`Label`.""" return not base_models.is_valid_label(port) @@ -55,7 +72,12 @@ def _sanitize_port(port: str) -> str: def _desanitize_port(port: str) -> str: - """Reverse :func:`_sanitize_port` when converting back to pwd.""" + """ + Reverse :func:`_sanitize_port` when converting back to pwd. + + Only input (``targetPort``) names need this; output ports never round-trip + through pwd by name — see :func:`_build_pwd_edges`. + """ if port.startswith(_PORT_SANITIZE_PREFIX): candidate = port[len(_PORT_SANITIZE_PREFIX) :] if candidate and _needs_sanitization(candidate): @@ -136,8 +158,10 @@ def flowrep2pwd( Raises: ValueError: If any child is non-atomic, or if *terminal_inputs* does not exactly cover the workflow's inputs. + OutputContractError: If any child has more than one output port. """ _validate_flat_workflow(wf) + _validate_mono_output(wf) _validate_terminal_inputs(wf, terminal_inputs) id_counter = _IdCounter() @@ -385,6 +409,33 @@ def _validate_flat_workflow(wf: workflow_recipe.WorkflowRecipe) -> None: ) +def _validate_mono_output(wf: workflow_recipe.WorkflowRecipe) -> None: + """ + Raise :class:`OutputContractError` if any child has more than one output. + + Must be called *after* :func:`_validate_flat_workflow`, which guarantees + every child is an :class:`AtomicRecipe`. + """ + offenders: list[str] = [] + for label, node in wf.nodes.items(): + # Guaranteed by _validate_flat_workflow + assert isinstance(node, atomic_recipe.AtomicRecipe) + if len(node.outputs) > 1: + offenders.append( + f"'{label}' ({node.fully_qualified_name}) has " + f"{len(node.outputs)} outputs {node.outputs}" + ) + if offenders: + raise OutputContractError( + "flowrep2pwd requires every child to have exactly one output. " + "flowrep reads multiple output ports as unpacking a tuple, whereas " + "pwd reads them as keys of a single returned dict, so the converted " + "workflow would not execute correctly. Offending nodes: " + + "; ".join(offenders) + + "." + ) + + def _validate_terminal_inputs( wf: workflow_recipe.WorkflowRecipe, terminal_inputs: dict[str, Any], @@ -405,19 +456,6 @@ def _validate_terminal_inputs( ) -def _flowrep_port_to_pwd_source_port(port: str) -> str | None: - """ - Map a flowrep output-port name back to a pwd ``sourcePort`` value. - - Returns ``None`` for the default-output sentinel so that the PWD edge - validator stores it as :data:`pwd.INTERNAL_DEFAULT_HANDLE`. Otherwise - reverses any sanitisation applied by :func:`_sanitize_port`. - """ - if port == _DEFAULT_OUTPUT_PORT: - return None - return _desanitize_port(port) - - def _build_pwd_edges( wf: workflow_recipe.WorkflowRecipe, input_node_ids: dict[str, int], @@ -430,6 +468,11 @@ def _build_pwd_edges( Edges are emitted in a deterministic order that preserves the input-port ordering of each child node — this is important for consumers that rely on edge-list order (e.g. ``get_list``). + + Every function-node source emits ``sourcePort=None``. A flowrep child has + exactly one output (enforced by :func:`_validate_mono_output`) and that + output *is* the whole return value, which is precisely what pwd spells as a + null ``sourcePort``. The flowrep port's name is therefore not carried over. """ pwd_edges: list[pwd.PythonWorkflowDefinitionEdge] = [] @@ -456,7 +499,7 @@ def _build_pwd_edges( pwd_edges.append( pwd.PythonWorkflowDefinitionEdge( source=func_node_ids[source.node], - sourcePort=_flowrep_port_to_pwd_source_port(source.port), + sourcePort=None, target=func_node_ids[label], targetPort=target_port_pwd, ) @@ -480,7 +523,7 @@ def _build_pwd_edges( pwd_edges.append( pwd.PythonWorkflowDefinitionEdge( source=func_node_ids[source.node], - sourcePort=_flowrep_port_to_pwd_source_port(source.port), + sourcePort=None, target=output_node_ids[port], targetPort=None, ) diff --git a/tests/flowrep_static/python-workflow-definition/mono-workflow.json b/tests/flowrep_static/python-workflow-definition/mono-workflow.json new file mode 100644 index 00000000..0f77c830 --- /dev/null +++ b/tests/flowrep_static/python-workflow-definition/mono-workflow.json @@ -0,0 +1,23 @@ +{ + "version": "0.1.0", + "nodes": [ + {"id": 0, "type": "function", "value": "workflow.scale"}, + {"id": 1, "type": "function", "value": "workflow.scale"}, + {"id": 2, "type": "function", "value": "python_workflow_definition.shared.get_list"}, + {"id": 3, "type": "function", "value": "workflow.summarize"}, + {"id": 4, "type": "input", "value": 2, "name": "x"}, + {"id": 5, "type": "input", "value": {"a": [1, 2], "b": {"c": 3}}, "name": "options"}, + {"id": 6, "type": "output", "name": "summary"}, + {"id": 7, "type": "output", "name": "passthrough"} + ], + "edges": [ + {"target": 0, "targetPort": "v", "source": 4, "sourcePort": null}, + {"target": 1, "targetPort": "v", "source": 4, "sourcePort": null}, + {"target": 2, "targetPort": "0", "source": 0, "sourcePort": null}, + {"target": 2, "targetPort": "1", "source": 1, "sourcePort": null}, + {"target": 3, "targetPort": "values", "source": 2, "sourcePort": null}, + {"target": 3, "targetPort": "options", "source": 5, "sourcePort": null}, + {"target": 6, "targetPort": null, "source": 3, "sourcePort": null}, + {"target": 7, "targetPort": null, "source": 5, "sourcePort": null} + ] +} diff --git a/tests/unit/converters/test_python_workflow_definition.py b/tests/unit/converters/test_python_workflow_definition.py index 707739e4..7941126b 100644 --- a/tests/unit/converters/test_python_workflow_definition.py +++ b/tests/unit/converters/test_python_workflow_definition.py @@ -121,6 +121,36 @@ def _input_defaults( tc.assertEqual(_input_defaults(pwd1), _input_defaults(pwd2)) +def _canonical_outputs(node: atomic_recipe.AtomicRecipe) -> list[str]: + """A single output's name is canonicalized to the sentinel by a round-trip.""" + if len(node.outputs) == 1: + return [pwd_conv._DEFAULT_OUTPUT_PORT] + return list(node.outputs) + + +def _canonical_edge_map( + wf: workflow_recipe.WorkflowRecipe, + edge_map: dict, +) -> dict: + """Rewrite SourceHandles onto the canonical mono-output port name.""" + mono = { + label: node.outputs[0] + for label, node in wf.nodes.items() + if len(node.outputs) == 1 + } + canonical = {} + for target, source in edge_map.items(): + if ( + isinstance(source, edge_models.SourceHandle) + and mono.get(source.node) == source.port + ): + source = edge_models.SourceHandle( + node=source.node, port=pwd_conv._DEFAULT_OUTPUT_PORT + ) + canonical[target] = source + return canonical + + def _assert_flowrep_roundtrip_equal( tc: unittest.TestCase, wf_orig: workflow_recipe.WorkflowRecipe, @@ -131,22 +161,30 @@ def _assert_flowrep_roundtrip_equal( """ Compare original and round-tripped flowrep. - The pwd format preserves explicit ``sourcePort`` names, so output-port - names are not lost in the round-trip. ``reference``, ``source_code``, - and ``inputs_with_defaults`` are not represented in pwd and are therefore - excluded from comparison. + A node's sole output port name is *not* preserved: ``flowrep2pwd`` emits it + as pwd's unnamed ``sourcePort``, and it returns as the default sentinel. + The original is canonicalized the same way before comparison, so a failure + here means something other than that known loss went wrong. + + ``reference``, ``source_code``, and ``inputs_with_defaults`` are not + represented in pwd and are therefore excluded from comparison. """ tc.assertEqual(wf_orig.inputs, wf_rt.inputs) tc.assertEqual(wf_orig.outputs, wf_rt.outputs) tc.assertEqual(set(wf_orig.nodes.keys()), set(wf_rt.nodes.keys())) for label in wf_orig.nodes: n1, n2 = wf_orig.nodes[label], wf_rt.nodes[label] + # pwd conversion only ever produces AtomicRecipe children. + assert isinstance(n1, atomic_recipe.AtomicRecipe) + assert isinstance(n2, atomic_recipe.AtomicRecipe) tc.assertEqual(n1.inputs, n2.inputs) - tc.assertEqual(n1.outputs, n2.outputs) + tc.assertEqual(_canonical_outputs(n1), n2.outputs) tc.assertEqual(n1.fully_qualified_name, n2.fully_qualified_name) tc.assertEqual(wf_orig.input_edges, wf_rt.input_edges) - tc.assertEqual(wf_orig.edges, wf_rt.edges) - tc.assertEqual(wf_orig.output_edges, wf_rt.output_edges) + tc.assertEqual(_canonical_edge_map(wf_orig, wf_orig.edges), wf_rt.edges) + tc.assertEqual( + _canonical_edge_map(wf_orig, wf_orig.output_edges), wf_rt.output_edges + ) tc.assertEqual(terminal_inputs, defaults_rt) @@ -321,6 +359,73 @@ def test_get_list_ports_sanitized(self): self.assertTrue(port.startswith(pwd_conv._PORT_SANITIZE_PREFIX)) +@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") +class TestPwd2FlowrepMono(unittest.TestCase): + """ + Structural conversion of the all-single-output fixture. + + Every ``sourcePort`` in this fixture is null, so it is the only shipped + fixture that survives the output-contract guard. It deliberately carries + the structural features the real-world fixtures exercise: input fan-out, + a repeated function value, nested-JSON defaults, non-identifier input + ports, and a pass-through edge. + """ + + def setUp(self): + pwd_wf = _load_pwd_workflow("mono-workflow.json") + self.wf, self.defaults = pwd_conv.pwd2flowrep(pwd_wf) + + def test_inputs(self): + self.assertEqual(self.wf.inputs, ["x", "options"]) + + def test_outputs(self): + self.assertEqual(self.wf.outputs, ["summary", "passthrough"]) + + def test_nested_defaults(self): + self.assertEqual( + self.defaults, + {"x": 2, "options": {"a": [1, 2], "b": {"c": 3}}}, + ) + + def test_repeated_function_labels(self): + """Two nodes sharing a value get distinct, index-suffixed labels.""" + self.assertIn("scale_0", self.wf.nodes) + self.assertIn("scale_1", self.wf.nodes) + + def test_all_atomic(self): + for node in self.wf.nodes.values(): + self.assertIsInstance(node, atomic_recipe.AtomicRecipe) + + def test_every_node_has_one_output(self): + for label, node in self.wf.nodes.items(): + with self.subTest(label=label): + self.assertEqual(node.outputs, [pwd_conv._DEFAULT_OUTPUT_PORT]) + + def test_fan_out_input(self): + """Input 'x' feeds both scale nodes.""" + targets = [ + target.node + for target, source in self.wf.input_edges.items() + if source.port == "x" + ] + self.assertEqual(sorted(targets), ["scale_0", "scale_1"]) + + def test_get_list_ports_sanitized(self): + get_list = self.wf.nodes["get_list_0"] + self.assertEqual( + get_list.inputs, + [ + pwd_conv._PORT_SANITIZE_PREFIX + "0", + pwd_conv._PORT_SANITIZE_PREFIX + "1", + ], + ) + + def test_pass_through_edge(self): + """Workflow input wired straight to a workflow output.""" + source = self.wf.output_edges[edge_models.OutputTarget(port="passthrough")] + self.assertEqual(source, edge_models.InputSource(port="options")) + + @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestPwd2FlowrepErrorCases(unittest.TestCase): @@ -525,6 +630,10 @@ def test_non_atomic_child_raises(self): self.assertIn("AtomicRecipe", str(ctx.exception)) self.assertIn("nested", str(ctx.exception)) + def test_output_contract_error_is_a_value_error(self): + """Subclassing ValueError keeps existing ``except ValueError`` callers working.""" + self.assertTrue(issubclass(pwd_conv.OutputContractError, ValueError)) + def test_exact_terminal_inputs_succeeds(self): wf = self._make_flat_workflow() result = pwd_conv.flowrep2pwd(wf, x=1, y=2) @@ -546,6 +655,9 @@ def _assert_roundtrip(self, filename: str) -> None: # Also verify the pwd representations directly, modulo node IDs _assert_pwd_structurally_equal(self, pwd_orig, pwd_rt) + def test_mono(self): + self._assert_roundtrip("mono-workflow.json") + def test_arithmetic(self): self._assert_roundtrip("arithmetic-workflow.json") @@ -591,8 +703,20 @@ def wf(x: float, y: float) -> float: node = workflow_parser.parse_workflow(wf) self._assert_roundtrip(node, {"x": 3.0, "y": 4.0}) - def test_multi_output(self): - """A function returning multiple values.""" + def test_multi_output_raises(self): + """A node with >1 output cannot be represented in pwd.""" + + def wf(x: float) -> float: + a, b = library.multi_result(x) + c = library.typed_add(a, b) + return c + + node = workflow_parser.parse_workflow(wf) + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.flowrep2pwd(node, x=5.0) + + def test_multi_output_error_names_the_node(self): + """The error identifies the offending node, its function, and its ports.""" def wf(x: float) -> float: a, b = library.multi_result(x) @@ -600,7 +724,14 @@ def wf(x: float) -> float: return c node = workflow_parser.parse_workflow(wf) - self._assert_roundtrip(node, {"x": 5.0}) + with self.assertRaises(pwd_conv.OutputContractError) as ctx: + pwd_conv.flowrep2pwd(node, x=5.0) + + message = str(ctx.exception) + self.assertIn("multi_result_0", message) + self.assertIn("multi_result", message) + for port in node.nodes["multi_result_0"].outputs: + self.assertIn(port, message) def test_single_node(self): """Simplest case: one function, all inputs wired, one output.""" @@ -612,44 +743,30 @@ def wf(x: float, y: float) -> float: node = workflow_parser.parse_workflow(wf) self._assert_roundtrip(node, {"x": 10.0, "y": 1.0}) - def test_multi_output_ports_preserved(self): - """Named output ports on multi-output nodes survive the round-trip.""" - - def wf(x: float) -> float: - a, b = library.multi_result(x) - c = library.typed_add(a, b) - return c - - node = workflow_parser.parse_workflow(wf) - pwd_wf = pwd_conv.flowrep2pwd(node, x=5.0) - fr_rt, _ = pwd_conv.pwd2flowrep(pwd_wf) - - # multi_result has >1 output, so port names are preserved exactly - orig_mr = [ - n for n in node.nodes.values() if "multi_result" in n.fully_qualified_name - ][0] - rt_mr = [ - n for n in fr_rt.nodes.values() if "multi_result" in n.fully_qualified_name - ][0] - self.assertEqual(orig_mr.outputs, rt_mr.outputs) + def test_single_output_port_name_canonicalized(self): + """ + A single output's name is deliberately *not* preserved. - def test_single_output_port_name_preserved(self): - """Explicit single-output port names round-trip via sourcePort strings.""" + A flowrep mono-output is the whole return value, and pwd spells that + ``sourcePort: null`` — which comes back as the default sentinel. The + name is the price of a recipe that actually executes. + """ def wf(x: float, y: float) -> float: z = library.typed_add(x, y) return z node = workflow_parser.parse_workflow(wf) - # The add node's single output has a real name (not __result__) add_node = node.nodes["typed_add_0"] self.assertNotEqual(add_node.outputs[0], pwd_conv._DEFAULT_OUTPUT_PORT) pwd_wf = pwd_conv.flowrep2pwd(node, x=10.0, y=1.0) fr_rt, _ = pwd_conv.pwd2flowrep(pwd_wf) - # Port name must survive - self.assertEqual(add_node.outputs, fr_rt.nodes["typed_add_0"].outputs) + self.assertEqual( + fr_rt.nodes["typed_add_0"].outputs, + [pwd_conv._DEFAULT_OUTPUT_PORT], + ) @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") @@ -658,59 +775,89 @@ class TestDefaultOutputPort(unittest.TestCase): def test_single_output_uses_sentinel(self): """A PWD node with sourcePort=null produces an output with the default name.""" - pwd_wf = _load_pwd_workflow("arithmetic-workflow.json") + pwd_wf = _load_pwd_workflow("mono-workflow.json") wf, _ = pwd_conv.pwd2flowrep(pwd_wf) - # get_square (the last function node) has a single unnamed output - square_nodes = [ - n - for n in wf.nodes.values() - if n.fully_qualified_name == "workflow.get_square" - ] - self.assertEqual(len(square_nodes), 1) - self.assertIn(pwd_conv._DEFAULT_OUTPUT_PORT, square_nodes[0].outputs) + self.assertEqual( + wf.nodes["summarize_0"].outputs, + [pwd_conv._DEFAULT_OUTPUT_PORT], + ) def test_sentinel_roundtrips_to_null(self): - """default name port serializes back to sourcePort=null in PWD.""" - pwd_orig = _load_pwd_workflow("arithmetic-workflow.json") + """The default-name port serializes back to sourcePort=null in PWD.""" + pwd_orig = _load_pwd_workflow("mono-workflow.json") fr, defaults = pwd_conv.pwd2flowrep(pwd_orig) pwd_rt = pwd_conv.flowrep2pwd(fr, **defaults) - null_source_edges = [ - e - for e in pwd_rt.edges - if e.sourcePort == pwd_models.INTERNAL_DEFAULT_HANDLE - ] - orig_null = [ - e - for e in pwd_orig.edges - if e.sourcePort == pwd_models.INTERNAL_DEFAULT_HANDLE - ] - self.assertEqual(len(null_source_edges), len(orig_null)) + self.assertTrue( + all( + e.sourcePort == pwd_models.INTERNAL_DEFAULT_HANDLE for e in pwd_rt.edges + ) + ) - def test_explicit_source_port_preserved(self): - """Named sourcePorts like 'prod' must NOT become null in round-trip.""" - pwd_orig = _load_pwd_workflow("arithmetic-workflow.json") - fr, defaults = pwd_conv.pwd2flowrep(pwd_orig) - pwd_rt = pwd_conv.flowrep2pwd(fr, **defaults) + def test_named_single_output_canonicalized_to_null(self): + """ + A single output named anything else still emits sourcePort=null. - named_orig = { - (e.source, e.sourcePort) - for e in pwd_orig.edges - if e.sourcePort != pwd_models.INTERNAL_DEFAULT_HANDLE - } - named_rt = { - (e.source, e.sourcePort) - for e in pwd_rt.edges - if e.sourcePort != pwd_models.INTERNAL_DEFAULT_HANDLE - } - # Can't compare source IDs across pwd instances, but count must match - self.assertEqual(len(named_orig), len(named_rt)) + pwd reads a named ``sourcePort`` as a dict key, so emitting the flowrep + port name would produce a workflow that subscripts a non-dict at run time. + """ + + def wf(x: float, y: float) -> float: + z = library.typed_add(x, y) + return z + + node = workflow_parser.parse_workflow(wf) + self.assertNotEqual( + node.nodes["typed_add_0"].outputs[0], + pwd_conv._DEFAULT_OUTPUT_PORT, + ) + + pwd_wf = pwd_conv.flowrep2pwd(node, x=1.0, y=2.0) + self.assertTrue( + all( + e.sourcePort == pwd_models.INTERNAL_DEFAULT_HANDLE for e in pwd_wf.edges + ) + ) @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestSanitizedPortRoundTrip(unittest.TestCase): """Integer-string ports from ``get_list`` survive a full round-trip.""" + def test_mono_get_list_ports(self): + """Ports '0' and '1' on the get_list node survive a full round-trip.""" + pwd_orig = _load_pwd_workflow("mono-workflow.json") + fr_1, defaults_1 = pwd_conv.pwd2flowrep(pwd_orig) + + for port in fr_1.nodes["get_list_0"].inputs: + self.assertTrue(port.startswith(pwd_conv._PORT_SANITIZE_PREFIX)) + + pwd_rt = pwd_conv.flowrep2pwd(fr_1, **defaults_1) + fr_2, defaults_2 = pwd_conv.pwd2flowrep(pwd_rt) + + _assert_flowrep_roundtrip_equal(self, fr_1, fr_2, defaults_1, defaults_2) + + def test_mono_desanitized_ports_match_original(self): + """PWD edge targetPorts are restored to the original integer strings.""" + pwd_orig = _load_pwd_workflow("mono-workflow.json") + fr, defaults = pwd_conv.pwd2flowrep(pwd_orig) + pwd_rt = pwd_conv.flowrep2pwd(fr, **defaults) + + def _numeric_target_ports( + wf: pwd_models.PythonWorkflowDefinitionWorkflow, + ) -> set[str]: + return { + e.targetPort + for e in wf.edges + if e.targetPort is not None and not e.targetPort.isidentifier() + } + + self.assertEqual(_numeric_target_ports(pwd_orig), {"0", "1"}) + self.assertEqual( + _numeric_target_ports(pwd_orig), + _numeric_target_ports(pwd_rt), + ) + def test_quantum_espresso_get_list_ports(self): """Ports '0'–'4' on get_list nodes round-trip correctly.""" pwd_orig = _load_pwd_workflow("quantum_espresso-workflow.json") @@ -772,6 +919,9 @@ def _assert_counts_preserved(self, filename: str) -> None: self.assertEqual(len(pwd_orig.nodes), len(pwd_rt.nodes)) self.assertEqual(len(pwd_orig.edges), len(pwd_rt.edges)) + def test_mono(self): + self._assert_counts_preserved("mono-workflow.json") + def test_arithmetic(self): self._assert_counts_preserved("arithmetic-workflow.json") From a553f5954f5b612bbe88200a85f43516e0d95f3b Mon Sep 17 00:00:00 2001 From: Liam Huber Date: Tue, 11 Aug 2026 12:39:38 -0700 Subject: [PATCH 2/4] Cover reverse conversion and update notebook Co-authored-by: Claude Signed-off-by: Liam Huber --- notebooks/user-guide.ipynb | 5 + .../converters/python_workflow_definition.py | 87 ++++-- tests/flowrep_static/library.py | 10 + .../test_python_workflow_definition.py | 249 +++++------------- 4 files changed, 159 insertions(+), 192 deletions(-) diff --git a/notebooks/user-guide.ipynb b/notebooks/user-guide.ipynb index 4c4d50fd..8ce73219 100644 --- a/notebooks/user-guide.ipynb +++ b/notebooks/user-guide.ipynb @@ -3400,6 +3400,11 @@ "- Only **flat workflows** (all children are atomic) can be converted.\n", "- **Default values** must be supplied for every workflow input (PWD input nodes\n", " carry concrete values).\n", + "- Every node must have **exactly one output**. PWD reads multiple output ports as\n", + " keys of a single returned dict, while flowrep reads them as unpacking a tuple,\n", + " so rather than silently mistranslate, the converter raises\n", + " `OutputContractError`. Relatedly, a single output port's *name* is not\n", + " preserved — it becomes PWD's unnamed `sourcePort` and returns as `__result__`.\n", "\n", "### 7.1 flowrep → PWD" ], diff --git a/src/flowrep/converters/python_workflow_definition.py b/src/flowrep/converters/python_workflow_definition.py index 4d0c7a8a..c145d293 100644 --- a/src/flowrep/converters/python_workflow_definition.py +++ b/src/flowrep/converters/python_workflow_definition.py @@ -4,6 +4,24 @@ The ``python_workflow_definition`` (pwd) package is an **optional** dependency. It represents workflows as flat, non-nested DAGs of atomic function calls with explicit input/output nodes carrying JSON-serializable default values. + +The two formats hold different axioms about how a function's return value maps +onto a node's output ports: + +=========== ========================== ================================ + whole return value component of return +=========== ========================== ================================ +pwd ``sourcePort: null`` ``sourcePort: ""`` → + ``result[""]`` +flowrep exactly one output port N>1 ports unpack an N-tuple +=========== ========================== ================================ + +They overlap only on *exactly one output, meaning the whole return value*. +Anything outside that overlap raises :class:`OutputContractError` rather than +being mistranslated into a recipe that computes the wrong answer. As a +corollary, a flowrep node's sole output port name is not preserved across a +round-trip: it becomes pwd's unnamed ``sourcePort`` and returns as +:data:`_DEFAULT_OUTPUT_PORT`. """ from __future__ import annotations @@ -99,8 +117,13 @@ def pwd2flowrep( A ``(WorkflowRecipe, defaults)`` pair where *defaults* maps each workflow-input name to the default value carried by the corresponding PWD input node. + + Raises: + OutputContractError: If any function node is consumed via a named + ``sourcePort``. """ input_nodes, output_nodes, function_nodes = _categorize_pwd_nodes(wf.nodes) + _validate_pwd_output_contracts(wf.edges, function_nodes) label_map = _build_label_map(function_nodes) node_inputs, node_outputs = _collect_function_node_ports( @@ -240,6 +263,45 @@ def _categorize_pwd_nodes( return input_nodes, output_nodes, function_nodes +def _validate_pwd_output_contracts( + edges: list[pwd.PythonWorkflowDefinitionEdge], + function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode], +) -> None: + """ + Raise :class:`OutputContractError` for function nodes consumed by key. + + A named ``sourcePort`` tells pwd to subscript the node's single return value + with that key. A flowrep atomic node cannot express that: its outputs are + either the whole return value (one port) or a tuple unpacking (many ports). + Inspects raw ``sourcePort`` values, before sanitisation, so the message + quotes the author's own port names. + """ + named: dict[int, list[str]] = {} + for edge in edges: + if edge.source not in function_nodes: + continue + port = edge.sourcePort + if port is None or port == pwd.INTERNAL_DEFAULT_HANDLE: + continue + ports = named.setdefault(edge.source, []) + if port not in ports: + ports.append(port) + + if named: + offenders = [ + f"node {nid} ({function_nodes[nid].value}) is consumed via " + f"sourcePort(s) {sorted(named[nid])}" + for nid in sorted(named) + ] + raise OutputContractError( + "pwd2flowrep requires every function node to be consumed via a null " + "sourcePort. pwd reads a named sourcePort as a key of a single " + "returned dict, which a flowrep atomic node cannot express — its " + "outputs are either the whole return value (one port) or a tuple " + "unpacking (many). Offending nodes: " + "; ".join(offenders) + "." + ) + + def _build_label_map( function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode], ) -> dict[int, str]: @@ -261,13 +323,6 @@ def _build_label_map( return label_map -def _resolve_source_port(port: str | None) -> str: - """Map pwd's internal sourcePort to a flowrep output-port name.""" - if port is None or port == pwd.INTERNAL_DEFAULT_HANDLE: - return _DEFAULT_OUTPUT_PORT - return _sanitize_port(port) - - def _collect_function_node_ports( edges: list[pwd.PythonWorkflowDefinitionEdge], function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode], @@ -275,8 +330,9 @@ def _collect_function_node_ports( """ Collect ordered, unique input/output port names for each function node. - Port names that are not valid flowrep Labels are sanitized via - :func:`_sanitize_port`. + Input port names that are not valid flowrep Labels are sanitized via + :func:`_sanitize_port`. Output ports are always the single + :data:`_DEFAULT_OUTPUT_PORT`. """ node_inputs: dict[int, list[str]] = {nid: [] for nid in function_nodes} node_outputs: dict[int, list[str]] = {nid: [] for nid in function_nodes} @@ -289,12 +345,13 @@ def _collect_function_node_ports( if port not in ports: ports.append(port) - # Source ports (outputs of the function node) + # Outputs of the function node. Guaranteed null-sourced by + # _validate_pwd_output_contracts, so every such node has exactly the + # one default-named output port. if edge.source in function_nodes: - port = _resolve_source_port(edge.sourcePort) ports = node_outputs[edge.source] - if port not in ports: - ports.append(port) + if _DEFAULT_OUTPUT_PORT not in ports: + ports.append(_DEFAULT_OUTPUT_PORT) return node_inputs, node_outputs @@ -369,7 +426,7 @@ def _build_flowrep_edges( # Function node → workflow output out_name = output_nodes[edge.target].name source_label = label_map[edge.source] - source_port = _resolve_source_port(edge.sourcePort) + source_port = _DEFAULT_OUTPUT_PORT fr_output_edges[edge_models.OutputTarget(port=out_name)] = ( edge_models.SourceHandle(node=source_label, port=source_port) ) @@ -378,7 +435,7 @@ def _build_flowrep_edges( # Function node → function node (sibling edge) source_label = label_map[edge.source] target_label = label_map[edge.target] - source_port = _resolve_source_port(edge.sourcePort) + source_port = _DEFAULT_OUTPUT_PORT target_port = _sanitize_port(edge.targetPort) fr_edges[edge_models.TargetHandle(node=target_label, port=target_port)] = ( edge_models.SourceHandle(node=source_label, port=source_port) diff --git a/tests/flowrep_static/library.py b/tests/flowrep_static/library.py index 3e6be75d..4e51516d 100644 --- a/tests/flowrep_static/library.py +++ b/tests/flowrep_static/library.py @@ -216,3 +216,13 @@ def single_autoencoder(only): dc = Single(only) o = Single.flowrep_recipe_unpacking(dc) return o + + +def prod_and_div_dict(x, y): + """ + PWD's output axiom: several outputs are keys of one returned dict. + + Deliberately *not* decorated -- pwd's ``purepython`` runner filters nodes + with :func:`inspect.isfunction`. + """ + return {"prod": x * y, "div": x / y} diff --git a/tests/unit/converters/test_python_workflow_definition.py b/tests/unit/converters/test_python_workflow_definition.py index 7941126b..a980a32b 100644 --- a/tests/unit/converters/test_python_workflow_definition.py +++ b/tests/unit/converters/test_python_workflow_definition.py @@ -249,116 +249,6 @@ def test_sanitized_port_is_valid_label(self): self.assertTrue(sanitized.isidentifier()) -@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") -class TestPwd2FlowrepArithmetic(unittest.TestCase): - """Smoke-test conversion of the arithmetic example.""" - - def setUp(self): - pwd_wf = _load_pwd_workflow("arithmetic-workflow.json") - self.wf, self.defaults = pwd_conv.pwd2flowrep(pwd_wf) - - def test_inputs(self): - self.assertEqual(set(self.wf.inputs), {"x", "y"}) - - def test_outputs(self): - self.assertEqual(self.wf.outputs, ["result"]) - - def test_defaults(self): - self.assertEqual(self.defaults, {"x": 1, "y": 2}) - - def test_node_count(self): - self.assertEqual(len(self.wf.nodes), 3) - - def test_all_atomic(self): - for node in self.wf.nodes.values(): - self.assertIsInstance(node, atomic_recipe.AtomicRecipe) - - def test_multi_output_node(self): - """get_prod_and_div should have two named outputs.""" - prod_div = [ - n - for n in self.wf.nodes.values() - if n.fully_qualified_name == "workflow.get_prod_and_div" - ] - self.assertEqual(len(prod_div), 1) - self.assertEqual(set(prod_div[0].outputs), {"prod", "div"}) - - -@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") -class TestPwd2FlowrepNfdi(unittest.TestCase): - """Smoke-test conversion of the NFDI example.""" - - def setUp(self): - pwd_wf = _load_pwd_workflow("nfdi-workflow.json") - self.wf, self.defaults = pwd_conv.pwd2flowrep(pwd_wf) - - def test_inputs(self): - self.assertEqual(set(self.wf.inputs), {"domain_size", "source_directory"}) - - def test_defaults(self): - self.assertEqual(self.defaults["domain_size"], 2.0) - self.assertEqual(self.defaults["source_directory"], "source") - - def test_node_count(self): - self.assertEqual(len(self.wf.nodes), 6) - - def test_fan_out_input(self): - """source_directory feeds multiple children.""" - source_edges = [ - target - for target, source in self.wf.input_edges.items() - if source.port == "source_directory" - ] - # Nodes 2, 3, 4, 5 in the original all receive source_directory - self.assertGreater(len(source_edges), 1) - - -@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") -class TestPwd2FlowrepQuantumEspresso(unittest.TestCase): - """Smoke-test conversion of the quantum-espresso example.""" - - def setUp(self): - pwd_wf = _load_pwd_workflow("quantum_espresso-workflow.json") - self.wf, self.defaults = pwd_conv.pwd2flowrep(pwd_wf) - - def test_complex_defaults(self): - """Dict and list default values survive conversion.""" - self.assertEqual( - self.defaults["pseudopotentials"], - {"Al": "Al.pbe-n-kjpaw_psl.1.0.0.UPF"}, - ) - self.assertEqual(self.defaults["kpts"], [3, 3, 3]) - self.assertEqual(self.defaults["strain_lst"], [0.9, 0.95, 1.0, 1.05, 1.1]) - - def test_repeated_function(self): - """calculate_qe appears multiple times → distinct labels.""" - calc_nodes = [ - label - for label, n in self.wf.nodes.items() - if "calculate_qe" in n.fully_qualified_name - ] - self.assertEqual(len(calc_nodes), 6) - # All labels must be unique - self.assertEqual(len(calc_nodes), len(set(calc_nodes))) - - def test_get_list_ports_sanitized(self): - """get_list input ports like '0', '1' must be sanitized to valid labels.""" - list_nodes = [ - n - for n in self.wf.nodes.values() - if n.fully_qualified_name == "python_workflow_definition.shared.get_list" - ] - self.assertGreater(len(list_nodes), 0) - for node in list_nodes: - for port in node.inputs: - with self.subTest(port=port): - self.assertTrue( - port.isidentifier(), - f"Port {port!r} is not a valid identifier", - ) - self.assertTrue(port.startswith(pwd_conv._PORT_SANITIZE_PREFIX)) - - @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestPwd2FlowrepMono(unittest.TestCase): """ @@ -426,6 +316,77 @@ def test_pass_through_edge(self): self.assertEqual(source, edge_models.InputSource(port="options")) +@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") +class TestOutputContractRejection(unittest.TestCase): + """ + Real-world PWD workflows that flowrep cannot faithfully represent. + + A named ``sourcePort`` means "subscript the return value with this key". + A flowrep atomic node has no way to say that — its outputs are either the + whole return value (one port) or a tuple unpacking (many) — so converting + would silently change what the workflow computes. All three shipped + real-world fixtures fall in this category, which is why the mono fixture + exists. + """ + + def test_arithmetic_rejected(self): + pwd_wf = _load_pwd_workflow("arithmetic-workflow.json") + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.pwd2flowrep(pwd_wf) + + def test_nfdi_rejected(self): + pwd_wf = _load_pwd_workflow("nfdi-workflow.json") + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.pwd2flowrep(pwd_wf) + + def test_quantum_espresso_rejected(self): + pwd_wf = _load_pwd_workflow("quantum_espresso-workflow.json") + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.pwd2flowrep(pwd_wf) + + def test_error_names_node_function_and_ports(self): + pwd_wf = _load_pwd_workflow("arithmetic-workflow.json") + with self.assertRaises(pwd_conv.OutputContractError) as ctx: + pwd_conv.pwd2flowrep(pwd_wf) + + message = str(ctx.exception) + self.assertIn("workflow.get_prod_and_div", message) + self.assertIn("prod", message) + self.assertIn("div", message) + + def test_single_named_source_port_rejected(self): + """ + One named key is no better than several. + + flowrep cannot express "the 'value' key of this node's single return" + without inserting a getter node, which is out of scope. + """ + wf = pwd_models.PythonWorkflowDefinitionWorkflow.model_validate( + { + "version": "0.1.0", + "nodes": [ + {"id": 0, "type": "function", "value": "workflow.make"}, + {"id": 1, "type": "function", "value": "workflow.use"}, + {"id": 2, "type": "input", "value": 1, "name": "x"}, + {"id": 3, "type": "output", "name": "result"}, + ], + "edges": [ + {"target": 0, "targetPort": "x", "source": 2, "sourcePort": None}, + { + "target": 1, + "targetPort": "v", + "source": 0, + "sourcePort": "value", + }, + {"target": 3, "targetPort": None, "source": 1, "sourcePort": None}, + ], + } + ) + with self.assertRaises(pwd_conv.OutputContractError) as ctx: + pwd_conv.pwd2flowrep(wf) + self.assertIn("value", str(ctx.exception)) + + @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestPwd2FlowrepErrorCases(unittest.TestCase): @@ -658,15 +619,6 @@ def _assert_roundtrip(self, filename: str) -> None: def test_mono(self): self._assert_roundtrip("mono-workflow.json") - def test_arithmetic(self): - self._assert_roundtrip("arithmetic-workflow.json") - - def test_nfdi(self): - self._assert_roundtrip("nfdi-workflow.json") - - def test_quantum_espresso(self): - self._assert_roundtrip("quantum_espresso-workflow.json") - @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestRoundTripFlowrepToPwd(unittest.TestCase): @@ -822,7 +774,7 @@ def wf(x: float, y: float) -> float: @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestSanitizedPortRoundTrip(unittest.TestCase): - """Integer-string ports from ``get_list`` survive a full round-trip.""" + """Integer-string ports from a ``get_list`` node survive a full round-trip.""" def test_mono_get_list_ports(self): """Ports '0' and '1' on the get_list node survive a full round-trip.""" @@ -858,54 +810,6 @@ def _numeric_target_ports( _numeric_target_ports(pwd_rt), ) - def test_quantum_espresso_get_list_ports(self): - """Ports '0'–'4' on get_list nodes round-trip correctly.""" - pwd_orig = _load_pwd_workflow("quantum_espresso-workflow.json") - fr_1, defaults_1 = pwd_conv.pwd2flowrep(pwd_orig) - - # Verify sanitized port names in flowrep - list_nodes = [ - (label, n) - for label, n in fr_1.nodes.items() - if n.fully_qualified_name == "python_workflow_definition.shared.get_list" - ] - for _, node in list_nodes: - for port in node.inputs: - self.assertTrue(port.startswith(pwd_conv._PORT_SANITIZE_PREFIX)) - - # Full round-trip - pwd_rt = pwd_conv.flowrep2pwd(fr_1, **defaults_1) - fr_2, defaults_2 = pwd_conv.pwd2flowrep(pwd_rt) - - _assert_flowrep_roundtrip_equal(self, fr_1, fr_2, defaults_1, defaults_2) - - def test_desanitized_ports_match_original(self): - """PWD edge targetPorts are restored to original integer strings.""" - pwd_orig = _load_pwd_workflow("quantum_espresso-workflow.json") - fr, defaults = pwd_conv.pwd2flowrep(pwd_orig) - pwd_rt = pwd_conv.flowrep2pwd(fr, **defaults) - - # Collect targetPorts targeting get_list nodes in both - def _get_list_target_ports( - wf: pwd_models.PythonWorkflowDefinitionWorkflow, - ) -> set[str]: - list_ids = { - n.id - for n in wf.nodes - if isinstance(n, pwd_models.PythonWorkflowDefinitionFunctionNode) - and n.value == "python_workflow_definition.shared.get_list" - } - return { - e.targetPort - for e in wf.edges - if e.target in list_ids and e.targetPort is not None - } - - self.assertEqual( - _get_list_target_ports(pwd_orig), - _get_list_target_ports(pwd_rt), - ) - @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestRoundTripEdgeNodeCounts(unittest.TestCase): @@ -922,15 +826,6 @@ def _assert_counts_preserved(self, filename: str) -> None: def test_mono(self): self._assert_counts_preserved("mono-workflow.json") - def test_arithmetic(self): - self._assert_counts_preserved("arithmetic-workflow.json") - - def test_nfdi(self): - self._assert_counts_preserved("nfdi-workflow.json") - - def test_quantum_espresso(self): - self._assert_counts_preserved("quantum_espresso-workflow.json") - @unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") class TestPassThroughEdge(unittest.TestCase): From 71983ec16418091e94b69d7af1382da1c1ee1603 Mon Sep 17 00:00:00 2001 From: Liam Huber Date: Tue, 11 Aug 2026 12:41:57 -0700 Subject: [PATCH 3/4] Rename function For symmetry with validating the opposite direction Signed-off-by: Liam Huber --- src/flowrep/converters/python_workflow_definition.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/flowrep/converters/python_workflow_definition.py b/src/flowrep/converters/python_workflow_definition.py index c145d293..9197e342 100644 --- a/src/flowrep/converters/python_workflow_definition.py +++ b/src/flowrep/converters/python_workflow_definition.py @@ -184,7 +184,7 @@ def flowrep2pwd( OutputContractError: If any child has more than one output port. """ _validate_flat_workflow(wf) - _validate_mono_output(wf) + _validate_flowrep_output_contract(wf) _validate_terminal_inputs(wf, terminal_inputs) id_counter = _IdCounter() @@ -466,7 +466,7 @@ def _validate_flat_workflow(wf: workflow_recipe.WorkflowRecipe) -> None: ) -def _validate_mono_output(wf: workflow_recipe.WorkflowRecipe) -> None: +def _validate_flowrep_output_contract(wf: workflow_recipe.WorkflowRecipe) -> None: """ Raise :class:`OutputContractError` if any child has more than one output. @@ -527,7 +527,7 @@ def _build_pwd_edges( edge-list order (e.g. ``get_list``). Every function-node source emits ``sourcePort=None``. A flowrep child has - exactly one output (enforced by :func:`_validate_mono_output`) and that + exactly one output (enforced by :func:`_validate_flowrep_output_contract`) and that output *is* the whole return value, which is precisely what pwd spells as a null ``sourcePort``. The flowrep port's name is therefore not carried over. """ From 3cc6b5ba41def3e85292f6e7b899e52d4aa70486 Mon Sep 17 00:00:00 2001 From: Liam Huber Date: Tue, 11 Aug 2026 12:43:14 -0700 Subject: [PATCH 4/4] Add integration test Where we actually run the converted recipes. Co-authored-by: Claude Signed-off-by: Liam Huber --- tests/integration/test_pwd_execution.py | 163 ++++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 tests/integration/test_pwd_execution.py diff --git a/tests/integration/test_pwd_execution.py b/tests/integration/test_pwd_execution.py new file mode 100644 index 00000000..b23926a2 --- /dev/null +++ b/tests/integration/test_pwd_execution.py @@ -0,0 +1,163 @@ +""" +Cross-format execution parity for the pwd converter. + +Structural round-tripping is covered by the unit tests. These tests check the +thing that actually matters: that a converted recipe *computes the same answer* +under the target workflow manager — and that recipes which could not compute the +same answer are refused instead. + +Requires ``PYTHONPATH=tests`` so that pwd's ``purepython`` runner can import +``flowrep_static.library`` by dotted path. +""" + +from __future__ import annotations + +import pathlib +import tempfile +import unittest + +from flowrep import wfms +from flowrep.converters import python_workflow_definition as pwd_conv +from flowrep.parsers import workflow_parser + +from flowrep_static import library, makers # noqa: E402 + +try: + from python_workflow_definition import models as pwd_models + from python_workflow_definition import purepython + + _has_pwd = True +except ImportError: + _has_pwd = False + + +def _run_purepython(wf: pwd_models.PythonWorkflowDefinitionWorkflow): + """Execute a pwd workflow via its reference runner, which needs a file.""" + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "workflow.json" + path.write_text(wf.model_dump_json(), encoding="utf-8") + return purepython.load_workflow_json(str(path)) + + +def _linear(x: float, y: float) -> float: + """s = x + y; p = s * y. Every node has exactly one output.""" + s = library.typed_add(x, y) + p = library.typed_multiply(s, y) + return p + + +_KEYED_PWD_WORKFLOW = { + "version": "0.1.0", + "nodes": [ + { + "id": 0, + "type": "function", + "value": "flowrep_static.library.prod_and_div_dict", + }, + {"id": 1, "type": "function", "value": "flowrep_static.library.typed_add"}, + {"id": 2, "type": "input", "value": 6, "name": "x"}, + {"id": 3, "type": "input", "value": 3, "name": "y"}, + {"id": 4, "type": "output", "name": "result"}, + ], + "edges": [ + {"target": 0, "targetPort": "x", "source": 2, "sourcePort": None}, + {"target": 0, "targetPort": "y", "source": 3, "sourcePort": None}, + {"target": 1, "targetPort": "x", "source": 0, "sourcePort": "prod"}, + {"target": 1, "targetPort": "y", "source": 0, "sourcePort": "div"}, + {"target": 4, "targetPort": None, "source": 1, "sourcePort": None}, + ], +} + +_MONO_PWD_WORKFLOW = { + "version": "0.1.0", + "nodes": [ + {"id": 0, "type": "function", "value": "flowrep_static.library.typed_add"}, + {"id": 1, "type": "function", "value": "flowrep_static.library.typed_multiply"}, + {"id": 2, "type": "input", "value": 3.0, "name": "x"}, + {"id": 3, "type": "input", "value": 4.0, "name": "y"}, + {"id": 4, "type": "output", "name": "result"}, + ], + "edges": [ + {"target": 0, "targetPort": "x", "source": 2, "sourcePort": None}, + {"target": 0, "targetPort": "y", "source": 3, "sourcePort": None}, + {"target": 1, "targetPort": "x", "source": 0, "sourcePort": None}, + {"target": 1, "targetPort": "y", "source": 3, "sourcePort": None}, + {"target": 4, "targetPort": None, "source": 1, "sourcePort": None}, + ], +} + + +@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") +class TestMonoOutputExecutionParity(unittest.TestCase): + """Naked python == flowrep WfMS == pwd purepython, in both directions.""" + + def setUp(self): + self.expected = library.typed_multiply(library.typed_add(3.0, 4.0), 4.0) + + def test_expected_value(self): + """Guard the guard: (3 + 4) * 4 == 28.""" + self.assertEqual(self.expected, 28.0) + + def test_flowrep_to_pwd(self): + recipe = workflow_parser.parse_workflow(_linear) + + flowrep_result = wfms.run_recipe(recipe, x=3.0, y=4.0) + self.assertEqual( + flowrep_result.output_ports[recipe.outputs[0]].value, self.expected + ) + + pwd_wf = pwd_conv.flowrep2pwd(recipe, x=3.0, y=4.0) + self.assertEqual(_run_purepython(pwd_wf), self.expected) + + def test_pwd_to_flowrep(self): + pwd_wf = pwd_models.PythonWorkflowDefinitionWorkflow.model_validate( + _MONO_PWD_WORKFLOW + ) + self.assertEqual(_run_purepython(pwd_wf), self.expected) + + recipe, defaults = pwd_conv.pwd2flowrep(pwd_wf) + flowrep_result = wfms.run_recipe(recipe, **defaults) + self.assertEqual(flowrep_result.output_ports["result"].value, self.expected) + + +@unittest.skipUnless(_has_pwd, "python_workflow_definition not installed") +class TestCrossAxiomRefusal(unittest.TestCase): + """Workflows each format runs happily but neither can hand to the other.""" + + def test_pwd_keyed_workflow_runs_but_does_not_convert(self): + """ + The pwd side is valid and runnable — it is only *unconvertible*. + + Before the guard this produced a flowrep recipe that silently computed + ``'proddiv'`` instead of ``20.0``, by unpacking the dict's keys. + """ + pwd_wf = pwd_models.PythonWorkflowDefinitionWorkflow.model_validate( + _KEYED_PWD_WORKFLOW + ) + self.assertEqual(_run_purepython(pwd_wf), 20.0) + + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.pwd2flowrep(pwd_wf) + + def test_flowrep_tuple_workflow_runs_but_does_not_convert(self): + """ + The flowrep side is valid and runnable — it is only *unconvertible*. + + Before the guard the converted pwd workflow raised + ``TypeError: tuple indices must be integers`` at run time. + """ + + def wf(x: float) -> float: + a, b = library.multi_result(x) + c = library.typed_add(a, b) + return c + + # reference_free (rather than a bare parse_workflow) because `wf` is + # locally scoped here; wfms.run_recipe would otherwise try to import it + # by its -bearing qualified name and fail. + recipe = makers.reference_free(wf) + result = wfms.run_recipe(recipe, x=5.0) + self.assertEqual(result.output_ports[recipe.outputs[0]].value, 10.0) + + with self.assertRaises(pwd_conv.OutputContractError): + pwd_conv.flowrep2pwd(recipe, x=5.0)