Expand TRPAttributes

README.md

@@ -97,9 +97,9 @@ $ tro-utils --declaration sample_tro.jsonld performance add \
   -m "My magic workflow" \
   -s 2024-03-01T09:22:01 \
   -e 2024-03-02T10:00:11 \
-  -c trov:InternetIsolation \
-  -c trov:InternetAccessRecording \
-  -a arrangement/0 \
+  -a trov:InternetIsolation \
+  -a trov:InternetAccessRecording \
+  -A arrangement/0 \
   -M arrangement/1
  $ tro-utils --declaration sample_tro.jsonld sign
  $ tro-utils --declaration sample_tro.jsonld verify

pyproject.toml

@@ -13,7 +13,7 @@ maintainers = [
 ]
 description = "Utilities for creating, editing and interacting with TROs"
 readme = "README.rst"
-version = "0.1.3"
+version = "0.2.0"
 dependencies = [
   "Click>=7.0",
   "jinja2",

tests/test_tro_utils.py

@@ -306,7 +306,7 @@ def test_add_performance(self, temp_workspace, tmp_path, gpg_setup, trs_profile)
             comment="Data processing workflow",
             accessed_arrangement="arrangement/0",
             modified_arrangement="arrangement/1",
-            caps=[TRPAttribute.ISOLATION, TRPAttribute.RECORD_NETWORK],
+            attrs=[TRPAttribute.NET_ISOLATION, TRPAttribute.RECORD_NETWORK],
         )
 
         # Verify performance was added
@@ -341,7 +341,7 @@ def test_add_performance_invalid_arrangement(
                 end_time=end_time,
                 accessed_arrangement="arrangement/99",
                 modified_arrangement="arrangement/0",
-                caps=[],
+                attrs=[],
             )
 
 
@@ -629,7 +629,7 @@ def test_generate_report(self, temp_workspace, tmp_path, gpg_setup, trs_profile)
             comment="Test workflow",
             accessed_arrangement="arrangement/0",
             modified_arrangement="arrangement/1",
-            caps=[TRPAttribute.ISOLATION],
+            attrs=[TRPAttribute.NET_ISOLATION],
         )
 
         # Create a simple template
@@ -756,7 +756,7 @@ def test_complete_data_processing_workflow(
             comment=f"Data filtering with threshold={threshold}",
             accessed_arrangement="arrangement/0",
             modified_arrangement="arrangement/1",
-            caps=[TRPAttribute.ISOLATION, TRPAttribute.RECORD_NETWORK],
+            attrs=[TRPAttribute.NET_ISOLATION, TRPAttribute.RECORD_NETWORK],
         )
 
         # Save the TRO

tro_utils/__init__.py

@@ -1,21 +1,66 @@
 """Top-level package for Transparent Research Object utils."""
 
+from enum import Enum, EnumMeta
+
 __author__ = """Kacper Kowalik"""
 __email__ = "xarthisius.kk@gmail.com"
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 
 
-class TROVCapability:
-    RECORD_NETWORK = "trov:CanRecordInternetAccess"
-    ISOLATION = "trov:CanProvideInternetIsolation"
+class MetaEnum(EnumMeta):
+    @property
+    def values(cls):
+        return [member.value for member in cls]
 
+    def __contains__(cls, item):
+        return item in cls.values
 
-class TRPAttribute:
-    RECORD_NETWORK = "trov:InternetAccessRecording"
-    ISOLATION = "trov:InternetIsolation"
+    def translate(cls, source_member):
+        try:
+            return cls[source_member.name]
+        except (KeyError, AttributeError):
+            raise ValueError(
+                f"No corresponding member in {cls.__name__} for {source_member}"
+            )
+
+
+class TROVTypeEnum(Enum, metaclass=MetaEnum):
+    pass
 
 
-caps_mapping = {
-    TRPAttribute.RECORD_NETWORK: TROVCapability.RECORD_NETWORK,
-    TRPAttribute.ISOLATION: TROVCapability.ISOLATION,
-}
+class TROVCapability(TROVTypeEnum):
+    RECORD_NETWORK = "trov:CanRecordInternetAccess"
+    NET_ISOLATION = "trov:CanProvideInternetIsolation"
+    # NET_ISOLATION = "trov:CanEnforceInternetIsolation"
+    ENV_ISOLATION = "trov:CanIsolateEnvironment"
+    NON_INTERACTIVE = "trov:CanPreventAuthorIntervention"
+    # NON_INTERACTIVE = "trov:CanPreventUserInteractionDuringRun"
+    EXCLUDE_INPUT = "trov:CanExcludeInputs"
+    EXCLUDE_OUTPUT = "trov:CanExcludeOutputs"
+    ALL_DATA_INCLUDED = "trov:CanEnsureInputDataIncludedInTROPackage"
+    REQUIRE_INPUT_DATA = "trov:CanRequireInputDataExistsBeforeRun"
+    REQUIRE_LOCAL_DATA = "trov:CanRequireInputDataLocalBeforeRun"
+    DATA_PERSIST = "trov:CanEnsureInputDataPersistsAfterRun"
+    OUTPUT_INCLUDED = "trov:CanEnsureOutputDataIncludedInTROPackage"
+    CODE_INCLUDED = "trov:CanEnsureCodeIncludedInTROPackage"
+    SOFTWARE_RECORD = "trov:CanRecordSoftwareEnvironment"
+    NET_ACCESS = "trov:CanDetectInternetAccess"
+    MACHINE_ENFORCEMENT = "trov:CanEnforceCapabilitiesTechnically"
+
+
+class TRPAttribute(TROVTypeEnum):
+    RECORD_NETWORK = "trov:InternetAccessRecording"
+    NET_ISOLATION = "trov:InternetIsolation"
+    ENV_ISOLATION = "trov:EnvironmentIsolation"
+    NON_INTERACTIVE = "trov:NonInteractiveExecution"
+    EXCLUDE_INPUT = "trov:InputsExcluded"
+    EXCLUDE_OUTPUT = "trov:OutputsExcluded"
+    ALL_DATA_INCLUDED = "trov:AllInputDataIncludedInTROPackage"
+    REQUIRE_INPUT_DATA = "trov:RequiredInputDataExistsBeforeRun"
+    REQUIRE_LOCAL_DATA = "trov:RequiredInputDataLocalBeforeRun"
+    DATA_PERSIST = "trov:InputDataPersistedAfterRun"
+    OUTPUT_INCLUDED = "trov:OutputDataIncludedInTROPackage"
+    CODE_INCLUDED = "trov:CodeIncludedInTROPackage"
+    SOFTWARE_RECORD = "trov:SoftwareEnvironmentRecorded"
+    NET_ACCESS = "trov:InternetAccessDetection"
+    MACHINE_ENFORCEMENT = "trov:CapabilitiesTechnicallyEnforced"

tro_utils/cli.py

@@ -353,21 +353,21 @@ def generate_report(ctx, template, output):
     help="End time of the performance",
 )
 @click.option(
-    "--caps",
-    "-c",
-    type=click.Choice([TRPAttribute.ISOLATION, TRPAttribute.RECORD_NETWORK]),
+    "--attribute",
+    "-a",
+    type=click.Choice([cap.value for cap in TRPAttribute]),
     required=False,
     multiple=True,
     help="Capabilities of the performance",
 )
 @click.option(
-    "--accessed", "-a", type=click.STRING, required=False, help="Accessed Arrangement"
+    "--accessed", "-A", type=click.STRING, required=False, help="Accessed Arrangement"
 )
 @click.option(
     "--modified", "-M", type=click.STRING, required=False, help="Modified Arrangement"
 )
 @click.pass_context
-def performance_add(ctx, comment, start, end, caps, accessed, modified):
+def performance_add(ctx, comment, start, end, attribute, accessed, modified):
     ctx = ctx.parent.parent
     declaration = ctx.params.get("declaration")
     gpg_fingerprint = ctx.params.get("gpg_fingerprint")
@@ -385,7 +385,7 @@ def performance_add(ctx, comment, start, end, caps, accessed, modified):
         comment=comment,
         accessed_arrangement=accessed,
         modified_arrangement=modified,
-        caps=caps,
+        attrs=attribute,
     )
     tro.save()
 

tro_utils/tro_utils.py

@@ -17,7 +17,7 @@
 import graphviz
 from pyasn1.codec.der import encoder
 
-from . import TRPAttribute, caps_mapping
+from . import TROVCapability, TRPAttribute
 
 GPG_HOME = os.environ.get("GPG_HOME")
 
@@ -249,7 +249,7 @@ def add_arrangement(
         arrangement_id = f"arrangement/{self.get_arrangement_seq()}"
         arrangement = {
             "@id": arrangement_id,
-            "@type": "trov:Artifact Arrangement",
+            "@type": "trov:ArtifactArrangement",
             "rdfs:comment": comment,
             "trov:hasLocus": [],
         }
@@ -447,11 +447,11 @@ def add_performance(
         comment=None,
         accessed_arrangement=None,
         modified_arrangement=None,
-        caps=None,
+        attrs=None,
         extra_attributes=None,
     ):
-        if caps is None:
-            caps = []
+        if attrs is None:
+            attrs = []
         if extra_attributes is None:
             extra_attributes = {}
 
@@ -494,14 +494,20 @@ def add_performance(
         }
 
         i = 0
-        for cap in caps:
-            assert cap in [TRPAttribute.RECORD_NETWORK, TRPAttribute.ISOLATION]
-            assert caps_mapping[cap] in trs_caps
+        for attr in attrs:
+            assert attr.value in TRPAttribute
+            cap = TROVCapability.translate(attr)
+            if cap.value not in trs_caps.keys():
+                raise ValueError(
+                    f"Capability {cap.value} is required for attribute {attr.value} but is not present in TRS capabilities"
+                    "List of TRS capabilities: "
+                    f"{list(trs_caps.keys())}"
+                )
             trp["trov:hasPerformanceAttribute"].append(
                 {
                     "@id": f"{trp['@id']}/attribute/{i}",
-                    "@type": cap,
-                    "trov:warrantedBy": {"@id": trs_caps[caps_mapping[cap]]},
+                    "@type": attr.value,
+                    "trov:warrantedBy": {"@id": trs_caps[cap.value]},
                 }
             )
             i += 1