diff --git a/tests/test_executorlib.py b/tests/test_executorlib.py index 9787d87..b34a771 100644 --- a/tests/test_executorlib.py +++ b/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") diff --git a/tests/test_jobflow.py b/tests/test_jobflow.py index 12f9433..7139c94 100644 --- a/tests/test_jobflow.py +++ b/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") diff --git a/tests/test_models.py b/tests/test_models.py index d353f46..61eb356 100644 --- a/tests/test_models.py +++ b/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}, diff --git a/tests/test_purepython.py b/tests/test_purepython.py index 1d3944c..096d49a 100644 --- a/tests/test_purepython.py +++ b/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") diff --git a/tests/test_pyiron_base.py b/tests/test_pyiron_base.py index 4be8589..0398f7a 100644 --- a/tests/test_pyiron_base.py +++ b/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") diff --git a/tests/test_pyiron_workflow.py b/tests/test_pyiron_workflow.py index cbeb836..5acaf1c 100644 --- a/tests/test_pyiron_workflow.py +++ b/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))