[test][loop_executor] Update tests for loop output

beat/backend/python/test/prefix/algorithms/autonomous/invalid_loop_validate_output/1.json

@@ -11,6 +11,11 @@
                     "type": "user/single_integer/1"
                 }
             },
+            "outputs": {
+                "out": {
+                    "type": "user/single_integer/1"
+                }
+            },
             "loop": {
                 "request": {
                     "type": "user/single_integer/1"

beat/backend/python/test/prefix/algorithms/autonomous/invalid_loop_validate_output/1.py

@@ -39,3 +39,7 @@ class Algorithm:
         value = result.value
 
         return value < 6
+
+    def write(self, outputs):
+        outputs["out"].write({"value": self.output})
+        return True

beat/backend/python/test/prefix/algorithms/autonomous/loop/1.json

@@ -10,6 +10,11 @@
                     "type": "user/single_integer/1"
                 }
             },
+            "outputs": {
+                "out": {
+                    "type": "user/single_integer/1"
+                }
+            },
             "loop": {
                 "request": {
                     "type": "user/1d_array_of_integers/1"

beat/backend/python/test/prefix/algorithms/autonomous/loop/1.py

@@ -39,6 +39,7 @@ import numpy as np
 class Algorithm:
     def __init__(self):
         self.threshold = None
+        self.output = None
 
     def setup(self, parameters):
         self.threshold = parameters["threshold"]
@@ -46,4 +47,13 @@ class Algorithm:
 
     def validate(self, result):
         value = result.value[0]
-        return (value < self.threshold, {"value": np.int64(value)})
+        result = value < self.threshold
+        if not result:
+            self.output = value
+
+        return (result, {"value": np.int64(value)})
+
+    def write(self, outputs):
+        outputs["out"].write({"value": np.int32(self.output)})
+        return True
+

beat/backend/python/test/test_loop_executor.py

@@ -80,6 +80,7 @@ CONFIGURATION = {
         "channel": "main",
         "parameters": {"threshold": 1},
         "inputs": {"in": {"path": "INPUT", "channel": "main"}},
+        "outputs": {"out": {"path": "LOOP_OUTPUT", "channel": "main"}},
     },
 }
 
@@ -207,6 +208,8 @@ class TestExecution(unittest.TestCase):
 
         self.assertTrue(executor.process())
 
+        self.loop_executor.close()
+
         cached_file = CachedDataSource()
         self.assertTrue(
             cached_file.setup(
@@ -223,6 +226,22 @@ class TestExecution(unittest.TestCase):
             self.assertEqual(start, i)
             self.assertEqual(end, i)
 
+        cached_file = CachedDataSource()
+        success = cached_file.setup(
+            os.path.join(
+                self.cache_root,
+                CONFIGURATION["loop"]["outputs"]["out"]["path"] + ".data",
+            ),
+            prefix,
+        )
+
+        self.assertTrue(success)
+        for i in range(len(cached_file)):
+            data, start, end = cached_file[i]
+            self.assertEqual(data.value, 1)
+            self.assertEqual(start, i)
+            self.assertEqual(end, i)
+
     def test_autonomous_loop_user(self):
         self.process("autonomous/loop_user/1", "autonomous/loop/1")