diff --git a/.github/workflows/pipeline.yml b/.github/workflows/pipeline.yml index 8c07cf8..cab65cd 100644 --- a/.github/workflows/pipeline.yml +++ b/.github/workflows/pipeline.yml @@ -159,3 +159,38 @@ jobs: jupyter-book build . --path-output public mv public/_build/html public_html touch public_html/.nojekyll + + unittest_matrix: + runs-on: ${{ matrix.operating-system }} + strategy: + matrix: + include: + - operating-system: macos-latest + python-version: '3.12' + + - operating-system: ubuntu-latest + python-version: '3.12' + + - operating-system: ubuntu-latest + python-version: '3.11' + + steps: + - uses: actions/checkout@v4 + - name: Setup environment + shell: bash -l {0} + run: | + echo -e "channels:\n - conda-forge\n" > .condarc + sed '/- python/d' binder/environment.yml > environment.yml + - name: Setup Mambaforge + uses: conda-incubator/setup-miniconda@v3 + with: + python-version: ${{ matrix.python-version }} + miniforge-version: latest + condarc-file: .condarc + environment-file: environment.yml + - name: Test + shell: bash -l {0} + timeout-minutes: 30 + run: | + pip install . --no-deps --no-build-isolation + python -m unittest discover tests diff --git a/tests/test_shared.py b/tests/test_shared.py new file mode 100644 index 0000000..db908e4 --- /dev/null +++ b/tests/test_shared.py @@ -0,0 +1,156 @@ +import unittest +from python_workflow_definition.shared import ( + get_dict, + get_list, + get_kwargs, + get_source_handles, + convert_nodes_list_to_dict, + update_node_names, + set_result_node, + remove_result, + EDGES_LABEL, + NODES_LABEL, + SOURCE_LABEL, + SOURCE_PORT_LABEL, + TARGET_LABEL, + TARGET_PORT_LABEL, +) + + +class TestShared(unittest.TestCase): + def test_get_dict(self): + self.assertEqual({"a": 1, "b": 2, "c": 3}, get_dict(a=1, b=2, c=3)) + + def test_get_list(self): + self.assertEqual([1, 2, 3], get_list(a=1, b=2, c=3)) + + def test_get_kwargs(self): + lst = [ + { + TARGET_PORT_LABEL: "a", + SOURCE_LABEL: "s1", + SOURCE_PORT_LABEL: "sp1", + }, + { + TARGET_PORT_LABEL: "b", + SOURCE_LABEL: "s2", + SOURCE_PORT_LABEL: "sp2", + }, + ] + self.assertEqual( + { + "a": {SOURCE_LABEL: "s1", SOURCE_PORT_LABEL: "sp1"}, + "b": {SOURCE_LABEL: "s2", SOURCE_PORT_LABEL: "sp2"}, + }, + get_kwargs(lst), + ) + + def test_get_kwargs_empty(self): + self.assertEqual({}, get_kwargs([])) + + def test_get_source_handles(self): + edges_lst = [ + {SOURCE_LABEL: "s1", SOURCE_PORT_LABEL: "sp1"}, + {SOURCE_LABEL: "s1", SOURCE_PORT_LABEL: "sp2"}, + {SOURCE_LABEL: "s2", SOURCE_PORT_LABEL: "sp3"}, + ] + self.assertEqual( + {"s1": ["sp1", "sp2"], "s2": ["sp3"]}, get_source_handles(edges_lst) + ) + + def test_get_source_handles_no_port(self): + edges_lst = [ + {SOURCE_LABEL: "s1", SOURCE_PORT_LABEL: None}, + {SOURCE_LABEL: "s1", SOURCE_PORT_LABEL: None}, + ] + self.assertEqual({"s1": [0, 1]}, get_source_handles(edges_lst)) + + def test_get_source_handles_empty(self): + self.assertEqual({}, get_source_handles([])) + + def test_convert_nodes_list_to_dict(self): + nodes_list = [ + {"id": 1, "name": "n1"}, + {"id": 2, "value": "v2"}, + {"id": 0, "name": "n0"}, + ] + self.assertEqual( + {"0": "n0", "1": "n1", "2": "v2"}, + convert_nodes_list_to_dict(nodes_list), + ) + + def test_convert_nodes_list_to_dict_empty(self): + self.assertEqual({}, convert_nodes_list_to_dict([])) + + def test_update_node_names(self): + workflow_dict = { + NODES_LABEL: [ + {"id": 0, "type": "input", "name": ""}, + {"id": 1, "type": "input", "name": ""}, + {"id": 2, "type": "input", "name": ""}, + {"id": 3, "type": "other", "name": ""}, + ], + EDGES_LABEL: [ + {SOURCE_LABEL: 0, TARGET_PORT_LABEL: "a"}, + {SOURCE_LABEL: 1, TARGET_PORT_LABEL: "b"}, + {SOURCE_LABEL: 2, TARGET_PORT_LABEL: "a"}, + ], + } + update_node_names(workflow_dict) + self.assertEqual("a_0", workflow_dict[NODES_LABEL][0]["name"]) + self.assertEqual("b", workflow_dict[NODES_LABEL][1]["name"]) + self.assertEqual("a_1", workflow_dict[NODES_LABEL][2]["name"]) + + def test_set_result_node(self): + workflow_dict = { + NODES_LABEL: [{"id": 0, "type": "input"}], + EDGES_LABEL: [], + } + set_result_node(workflow_dict) + self.assertEqual(2, len(workflow_dict[NODES_LABEL])) + self.assertEqual("output", workflow_dict[NODES_LABEL][1]["type"]) + self.assertEqual(1, len(workflow_dict[EDGES_LABEL])) + + def test_set_result_node_already_exists(self): + workflow_dict = { + NODES_LABEL: [ + {"id": 0, "type": "input"}, + {"id": 1, "type": "output"}, + ], + EDGES_LABEL: [{SOURCE_LABEL: 0, TARGET_LABEL: 1}], + } + set_result_node(workflow_dict) + self.assertEqual(3, len(workflow_dict[NODES_LABEL])) + self.assertEqual("output", workflow_dict[NODES_LABEL][2]["type"]) + self.assertEqual(2, len(workflow_dict[EDGES_LABEL])) + + def test_set_result_node_no_end_node(self): + workflow_dict = { + NODES_LABEL: [{"id": 0}, {"id": 1}], + EDGES_LABEL: [ + {SOURCE_LABEL: 0, TARGET_LABEL: 1}, + {SOURCE_LABEL: 1, TARGET_LABEL: 0}, + ], + } + with self.assertRaises(IndexError): + set_result_node(workflow_dict) + + def test_remove_result(self): + workflow_dict = { + NODES_LABEL: [ + {"id": 0, "type": "input"}, + {"id": 1, "type": "output"}, + ], + EDGES_LABEL: [{SOURCE_LABEL: 0, TARGET_LABEL: 1}], + } + new_workflow = remove_result(workflow_dict) + self.assertEqual(1, len(new_workflow[NODES_LABEL])) + self.assertEqual(0, len(new_workflow[EDGES_LABEL])) + + def test_remove_result_no_output(self): + workflow_dict = { + NODES_LABEL: [{"id": 0, "type": "input"}], + EDGES_LABEL: [], + } + with self.assertRaises(IndexError): + remove_result(workflow_dict)