Merge https://github.com/nipype/pydra

Author: djarecka

pydra/engine/audit.py

@@ -5,6 +5,7 @@
 import attr
 from ..utils.messenger import send_message, make_message, gen_uuid, now, AuditFlag
 from .helpers import ensure_list, gather_runtime_info, hash_file
+from .specs import attr_fields, File, Directory
 
 
 class Audit:
@@ -173,22 +174,24 @@ def audit_task(self, task):
         import subprocess as sp
 
         label = task.name
-        entity_label = type(label)
 
-        if hasattr(task.inputs, "executable"):
-            command = task.cmdline
-        # assume function task
-        else:
-            command = None
-
-        if hasattr(task.inputs, "in_file"):
-            input_file = task.inputs.in_file
-            file_hash = hash_file(input_file)
-            at_location = os.path.abspath(input_file)
-        else:
-            file_hash = None
-            at_location = None
-            input_file = None
+        command = task.cmdline if hasattr(task.inputs, "executable") else None
+        attr_list = attr_fields(task.inputs)
+        for attrs in attr_list:
+            if attrs.type in [File, Directory]:
+                input_name = attrs.name
+                input_path = os.path.abspath(getattr(task.inputs, input_name))
+                file_hash = hash_file(input_path)
+                entity_id = f"uid:{gen_uuid()}"
+                entity_message = {
+                    "@id": entity_id,
+                    "Label": input_name,
+                    "AtLocation": input_path,
+                    "GeneratedBy": None,
+                    "@type": "input",
+                    "digest": file_hash,
+                }
+                self.audit_message(entity_message, AuditFlag.PROV)
 
         if command is not None:
             cmd_name = command.split()[0]
@@ -217,18 +220,4 @@ def audit_task(self, task):
             "AssociatedWith": version_cmd,
         }
 
-        entity_message = {
-            "@id": self.aid,
-            "Label": print(entity_label),
-            "AtLocation": at_location,
-            "GeneratedBy": "test",  # if not part of workflow, this will be none
-            "@type": "input",
-            "digest": file_hash,  # hash value under helpers.py
-        }
-
-        # new code to be added here for i/o tracking - WIP
-
         self.audit_message(start_message, AuditFlag.PROV)
-        self.audit_message(entity_message, AuditFlag.PROV)
-        # add more fields according to BEP208 doc
-        # with every field, check in tests

pydra/engine/tests/test_task.py

@@ -1007,9 +1007,6 @@ def testfunc(a: int, b: float = 0.1) -> ty.NamedTuple("Output", [("out", float)]
     funky.cache_dir = tmpdir
     funky()
     message_path = tmpdir / funky.checksum / "messages"
-    print(message_path)
-    # go through each jsonld file in message_path and check if the label field exists
-    json_content = []
 
     for file in glob(str(message_path) + "/*.jsonld"):
         with open(file, "r") as f:
@@ -1021,12 +1018,6 @@ def testfunc(a: int, b: float = 0.1) -> ty.NamedTuple("Output", [("out", float)]
             if "@type" in data:
                 if data["@type"] == "input":
                     assert None == data["Label"]
-                    # placeholder for atlocation until
-                    # new test is added
-                    assert None == data["AtLocation"]
-
-                # assert data["Type"] == "input"
-
             if "AssociatedWith" in data:
                 assert None == data["AssociatedWith"]
 
@@ -1072,13 +1063,27 @@ def test_audit_shellcommandtask(tmpdir):
 
 
 def test_audit_shellcommandtask_file(tmpdir):
+    # sourcery skip: use-fstring-for-concatenation
+    import glob
+    import shutil
+
     # create test.txt file with "This is a test" in it in the tmpdir
-    with open(tmpdir / "test.txt", "w") as f:
-        f.write("This is a test.")
+    # create txt file in cwd
+    with open("test.txt", "w") as f:
+        f.write("This is a test")
+
+    with open("test2.txt", "w") as f:
+        f.write("This is a test")
+
+    # copy the test.txt file to the tmpdir
+    shutil.copy("test.txt", tmpdir)
+    shutil.copy("test2.txt", tmpdir)
 
     cmd = "cat"
     file_in = tmpdir / "test.txt"
+    file_in_2 = tmpdir / "test2.txt"
     test_file_hash = hash_file(file_in)
+    test_file_hash_2 = hash_file(file_in_2)
     my_input_spec = SpecInfo(
         name="Input",
         fields=[
@@ -1093,29 +1098,45 @@ def test_audit_shellcommandtask_file(tmpdir):
                         "mandatory": True,
                     },
                 ),
-            )
+            ),
+            (
+                "in_file_2",
+                attr.ib(
+                    type=File,
+                    metadata={
+                        "position": 2,
+                        "argstr": "",
+                        "help_string": "text",
+                        "mandatory": True,
+                    },
+                ),
+            ),
         ],
         bases=(ShellSpec,),
     )
     shelly = ShellCommandTask(
         name="shelly",
         in_file=file_in,
+        in_file_2=file_in_2,
         input_spec=my_input_spec,
         executable=cmd,
         audit_flags=AuditFlag.PROV,
-        messengers=PrintMessenger(),
+        messengers=FileMessenger(),
     )
     shelly.cache_dir = tmpdir
     shelly()
     message_path = tmpdir / shelly.checksum / "messages"
     for file in glob.glob(str(message_path) + "/*.jsonld"):
-        with open(file, "r") as f:
-            data = json.load(f)
-            print(file_in)
-            if "AtLocation" in data:
-                assert data["AtLocation"] == str(file_in)
-            if "digest" in data:
-                assert test_file_hash == data["digest"]
+        with open(file, "r") as x:
+            data = json.load(x)
+            if "@type" in data:
+                if data["@type"] == "input":
+                    if data["Label"] == "in_file":
+                        assert data["AtLocation"] == str(file_in)
+                        assert data["digest"] == test_file_hash
+                    if data["Label"] == "in_file_2":
+                        assert data["AtLocation"] == str(file_in_2)
+                        assert data["digest"] == test_file_hash_2
 
 
 def test_audit_shellcommandtask_version(tmpdir):
@@ -1247,7 +1268,7 @@ def testfunc(a: int, b: float = 0.1) -> ty.NamedTuple("Output", [("out", float)]
     from glob import glob
 
     assert len(glob(str(tmpdir / funky.checksum / "proc*.log"))) == 1
-    assert len(glob(str(message_path / "*.jsonld"))) == 8
+    assert len(glob(str(message_path / "*.jsonld"))) == 7
 
     # commented out to speed up testing
     collect_messages(tmpdir / funky.checksum, message_path, ld_op="compact")