Merge pull request #206 from ICRAR/liu-304

Liu 304

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

@@ -37,7 +37,7 @@ jobs:
       - name: Install test dependencies
         run: |
           pip install -U coveralls pytest pytest-cov
-          pip install -U setuptools pip wheel dask dlg-example-cmpts
+          pip install -U setuptools pip wheel dask
 
       - name: Install daliuge-common
         run: pip install -e daliuge-common/

OpenAPI/composite_manager.yaml

@@ -35,7 +35,7 @@ paths:
                   type: string
       operationId: getCMNodes
       description: Returns the list of nodes covered by this Composite Manager
-  '/api/nodes/{node}':
+  '/api/node/{node}':
     parameters:
       - schema:
           type: string
@@ -58,7 +58,7 @@ paths:
         '200':
           description: OK
       description: "Add `node` to this Composite Manager's nodes list"
-  '/api/nodes/{node}/sessions':
+  '/api/node/{node}/sessions':
     parameters:
       - schema:
           type: string
@@ -103,7 +103,7 @@ paths:
                 description: Generic session information type
       operationId: getNodeSessions
       description: 'Get the list of sessions on `node`, managed by this Manager'
-  '/api/nodes/{node}/sessions/{sessionId}':
+  '/api/node/{node}/sessions/{sessionId}':
     parameters:
       - schema:
           type: string
@@ -127,7 +127,7 @@ paths:
                 $ref: './manager_common.yaml#/components/schemas/session'
       operationId: getNodeSessionInformation
       description: Returns the main information of session `sessionId` in node `node`
-  '/api/nodes/{node}/sessions/{sessionId}/status':
+  '/api/node/{node}/sessions/{sessionId}/status':
     parameters:
       - schema:
           type: string
@@ -151,7 +151,7 @@ paths:
                 type: integer
       operationId: getNodeSessionStatus
       description: Returns the status of `sessionId` in `node`
-  '/api/nodes/{node}/sessions/{sessionId}/graph':
+  '/api/node/{node}/sessions/{sessionId}/graph':
     parameters:
       - schema:
           type: string
@@ -175,7 +175,7 @@ paths:
                 $ref: './manager_common.yaml#/components/schemas/physical_graph'
       operationId: getNodeGraph
       description: Returns the physical graph of session `sessionId` in `node`
-  '/api/nodes/{node}/sessions/{sessionId}/graph/status':
+  '/api/node/{node}/sessions/{sessionId}/graph/status':
     parameters:
       - schema:
           type: string

OpenAPI/tests/managers_test_client.py

@@ -3,9 +3,9 @@
 from composite_manager_client.api.default_api import DefaultApi
 
 nm_config = nmc.Configuration()
-nm_config.host = "127.0.0.1:8000"
+nm_config.host = "localhost:8000"
 dim_config = cmc.Configuration()
-dim_config.host = "127.0.0.1:8001"
+dim_config.host = "localhost:8001"
 
 with nmc.ApiClient(nm_config) as nm_client, cmc.ApiClient(dim_config) as dim_client:
     dim = DefaultApi(dim_client)

OpenAPI/tests/test_managers_openapi.sh

@@ -30,7 +30,7 @@ rm config.properties
 tmpdir=`mktemp -d`
 dlg nm -w . -v --no-dlm -l $tmpdir -w $tmpdir &
 nm_pid=$!
-dlg dim -v -N 127.0.0.1 -l $tmpdir -w $tmpdir &
+dlg dim -v -N localhost -l $tmpdir -w $tmpdir &
 dim_pid=$!
 dlg lgweb -d $tmpdir -t $tmpdir -v &
 lgweb_pid=$!

OpenAPI/tests/translator_test_client.py

@@ -3,7 +3,7 @@
 import translator_client as tc
 
 translator_config = tc.Configuration()
-translator_config.host = "127.0.0.1:8084"
+translator_config.host = "localhost:8084"
 
 with open(sys.argv[1], "rt") as f:
     graph = f.read()

daliuge-common/dlg/clients.py

@@ -247,10 +247,10 @@ def nodes(self):
         return self._get_json("/nodes")
 
     def add_node(self, node):
-        self._POST(f"/nodes/{node}", content=None)
+        self._POST(f"/node/{node}", content=None)
 
     def remove_node(self, node):
-        self._DELETE(f"/nodes/{node}")
+        self._DELETE(f"/node/{node}")
 
 
 class DataIslandManagerClient(CompositeManagerClient):
@@ -278,27 +278,27 @@ def __init__(
 
     def create_island(self, island_host, nodes):
         self._post_json(
-            f"/managers/{quote(island_host)}/dataisland", {"nodes": nodes}
+            f"/managers/{quote(island_host)}/island", {"nodes": nodes}
         )
 
     def dims(self):
         return self._get_json("/islands")
 
     def add_dim(self, dim):
-        self._POST(f"/islands/{dim}", content=None)
+        self._POST(f"/island/{dim}", content=None)
 
     def remove_dim(self, dim):
-        self._DELETE(f"/islands/{dim}")
+        self._DELETE(f"/island/{dim}")
 
     def add_node_to_dim(self, dim, nm):
         """
         Adds a nm to a dim
         """
         self._POST(
-            f"managers/{dim}/nodes/{nm}", content=None, )
+            f"/managers/{dim}/node/{nm}", content=None, )
 
     def remove_node_from_dim(self, dim, nm):
         """
         Removes a nm from a dim
         """
-        self._DELETE(f"managers/{dim}/nodes/{nm}")
+        self._DELETE(f"/managers/{dim}/node/{nm}")

daliuge-common/dlg/common/__init__.py

@@ -75,6 +75,7 @@ class Categories:
     Categories.PLASMAFLIGHT,
     Categories.PARSET,
     Categories.ENVIRONMENTVARS,
+    Categories.S3,
 }
 APP_DROP_TYPES = [
     Categories.COMPONENT,
@@ -90,12 +91,25 @@ class Categories:
 
 
 class DropType:
-    PLAIN = "plain"
+    # this gives the mapping to fields containing class paths
+    DATA = "data" # TODO: need to drop this one
+    DATACLASS = "dataclass"
+    APPCLASS = "appclass"
     SOCKET = "socket"
-    APP = "app"  # Application drop that terminates onces executed
+    APP = "app"  # TODO: need to drop this one
     SERVICE_APP = "serviceapp"  # App drop that runs continously
     CONTAINER = "container"  # Drop that contains other drops
 
+class CategoryType:
+    DATA = "Data"
+    APPLICATION = "Application"
+    GROUP = "Group"
+    UNKNOWN = "Unknown"
+    SERVICE = "Service"
+    CONTAINER = "Container"
+    SOCKET = "Socket"
+    CONTROL = "Control"
+    OTHER = "Other"
 
 def b2s(b, enc="utf8"):
     "Converts bytes into a string"
@@ -200,7 +214,7 @@ def get_roots(pg_spec):
             if dropspec.get("outputs", None):
                 do = _sanitize_links(dropspec["outputs"])
                 nonroots |= set(do)
-        elif dropspec["type"] == DropType.PLAIN:
+        elif dropspec["type"] == DropType.DATA:
             if dropspec.get("producers", None):
                 nonroots.add(oid)
             if dropspec.get("consumers", None):
@@ -245,7 +259,7 @@ def get_leaves(pg_spec):
             if dropspec.get("inputs", None):
                 di = _sanitize_links(dropspec["inputs"])
                 nonleaves |= set(di)
-        elif dropspec["type"] == DropType.PLAIN:
+        elif dropspec["type"] == DropType.DATA:
             if dropspec.get("producers", None):
                 dp = _sanitize_links(dropspec["producers"])
                 nonleaves |= set(dp)

daliuge-common/dlg/common/reproducibility/reproducibility_fields.py

@@ -204,9 +204,9 @@ def pgt_unroll_block_fields(category_type, rmode: ReproducibilityFlags):
     if rmode != ReproducibilityFlags.NOTHING:
         data["type"] = FieldOps.STORE
     if rmode != ReproducibilityFlags.REPRODUCE:
-        if category_type != "plain":
+        if category_type != "data":
             data["dt"] = FieldOps.STORE
-    if category_type == "plain":
+    if category_type == "data":
         data["storage"] = FieldOps.STORE
     if rmode in (ReproducibilityFlags.RECOMPUTE, ReproducibilityFlags.REPLICATE_COMP):
         data["rank"] = FieldOps.STORE

daliuge-common/dlg/exceptions.py

@@ -76,6 +76,18 @@ class InvalidGraphException(DaliugeException):
     Daliuge.
     """
 
+class DropChecksumException(DaliugeException):
+    """
+    An exception thrown when a DROP does not pass a checksum check
+    """
+    def __init__(self, drop):
+        DaliugeException.__init__(self, drop)
+        self.drop = drop
+        self.msg = "DropChecksumException <%r> checksum: %s" % (self.drop, drop.checksum)
+
+    def __str__(self, *args, **kwargs):
+        return self.msg
+
 
 class NoDropException(DaliugeException):
     """

daliuge-engine/dlg/apps/dynlib.py

@@ -125,8 +125,8 @@ def _to_c_output(o):
     Convert an output drop into its corresponding C structure
     """
 
-    def _write(_o, buf, n):
-        return _o.write(buf[:n])
+    def _write(_o, buf, n, **kwargs):
+        return _o.write(buf[:n], **kwargs)
 
     w = _write_cb_type(functools.partial(_write, o))
     return CDlgOutput(