Merge 35a58a9 into 8f13053

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

@@ -3,24 +3,23 @@ name: Run unit tests
 on: [push, pull_request]
 
 jobs:
-
   run_tests:
     name: Run unit tests with python ${{matrix.python-version}} - ${{ matrix.desc }}
     runs-on: ubuntu-20.04
     strategy:
       matrix:
         include:
-          - python-version: '3.8'
+          - python-version: "3.8.10"
             test_number: 0
             engine: no
             translator: yes
             desc: "no engine"
-          - python-version: '3.8'
+          - python-version: "3.8.10"
             test_number: 1
             desc: "no translator"
             engine: yes
             translator: no
-          - python-version: '3.9'
+          - python-version: "3.9"
             test_number: 2
             desc: "full package"
             engine: yes
@@ -81,8 +80,8 @@ jobs:
     needs: run_tests
     runs-on: ubuntu-20.04
     steps:
-    - name: Coveralls Finished
-      uses: coverallsapp/github-action@master
-      with:
-        github-token: ${{ secrets.GITHUB_TOKEN }}
-        parallel-finished: true
+      - name: Coveralls Finished
+        uses: coverallsapp/github-action@master
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          parallel-finished: true

daliuge-common/dlg/common/__init__.py

@@ -40,6 +40,7 @@ class DropType:
 class CategoryType:
     DATA = "Data"
     APPLICATION = "Application"
+    CONSTRUCT = "Construct"
     GROUP = "Group"
     UNKNOWN = "Unknown"
     SERVICE = "Service"
@@ -81,7 +82,12 @@ def _addSomething(self, other, key, IdText=None):
         if key not in self:
             self[key] = []
         if other["oid"] not in self[key]:
+            # TODO: Returning just the other drop OID instead of the named
+            #       port list is not a good solution. Required for the dask
+            #       tests.
             append = {other["oid"]: IdText} if IdText else other["oid"]
+            # if IdText is None:
+            # raise ValueError
             self[key].append(append)
 
     def addConsumer(self, other, IdText=None):

daliuge-engine/dlg/utils.py → daliuge-common/dlg/utils.py

@@ -50,7 +50,9 @@ def timed_import(module_name):
     """Imports `module_name` and log how long it took to import it"""
     start = time.time()
     module = importlib.import_module(module_name)
-    logger.info("Imported %s in %.3f seconds", module_name, time.time() - start)
+    logger.info(
+        "Imported %s in %.3f seconds", module_name, time.time() - start
+    )
     return module
 
 
@@ -61,7 +63,9 @@ def get_local_ip_addr():
     PROTO = netifaces.AF_INET
     ifaces = netifaces.interfaces()
     if_addrs = [(netifaces.ifaddresses(iface), iface) for iface in ifaces]
-    if_inet_addrs = [(tup[0][PROTO], tup[1]) for tup in if_addrs if PROTO in tup[0]]
+    if_inet_addrs = [
+        (tup[0][PROTO], tup[1]) for tup in if_addrs if PROTO in tup[0]
+    ]
     iface_addrs = [
         (s["addr"], tup[1])
         for tup in if_inet_addrs
@@ -84,7 +88,9 @@ def get_all_ipv4_addresses():
     ]
 
 
-def register_service(zc, service_type_name, service_name, ipaddr, port, protocol="tcp"):
+def register_service(
+    zc, service_type_name, service_name, ipaddr, port, protocol="tcp"
+):
     """
     ZeroConf: Register service type, protocol, ipaddr and port
 
@@ -349,6 +355,8 @@ def _wrapper(*args, **kwargs):
 
             setattr(current_object, name, args[0])
             try:
+                if "self" in kwargs:
+                    kwargs.pop("self")
                 return f(*args, **kwargs)
             finally:
                 setattr(current_object, name, previous)
@@ -380,7 +388,6 @@ def __init__(self, content):
         self.buflen = 0
 
     def readall(self):
-
         if not self.decompressor:
             return b""
 
@@ -397,14 +404,15 @@ def readall(self):
             if not decompressed:
                 break
             response.write(decompressed)
-            to_decompress = decompressor.unconsumed_tail + content.read(blocksize)
+            to_decompress = decompressor.unconsumed_tail + content.read(
+                blocksize
+            )
 
         response.write(decompressor.flush())
         self.decompressor = None
         return response.getvalue()
 
     def read(self, n=-1):
-
         if n == -1:
             return self.readall()
 
@@ -431,7 +439,6 @@ def read(self, n=-1):
             return b"".join(response)
 
         while True:
-
             # We hope that reading n compressed bytes will yield n uncompressed
             # bytes at least; we loop anyway until we read n uncompressed bytes
             compressed = self.content.read(n)
@@ -511,13 +518,17 @@ def prepareUser(DLG_ROOT=getDlgDir()):
     gr = grp.getgrgid(pw.pw_gid)
     dgr = grp.getgrnam("docker")
     with open(os.path.join(workdir, "passwd"), "wt") as file:
-        file.write(open(os.path.join(template_dir, "passwd.template"), "rt").read())
+        file.write(
+            open(os.path.join(template_dir, "passwd.template"), "rt").read()
+        )
         file.write(
             f"{pw.pw_name}:x:{pw.pw_uid}:{pw.pw_gid}:{pw.pw_gecos}:{DLG_ROOT}:/bin/bash\n"
         )
         logger.debug(f"passwd file written {file.name}")
     with open(os.path.join(workdir, "group"), "wt") as file:
-        file.write(open(os.path.join(template_dir, "group.template"), "rt").read())
+        file.write(
+            open(os.path.join(template_dir, "group.template"), "rt").read()
+        )
         file.write(f"{gr.gr_name}:x:{gr.gr_gid}:\n")
         file.write(f"docker:x:{dgr.gr_gid}\n")
         logger.debug(f"Group file written {file.name}")

daliuge-engine/dlg/apps/app_base.py

@@ -452,7 +452,7 @@ def execute(self, _send_notifications=True):
                     return
                 tries += 1
                 logger.exception(
-                    "Error while executing %r (try %d/%d)",
+                    "Error while executing %r (try %s/%s)",
                     self,
                     tries,
                     self.n_tries,