Add tests for workflow frameworks

Author: jan-janssen

.github/workflows/pipeline.yml

@@ -216,6 +216,7 @@ jobs:
       shell: bash -l {0}
       timeout-minutes: 30
       run: |
+        verdi presto --profile-name pwd
         pip install . --no-deps --no-build-isolation
         coverage run
         coverage xml

tests/test_aiida.py

@@ -0,0 +1,44 @@
+import unittest
+import os
+from aiida_workgraph import WorkGraph, task
+from aiida import orm, load_profile
+load_profile()
+
+from python_workflow_definition.aiida import load_workflow_json, write_workflow_json
+
+
+def get_prod_and_div(x, y):
+    return {"prod": x * y, "div": x / y}
+
+
+def get_sum(x, y):
+    return x + y
+
+
+def get_square(x):
+    return x ** 2
+
+
+class TestAiiDA(unittest.TestCase):
+    def test_aiida(self):
+        workflow_json_filename = "aiida_simple.json"
+        wg = WorkGraph("arithmetic")
+        get_prod_and_div_task = wg.add_task(
+            task(outputs=['prod', 'div'])(get_prod_and_div),
+            x=orm.Float(1),
+            y=orm.Float(2),
+        )
+        get_sum_task = wg.add_task(
+            get_sum,
+            x=get_prod_and_div_task.outputs.prod,
+            y=get_prod_and_div_task.outputs.div,
+        )
+        get_square_task = wg.add_task(
+            get_square,
+            x=get_sum_task.outputs.result,
+        )
+        write_workflow_json(wg=wg, file_name=workflow_json_filename)
+        workgraph = load_workflow_json(file_name='workflow.json')
+        workgraph.run()
+
+        self.assertTrue(os.path.exists(workflow_json_filename))

tests/test_jobflow.py

@@ -0,0 +1,35 @@
+import unittest
+import os
+from jobflow import job, Flow
+from jobflow.managers.local import run_locally
+from python_workflow_definition.jobflow import load_workflow_json, write_workflow_json
+
+
+def get_prod_and_div(x, y):
+    return {"prod": x * y, "div": x / y}
+
+
+def get_sum(x, y):
+    return x + y
+
+
+def get_square(x):
+    return x ** 2
+
+
+class TestJobflow(unittest.TestCase):
+    def test_jobflow(self):
+        workflow_json_filename = "jobflow_simple.json"
+        get_sum_job = job(get_sum)
+        get_prod_and_div_job = job(get_prod_and_div)
+        get_square_job = job(get_square)
+        prod_and_div = get_prod_and_div_job(x=1, y=2)
+        tmp_sum = get_sum_job(x=prod_and_div.output.prod, y=prod_and_div.output.div)
+        result = get_square_job(x=tmp_sum.output)
+        flow = Flow([prod_and_div, tmp_sum, result])
+        write_workflow_json(flow=flow, file_name=workflow_json_filename)
+        flow = load_workflow_json(file_name=workflow_json_filename)
+        result = run_locally(flow)
+
+        self.assertTrue(os.path.exists(workflow_json_filename))
+        self.assertEqual(result[result.keys()[-1]][1].output, 6.25)

tests/test_pyiron_base.py

@@ -0,0 +1,33 @@
+import unittest
+import os
+from pyiron_base import job
+from python_workflow_definition.pyiron_base import load_workflow_json, write_workflow_json
+
+
+def get_prod_and_div(x, y):
+    return {"prod": x * y, "div": x / y}
+
+
+def get_sum(x, y):
+    return x + y
+
+
+def get_square(x):
+    return x ** 2
+
+
+class TestPyironBase(unittest.TestCase):
+    def test_pyiron_base(self):
+        workflow_json_filename = "pyiron_arithmetic.json"
+        get_sum_job_wrapper = job(get_sum)
+        get_prod_and_div_job_wrapper = job(get_prod_and_div, output_key_lst=["prod", "div"])
+        get_square_job_wrapper = job(get_square)
+
+        prod_and_div = get_prod_and_div_job_wrapper(x=1, y=2)
+        tmp_sum = get_sum_job_wrapper(x=prod_and_div.output.prod, y=prod_and_div.output.div)
+        result = get_square_job_wrapper(x=tmp_sum)
+        write_workflow_json(delayed_object=result, file_name=workflow_json_filename)
+        delayed_object_lst = load_workflow_json(file_name=workflow_json_filename)
+
+        self.assertTrue(os.path.exists(workflow_json_filename))
+        self.assertEqual(delayed_object_lst[-1].pull(), 6.25)