Fix/issue 770 matrix swallows errors

tests/connectors/test_matrix.py

@@ -82,6 +82,28 @@ def fn(value):
 
         assert sorted(test_vals) == ["a","b","c"]
 
+    def test_error_propagates(self):
+        """
+        Test that an error raised within a matrix run action is
+        correctly raised, rather than being silently swallowed.
+        """
+        def fail():
+            raise ValueError("intentional failure")
+
+        with pytest.raises(ValueError, match="intentional failure"):
+            wrangles.recipe.run(
+                """
+                run:
+                  on_start:
+                    - matrix:
+                        variables:
+                          var: [a,b,c]
+                        run:
+                          - custom.fail: {}
+                """,
+                functions=fail
+            )
+
 
 class TestRead:
     def test_list(self):
@@ -589,3 +611,52 @@ def test_matrix_column_set_non_hashable_numpy_arrays(self):
         # but that's acceptable since numpy array equality is complex)
         actual_ids = [key for key in memory.dataframes.keys() if key.startswith("test_non_hashable_arrays_")]
         assert len(actual_ids) >= 3  # May be 3 or 4 depending on how pandas handles numpy array deduplication
+
+    def test_error_propagates(self):
+        """
+        Test that an error raised within a matrix write action is
+        correctly raised, rather than being silently swallowed.
+        https://github.com/wrangleworks/WranglesPY/issues/770
+        """
+        def fail(df):
+            raise ValueError("intentional failure")
+
+        with pytest.raises(ValueError, match="intentional failure"):
+            wrangles.recipe.run(
+                """
+                write:
+                - matrix:
+                    variables:
+                        key: [a,b,c]
+                    write:
+                        - custom.fail: {}
+                """,
+                dataframe=pd.DataFrame({"col1": ["a","b","c"]}),
+                functions=fail
+            )
+
+    def test_error_propagates_partial_failure(self):
+        """
+        Test that if only one of several matrix write permutations
+        raises an error, the error still propagates. Previously this
+        was silently swallowed and the matrix write appeared to
+        succeed even though one permutation failed.
+        """
+        def fail_for_c(df, key):
+            if key == "c":
+                raise ValueError("intentional failure")
+
+        with pytest.raises(ValueError, match="intentional failure"):
+            wrangles.recipe.run(
+                """
+                write:
+                - matrix:
+                    variables:
+                        key: [a,b,c]
+                    write:
+                        - custom.fail_for_c:
+                            key: ${key}
+                """,
+                dataframe=pd.DataFrame({"col1": ["a","b","c"]}),
+                functions=fail_for_c
+            )

tests/recipes/wrangles/test_extract.py

@@ -4414,7 +4414,7 @@ def test_ai_invalid_model_per_row_error(self):
             })
         )
         assert all(
-            "OpenAI API error" in value and "status=400" in value
+            "OpenAI API error" in value and "status=404" in value
             for value in df['length']
         )
 

wrangles/connectors/matrix.py

@@ -141,16 +141,18 @@ def run(
             strategy=strategy,
             functions=functions
         ):
-            future = executor.submit(
-                _wrangles.recipe.run,
-                recipe={'run': {"on_start": run}},
-                variables=permutation,
-                functions=functions
+            futures.append(
+                executor.submit(
+                    _wrangles.recipe.run,
+                    recipe={'run': {"on_start": run}},
+                    variables=permutation,
+                    functions=functions
+                )
             )
 
-            # Wait for all futures to complete
-            for future in futures:
-                future.result()
+        # Wait for all futures to complete, raising any exceptions
+        for future in futures:
+            future.result()
 
 _schema['run'] = """
 type: object
@@ -323,17 +325,19 @@ def write(
             functions=functions,
             df=df
         ):
-            future = executor.submit(
-                _wrangles.recipe.run,
-                recipe={'write': write},
-                dataframe=df.copy(),
-                variables=permutation,
-                functions=functions
+            futures.append(
+                executor.submit(
+                    _wrangles.recipe.run,
+                    recipe={'write': write},
+                    dataframe=df.copy(),
+                    variables=permutation,
+                    functions=functions
+                )
             )
 
-            # Wait for all futures to complete
-            for future in futures:
-                future.result()
+        # Wait for all futures to complete, raising any exceptions
+        for future in futures:
+            future.result()
 
 _schema['write'] = """
 type: object