Fix StorageTask listing failing on tagged storage inputs (#1348)

src/python/txtai/workflow/task/storage.py

@@ -54,8 +54,8 @@ class StorageTask(Task):
         outputs = []
         for element in elements:
             if self.matches(element):
-                # Get directory listing and run actions
-                outputs.extend(super().__call__(self.list(element), executor))
+                # Unpack tagged input, list container and re-tag each listing to match input format
+                outputs.extend(self.pack(element, result) for result in super().__call__(self.list(self.upack(element, True)), executor))
             else:
                 outputs.append(element)
 

test/python/testworkflow.py

@@ -524,6 +524,36 @@ class TestWorkflow(unittest.TestCase):
 
         self.assertEqual(len(results), 22)
 
+    def testStorageTaggedWorkflow(self):
+        """
+        Test a storage task with tagged (id, data, tag) inputs (#1328)
+        """
+
+        task = StorageTask()
+
+        # matches() accepts untagged and tagged storage urls
+        self.assertIsNotNone(task.matches("local://" + Utils.PATH))
+        self.assertIsNotNone(task.matches(("id", "local://" + Utils.PATH, "tag")))
+        self.assertIsNone(task.matches("test string"))
+
+        workflow = Workflow([StorageTask()])
+
+        # Tagged inputs list the container and re-tag each file with the input (id, tag)
+        url = "local://" + Utils.PATH
+        results = list(workflow([("id", url, "tag")]))
+        self.assertEqual(len(results), 21)
+        for result in results:
+            self.assertEqual(len(result), 3)
+            self.assertEqual(result[0], "id")
+            # normpath: libcloud joins listing paths with os.path.join, which uses
+            # backslashes on Windows while Utils.PATH uses forward slashes
+            self.assertTrue(os.path.normpath(result[1]).startswith(os.path.normpath(Utils.PATH)))
+            self.assertEqual(result[2], "tag")
+
+        # Non-storage inputs pass through unmodified
+        results = list(workflow([("id2", "test string", "tag2")]))
+        self.assertEqual(results, [("id2", "test string", "tag2")])
+
     def testTemplateInput(self):
         """
         Test template task input