Synsation story runner

Problem:
Convert synsation samples to YAML story runner.

Solution:
New functest test_runner_synsation uses new locations
creator and types of assets. The story is creted via Jinja2
template....

Signed-off-by: Paul Hewlett <phewlett76@gmail.com>

Author: eccles

archivist/assets.py

@@ -193,10 +193,12 @@ def create_if_not_exists(
                   arc_serial_number: das-j1-01
                   arc_description: Electronic door entry system to Jitsuin France
                   wavestone_asset_id: paris.france.jitsuin.das
+                location:
+                  identity: locations/xxxxxxxxxxxxxxxxxxxxxxxxxx
                 location:
                   selector:
                     - display_name
-                  display_name: Jitsuin Paris,
+                  display_name: Jitsuin Paris
                   description: Sales and sales support for the French region
                   latitude: 48.8339211,
                   longitude: 2.371345,
@@ -213,6 +215,9 @@ def create_if_not_exists(
             If 'location' is specified then the 'selector' value is required and is used as a
             secondary key. Likewise the secondary key must exist in the attributes of the location.
 
+            Alternatively the identity of the location is specified - both
+            are shown - choose one.
+
         Returns:
             tuple of :class:`Asset` instance, Boolean is True if asset already existed
 
@@ -239,10 +244,13 @@ def create_if_not_exists(
 
         # is location present?
         if location is not None:
-            loc, _ = self._archivist.locations.create_if_not_exists(
-                location,
-            )
-            data["attributes"]["arc_home_location_identity"] = loc["identity"]
+            if "identity" in location:
+                data["attributes"]["arc_home_location_identity"] = location["identity"]
+            else:
+                loc, _ = self._archivist.locations.create_if_not_exists(
+                    location,
+                )
+                data["attributes"]["arc_home_location_identity"] = loc["identity"]
 
         # any attachments ?
         if attachments is not None:

archivist/cmds/template/__init__.py

@@ -0,0 +1,2 @@
+"""Archivist runner
+"""

archivist/cmds/template/__main__.py

@@ -0,0 +1,9 @@
+# pylint:  disable=missing-docstring
+
+
+from .main import main
+
+
+if __name__ == "__main__":
+    # execute only if run as a script
+    main()

archivist/cmds/template/main.py

@@ -0,0 +1,31 @@
+# pylint:  disable=missing-docstring
+
+from logging import getLogger
+from sys import exit as sys_exit
+from sys import stdout as sys_stdout
+
+from ...parser import common_parser, endpoint
+
+from .run import run
+
+LOGGER = getLogger(__name__)
+
+
+def main():
+    parser = common_parser("Executes the archivist runner from a template file")
+
+    parser.add_argument(
+        "values",
+        help="the values file describing the data to be injected into the template",
+    )
+    parser.add_argument(
+        "template", help="the template file describing the operations to conduct"
+    )
+    args = parser.parse_args()
+
+    arch = endpoint(args)
+
+    run(arch, args)
+
+    parser.print_help(sys_stdout)
+    sys_exit(1)

archivist/cmds/template/run.py

@@ -0,0 +1,54 @@
+# pylint:  disable=missing-docstring
+
+from logging import getLogger
+from os import environ
+from pathlib import PurePath
+from sys import exit as sys_exit
+
+from jinja2 import Environment, FileSystemLoader
+import yaml
+
+from ... import about
+
+# pylint:disable=unused-import      # To prevent cyclical import errors forward referencing is used
+# pylint:disable=cyclic-import      # but pylint doesn't understand this feature
+from ... import archivist as type_helper
+
+
+LOGGER = getLogger(__name__)
+
+
+def run(arch: "type_helper.Archivist", args):
+
+    LOGGER.info("Using version %s of jitsuin-archivist", about.__version__)
+    LOGGER.info("Namespace %s", args.namespace)
+
+    path = PurePath(args.template)
+    jinja = Environment(
+        loader=FileSystemLoader(path.parent),
+        trim_blocks=True,
+        lstrip_blocks=True,
+    )
+    template = jinja.get_template(path.name)
+
+    # if namespace is specified on the commandline then override any environment
+    # setting...
+    if args.namespace:
+        environ["ARCHIVIST_NAMESPACE"] = args.namespace
+
+    # environment is injected into the template
+    with open(args.values, "r", encoding="utf-8") as fd:
+        arch.runner(
+            yaml.load(
+                template.render(
+                    yaml.load(
+                        fd,
+                        Loader=yaml.SafeLoader,
+                    ),
+                    env=environ,
+                ),
+                Loader=yaml.SafeLoader,
+            ),
+        )
+
+    sys_exit(0)

archivist/events.py

@@ -159,7 +159,7 @@ def create_from_data(self, asset_id: str, data: Dict, *, confirm=False) -> Event
         Args:
             asset_id (str): asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx
             data (dict): request body of event.
-            confirm (bool): if True wait for event to be confirmed on DLT.
+            confirm (bool): if True wait for event to be confirmed.
 
         Returns:
             :class:`Event` instance
@@ -171,10 +171,13 @@ def create_from_data(self, asset_id: str, data: Dict, *, confirm=False) -> Event
         # is location present?
         location = data.pop("location", None)
         if location is not None:
-            loc, _ = self._archivist.locations.create_if_not_exists(
-                location,
-            )
-            event_attributes["arc_location_identity"] = loc["identity"]
+            if "identity" in location:
+                data["event_attributes"]["arc_location_identity"] = location["identity"]
+            else:
+                loc, _ = self._archivist.locations.create_if_not_exists(
+                    location,
+                )
+                event_attributes["arc_location_identity"] = loc["identity"]
 
         sbom = data.pop("sbom", None)
         if sbom is not None:

archivist/locations.py

@@ -34,6 +34,7 @@
 from .dictmerge import _deepmerge
 from .errors import ArchivistNotFoundError
 from .utils import selector_signature
+from .type_aliases import NoneOnError
 
 
 LOGGER = getLogger(__name__)
@@ -46,6 +47,17 @@ class Location(dict):
 
     """
 
+    @property
+    def name(self) -> NoneOnError[str]:
+        """str: name of the location"""
+        name = None
+        try:
+            name = self["display_name"]
+        except KeyError:
+            pass
+
+        return name
+
 
 class _LocationsClient:
     """LocationsClient

archivist/parser.py

@@ -12,8 +12,8 @@
 from warnings import filterwarnings
 
 from .archivist import Archivist
-from .logger import set_logger
 from .dictmerge import _deepmerge
+from .logger import set_logger
 from .proof_mechanism import ProofMechanism
 from .utils import get_auth
 
@@ -74,15 +74,15 @@ def common_parser(description: str):
         dest="url",
         action="store",
         default="https://rkvst.poc.jitsuin.io",
-        help="location of Archivist service",
+        help="url of Archivist service",
     )
     parser.add_argument(
         "-p",
         "--proof-mechanism",
         type=ProofMechanism,
         action=EnumAction,
         dest="proof_mechanism",
-        default=ProofMechanism.SIMPLE_HASH,
+        default=None,
         help="mechanism for proving the evidence for events on the Asset",
     )
     parser.add_argument(
@@ -132,11 +132,14 @@ def endpoint(args):
 
     arch = None
     LOGGER.info("Initialising connection to Jitsuin Archivist...")
-    fixtures = {
-        "assets": {
-            "proof_mechanism": args.proof_mechanism.name,
-        },
-    }
+    fixtures = {}
+    if args.proof_mechanism is not None:
+        fixtures = {
+            "assets": {
+                "proof_mechanism": args.proof_mechanism.name,
+            },
+        }
+
     if args.namespace is not None:
         fixtures = _deepmerge(
             fixtures,