Merge pull request #287 from ICRAR/LIU-407

LIU-407: Use EAGLE_test_repo as the basis for test graphs

.github/workflows/run-unit-tests.yml

@@ -43,11 +43,11 @@ jobs:
 
       - name: Install daliuge-translator
         if: ${{ matrix.translator == 'yes' }}
-        run: pip install -e daliuge-translator/
+        run: pip install -e "daliuge-translator/[test]"
 
       - name: Install daliuge-engine
         if: ${{ matrix.engine == 'yes' }}
-        run: pip install -e daliuge-engine/
+        run: pip install -e "daliuge-engine/[test]"
 
       - name: Run daliuge-translator tests
         if: ${{ matrix.translator == 'yes' }}

daliuge-engine/dlg/apps/app_base.py

@@ -200,9 +200,9 @@ def _generateNamedPorts(self, ports):
         named_ports: OrderedDict[str, DataDROP] = OrderedDict()
         port_dict = self.__getattribute__(f"_{ports}")
         if (
-                ports in self.parameters
-                and len(self.parameters[ports]) > 0
-                and isinstance(self.parameters[ports][0], dict)
+            ports in self.parameters
+            and len(self.parameters[ports]) > 0
+            and isinstance(self.parameters[ports][0], dict)
         ):
             for i in range(len(port_dict)):
                 key = list(self.parameters[ports][i].values())[0]
@@ -215,9 +215,9 @@ def _generateNamedPorts(self, ports):
                     else:
                         named_ports[key] = [named_ports[key], value]
         elif (
-                ports in self.parameters
-                and len(self.parameters[ports]) > 0
-                and isinstance(self.parameters[ports], list)
+            ports in self.parameters
+            and len(self.parameters[ports]) > 0
+            and isinstance(self.parameters[ports], list)
         ):
             # This enablkes the gather to work
             return {}

daliuge-engine/setup.py

@@ -167,12 +167,16 @@ def run(self):
     "MPI": ["mpi4py"],
     # AWS storage types
     "aws": ["boto3"],
+    "test": [
+        "pytest",
+        "eagle_test_graphs @ git+https://github.com/ICRAR/EAGLE_test_repo",
+    ],
 }
 
 setup(
     name="daliuge-engine",
     version=get_version_info()[0],
-    description="Data Activated \uF9CA (flow) Graph Engine - Execution Engine",
+    description="Data Activated \uf9ca (flow) Graph Engine - Execution Engine",
     long_description="""
         The element of the DALiuGE system executing the workflows. This replaces
         the former 'runtime' package (up to version 1.0). For more information 

daliuge-engine/test/apps/test_bash.py

@@ -78,9 +78,7 @@ def assert_message_is_correct(message, command):
             a.addOutput(b)
             with DROPWaiterCtx(self, b, 100):
                 a.async_execute()
-            self.assertEqual(
-                message.encode("utf8"), droputils.allDropContents(b)
-            )
+            self.assertEqual(message.encode("utf8"), droputils.allDropContents(b))
 
         msg = "This is a message with a single quote: '"
         assert_message_is_correct(msg, 'echo -n "{0}" > %o0'.format(msg))
@@ -106,9 +104,7 @@ def assert_envvar_is_there(varname, value):
             a.addOutput(b)
             with DROPWaiterCtx(self, b, 100):
                 a.async_execute()
-            self.assertEqual(
-                value.encode("utf8"), droputils.allDropContents(b)
-            )
+            self.assertEqual(value.encode("utf8"), droputils.allDropContents(b))
 
         assert_envvar_is_there("DLG_UID", app_uid)
         assert_envvar_is_there("DLG_SESSION_ID", session_id)
@@ -132,9 +128,7 @@ def test_reproducibility(self):
         a.reproducibility_level = ReproducibilityFlags.RECOMPUTE
         a.commit()
         self.assertNotEqual(a.merkleroot, b.merkleroot)
-        self.assertEqual(
-            a.generate_merkle_data(), {"command": "echo 'Hello world'"}
-        )
+        self.assertEqual(a.generate_merkle_data(), {"command": "echo 'Hello world'"})
 
         a.reproducibility_level = ReproducibilityFlags.REPRODUCE
         a.commit()
@@ -178,9 +172,7 @@ def test_single_pipe(self):
 
         output_fname = tempfile.mktemp()
 
-        a = StreamingOutputBashApp(
-            "a", "a", command=r"echo -en '5\n4\n3\n2\n1'"
-        )
+        a = StreamingOutputBashApp("a", "a", command=r"echo -en '5\n4\n3\n2\n1'")
         b = InMemoryDROP("b", "b")
         c = StreamingInputBashApp("c", "c", command="cat > %o0")
         d = FileDROP("d", "d", filepath=output_fname)
@@ -229,9 +221,7 @@ def test_two_simultaneous_pipes(self):
 
         output_fname = tempfile.mktemp()
 
-        a = StreamingOutputBashApp(
-            "a", "a", command=r"echo -en '5\n4\n3\n2\n1'"
-        )
+        a = StreamingOutputBashApp("a", "a", command=r"echo -en '5\n4\n3\n2\n1'")
         b = InMemoryDROP("b", "b")
         c = StreamingInputOutputBashApp("c", "c", command="cat")
         d = InMemoryDROP("d", "d")
@@ -253,10 +243,7 @@ def test_two_simultaneous_pipes(self):
             self.assertEqual(DROPStates.COMPLETED, drop.status)
         self.assertEqual(
             [1, 2, 3, 4, 5],
-            [
-                int(x)
-                for x in droputils.allDropContents(f).strip().split(b"\n")
-            ],
+            [int(x) for x in droputils.allDropContents(f).strip().split(b"\n")],
         )
 
         # Clean up and go