Add unit tests for filename input handling across all workflow frameworks

tests/test_executorlib.py

@@ -1,3 +1,4 @@
+import sys
 import unittest
 from executorlib import SingleNodeExecutor
 from python_workflow_definition.executorlib import load_workflow_json
@@ -36,6 +37,26 @@ def get_square(x):
   ]
 }"""
 
+echo_function_str = """
+def echo(filename):
+    return filename
+"""
+
+filename_workflow_str = """
+{
+  "version": "0.1.0",
+  "nodes": [
+    {"id": 0, "type": "function", "value": "echo_module.echo"},
+    {"id": 1, "type": "input", "value": "image.png", "name": "filename"},
+    {"id": 2, "type": "output", "name": "result"}
+  ],
+  "edges": [
+    {"target": 0, "targetPort": "filename", "source": 1, "sourcePort": null},
+    {"target": 2, "targetPort": null, "source": 0, "sourcePort": null}
+  ]
+}"""
+
+
 class TestExecutorlib(unittest.TestCase):
     def test_executorlib(self):
         with open("workflow.py", "w") as f:
@@ -46,3 +67,19 @@ def test_executorlib(self):
 
         with SingleNodeExecutor(max_workers=1) as exe:
             self.assertEqual(load_workflow_json(file_name="workflow.json", exe=exe).result(), 6.25)
+
+    def test_executorlib_filename_input(self):
+        """A filename string like 'image.png' must be passed through as a plain
+        string input, not interpreted as a Python module path or a float."""
+        with open("echo_module.py", "w") as f:
+            f.write(echo_function_str)
+        sys.modules.pop("echo_module", None)
+
+        with open("filename_workflow.json", "w") as f:
+            f.write(filename_workflow_str)
+
+        with SingleNodeExecutor(max_workers=1) as exe:
+            result = load_workflow_json(
+                file_name="filename_workflow.json", exe=exe
+            ).result()
+        self.assertEqual(result, "image.png")

tests/test_jobflow.py

@@ -1,5 +1,6 @@
-import unittest
+import json
 import os
+import unittest
 from jobflow import job, Flow
 from jobflow.managers.local import run_locally
 from python_workflow_definition.jobflow import load_workflow_json, write_workflow_json
@@ -17,6 +18,10 @@ def get_square(x):
     return x ** 2
 
 
+def echo(filename):
+    return filename
+
+
 class TestJobflow(unittest.TestCase):
     def test_jobflow(self):
         workflow_json_filename = "jobflow_simple.json"
@@ -33,3 +38,27 @@ def test_jobflow(self):
 
         self.assertTrue(os.path.exists(workflow_json_filename))
         self.assertEqual(result[list(result.keys())[-1]][1].output, 6.25)
+
+    def test_jobflow_filename_input(self):
+        """A filename string like 'image.png' must be passed through as a plain
+        string input, not interpreted as a Python module path or a float."""
+        workflow_json_filename = "jobflow_filename.json"
+        echo_job = job(echo)
+        result_job = echo_job(filename="image.png")
+        flow = Flow([result_job])
+
+        write_workflow_json(flow=flow, file_name=workflow_json_filename)
+        self.assertTrue(os.path.exists(workflow_json_filename))
+
+        with open(workflow_json_filename) as f:
+            saved = json.load(f)
+        input_values = [
+            n["value"]
+            for n in saved["nodes"]
+            if n["type"] == "input"
+        ]
+        self.assertIn("image.png", input_values)
+
+        flow = load_workflow_json(file_name=workflow_json_filename)
+        result = run_locally(flow)
+        self.assertEqual(result[list(result.keys())[-1]][1].output, "image.png")

tests/test_models.py

@@ -51,6 +51,9 @@ def test_input_node_valid_values(self):
             1,
             1.1,
             "string",
+            "image.png",
+            "path/to/file.tar.gz",
+            "my.module.like.string",
             True,
             None,
             [1, 2],
@@ -73,6 +76,50 @@ def test_input_node_valid_values(self):
                     ).value
                 )
 
+    def test_input_node_filename_value_roundtrip(self):
+        """Input nodes with filename-like values (e.g. 'image.png') must survive a
+        full JSON serialise/deserialise round-trip without being misinterpreted as
+        a Python module path or a floating-point number."""
+        filenames = [
+            "image.png",
+            "archive.tar.gz",
+            "report.2024.pdf",
+            "data.csv",
+        ]
+        for filename in filenames:
+            with self.subTest(filename=filename):
+                node = PythonWorkflowDefinitionInputNode(
+                    id=1, type="input", name="file_input", value=filename
+                )
+                self.assertEqual(node.value, filename)
+                dumped = node.model_dump(mode="json")
+                self.assertEqual(dumped["value"], filename)
+                reloaded = PythonWorkflowDefinitionInputNode.model_validate(dumped)
+                self.assertEqual(reloaded.value, filename)
+
+    def test_workflow_with_filename_input_roundtrip(self):
+        """A full workflow containing a filename as an input value must serialise and
+        deserialise correctly through dump_json / load_json_str."""
+        workflow_dict = {
+            "version": "1.0",
+            "nodes": [
+                {"id": 1, "type": "input", "name": "file_input", "value": "image.png"},
+                {"id": 2, "type": "function", "value": "module.process"},
+                {"id": 3, "type": "output", "name": "result"},
+            ],
+            "edges": [
+                {"source": 1, "target": 2, "targetPort": "filename"},
+                {"source": 2, "target": 3, "sourcePort": None},
+            ],
+        }
+        wf = PythonWorkflowDefinitionWorkflow(**workflow_dict)
+        json_str = wf.dump_json()
+        reloaded_dict = PythonWorkflowDefinitionWorkflow.load_json_str(json_str)
+        reloaded_wf = PythonWorkflowDefinitionWorkflow(**reloaded_dict)
+        input_node = reloaded_wf.nodes[0]
+        self.assertIsInstance(input_node, PythonWorkflowDefinitionInputNode)
+        self.assertEqual(input_node.value, "image.png")
+
     def test_input_node_invalid_value_raises(self):
         bad_values = (
             {1: 2},

tests/test_purepython.py

@@ -1,3 +1,4 @@
+import sys
 import unittest
 from python_workflow_definition.purepython import load_workflow_json
 
@@ -35,6 +36,26 @@ def get_square(x):
   ]
 }"""
 
+echo_function_str = """
+def echo(filename):
+    return filename
+"""
+
+filename_workflow_str = """
+{
+  "version": "0.1.0",
+  "nodes": [
+    {"id": 0, "type": "function", "value": "echo_module.echo"},
+    {"id": 1, "type": "input", "value": "image.png", "name": "filename"},
+    {"id": 2, "type": "output", "name": "result"}
+  ],
+  "edges": [
+    {"target": 0, "targetPort": "filename", "source": 1, "sourcePort": null},
+    {"target": 2, "targetPort": null, "source": 0, "sourcePort": null}
+  ]
+}"""
+
+
 class TestPurePython(unittest.TestCase):
     def test_pure_python(self):
         with open("workflow.py", "w") as f:
@@ -44,3 +65,42 @@ def test_pure_python(self):
             f.write(workflow_str)
 
         self.assertEqual(load_workflow_json(file_name="workflow.json"), 6.25)
+
+    def test_purepython_filename_input(self):
+        """A filename string like 'image.png' must be passed through as a plain
+        string input, not interpreted as a Python module path or a float."""
+        with open("echo_module.py", "w") as f:
+            f.write(echo_function_str)
+        sys.modules.pop("echo_module", None)
+
+        with open("filename_workflow.json", "w") as f:
+            f.write(filename_workflow_str)
+
+        result = load_workflow_json(file_name="filename_workflow.json")
+        self.assertEqual(result, "image.png")
+
+    def test_purepython_filename_input_multiple_dots(self):
+        """Filenames with multiple dots (e.g. 'archive.tar.gz') must also be
+        treated as plain string inputs, not as nested module references."""
+        multi_dot_workflow_str = """
+{
+  "version": "0.1.0",
+  "nodes": [
+    {"id": 0, "type": "function", "value": "echo_module.echo"},
+    {"id": 1, "type": "input", "value": "archive.tar.gz", "name": "filename"},
+    {"id": 2, "type": "output", "name": "result"}
+  ],
+  "edges": [
+    {"target": 0, "targetPort": "filename", "source": 1, "sourcePort": null},
+    {"target": 2, "targetPort": null, "source": 0, "sourcePort": null}
+  ]
+}"""
+        with open("echo_module.py", "w") as f:
+            f.write(echo_function_str)
+        sys.modules.pop("echo_module", None)
+
+        with open("multi_dot_workflow.json", "w") as f:
+            f.write(multi_dot_workflow_str)
+
+        result = load_workflow_json(file_name="multi_dot_workflow.json")
+        self.assertEqual(result, "archive.tar.gz")

tests/test_pyiron_base.py

@@ -1,5 +1,6 @@
-import unittest
+import json
 import os
+import unittest
 from pyiron_base import job
 from python_workflow_definition.pyiron_base import load_workflow_json, write_workflow_json
 
@@ -16,6 +17,10 @@ def get_square(x):
     return x ** 2
 
 
+def echo(filename):
+    return filename
+
+
 class TestPyironBase(unittest.TestCase):
     def test_pyiron_base(self):
         workflow_json_filename = "pyiron_arithmetic.json"
@@ -31,3 +36,25 @@ def test_pyiron_base(self):
 
         self.assertTrue(os.path.exists(workflow_json_filename))
         self.assertEqual(delayed_object_lst[-1].pull(), 6.25)
+
+    def test_pyiron_base_filename_input(self):
+        """A filename string like 'image.png' must be passed through as a plain
+        string input, not interpreted as a Python module path or a float."""
+        workflow_json_filename = "pyiron_filename.json"
+        echo_job_wrapper = job(echo)
+        result_delayed = echo_job_wrapper(filename="image.png")
+
+        write_workflow_json(delayed_object=result_delayed, file_name=workflow_json_filename)
+        self.assertTrue(os.path.exists(workflow_json_filename))
+
+        with open(workflow_json_filename) as f:
+            saved = json.load(f)
+        input_values = [
+            n["value"]
+            for n in saved["nodes"]
+            if n["type"] == "input"
+        ]
+        self.assertIn("image.png", input_values)
+
+        delayed_object_lst = load_workflow_json(file_name=workflow_json_filename)
+        self.assertEqual(delayed_object_lst[-1].pull(), "image.png")

tests/test_pyiron_workflow.py

@@ -1,5 +1,7 @@
-import unittest
+import json
 import os
+import sys
+import unittest
 from pyiron_workflow import Workflow, to_function_node
 from python_workflow_definition.pyiron_workflow import load_workflow_json, write_workflow_json
 
@@ -16,6 +18,25 @@ def get_square(x):
     return x ** 2
 """
 
+echo_function_str = """
+def echo(filename):
+    return filename
+"""
+
+filename_workflow_str = """
+{
+  "version": "0.1.0",
+  "nodes": [
+    {"id": 0, "type": "function", "value": "echo_module.echo"},
+    {"id": 1, "type": "input", "value": "image.png", "name": "filename"},
+    {"id": 2, "type": "output", "name": "result"}
+  ],
+  "edges": [
+    {"target": 0, "targetPort": "filename", "source": 1, "sourcePort": null},
+    {"target": 2, "targetPort": null, "source": 0, "sourcePort": null}
+  ]
+}"""
+
 
 class TestPyironWorkflow(unittest.TestCase):
     def test_pyiron_workflow(self):
@@ -41,3 +62,58 @@ def test_pyiron_workflow(self):
         wf.run()
 
         self.assertTrue(os.path.exists(workflow_json_filename))
+
+    def test_pyiron_workflow_filename_input(self):
+        """A filename string like 'image.png' must be passed through as a plain
+        string input, not interpreted as a Python module path or a float."""
+        workflow_json_filename = "pyiron_workflow_filename.json"
+        with open("echo_module.py", "w") as f:
+            f.write(echo_function_str)
+        sys.modules.pop("echo_module", None)
+
+        with open(workflow_json_filename, "w") as f:
+            f.write(filename_workflow_str)
+
+        with open(workflow_json_filename) as f:
+            saved = json.load(f)
+        input_values = [
+            n["value"]
+            for n in saved["nodes"]
+            if n["type"] == "input"
+        ]
+        self.assertIn("image.png", input_values)
+
+        wf = load_workflow_json(file_name=workflow_json_filename)
+        wf.run()
+        self.assertTrue(os.path.exists(workflow_json_filename))
+
+    def test_pyiron_workflow_filename_input_programmatic(self):
+        """Write and round-trip a workflow with a filename input using the
+        programmatic write_workflow_json / load_workflow_json path."""
+        workflow_json_filename = "pyiron_workflow_filename_prog.json"
+        with open("echo_module.py", "w") as f:
+            f.write(echo_function_str)
+        sys.modules.pop("echo_module", None)
+
+        from echo_module import echo as _echo
+
+        echo_node = to_function_node("echo", _echo, "echo")
+        wf = Workflow("filename_workflow")
+        wf.filename = "image.png"
+        wf.result = echo_node(filename=wf.filename)
+        write_workflow_json(graph_as_dict=wf.graph_as_dict, file_name=workflow_json_filename)
+        self.assertTrue(os.path.exists(workflow_json_filename))
+
+        with open(workflow_json_filename) as f:
+            saved = json.load(f)
+        input_values = [
+            n["value"]
+            for n in saved["nodes"]
+            if n["type"] == "input"
+        ]
+        self.assertIn("image.png", input_values)
+
+        sys.modules.pop("echo_module", None)
+        wf2 = load_workflow_json(file_name=workflow_json_filename)
+        wf2.run()
+        self.assertTrue(os.path.exists(workflow_json_filename))