feat: add utilities for handling multiple assertions

adbc_drivers_validation/utils.py

@@ -211,24 +211,54 @@ def arrow_type_name(
     return str(arrow_type)
 
 
+@contextlib.contextmanager
+def multiassert() -> typing.Generator[
+    typing.Callable[[], contextlib.AbstractContextManager[None]], None, None
+]:
+    failures = []
+
+    @contextlib.contextmanager
+    def _multiassert_inner() -> typing.Generator[None, None, None]:
+        nonlocal failures
+        try:
+            yield
+        except ExceptionGroup as e:
+            failures.extend(e.exceptions)
+        except Exception as e:
+            failures.append(e)
+
+    try:
+        yield _multiassert_inner
+    except Exception as e:
+        failures.append(e)
+    finally:
+        if len(failures) == 1:
+            raise failures[0]
+        elif len(failures) > 1:
+            raise ExceptionGroup("multiple failures", failures)
+
+
 def assert_field_type_name(
     driver: "DriverQuirks",
     context: typing.Literal["query", "execute_schema", "get_table_schema"],
     query: "Query",
     schema: pyarrow.Schema,
 ) -> None:
     field_name = (f"{driver.field_metadata_prefix}:type").encode("utf-8")
-    if driver.features.metadata_type_name:
-        for i, field in enumerate(schema):
-            assert field.metadata is not None
-            assert field_name in field.metadata
-            assert field.metadata[field_name].decode(
-                "utf-8"
-            ) == query.metadata().tags.metadata_type_name(context, i)
-    else:
-        for field in schema:
-            if field.metadata is not None:
-                assert field_name not in field.metadata
+    with multiassert() as ma:
+        if driver.features.metadata_type_name:
+            for i, field in enumerate(schema):
+                with ma(), scoped_trace(f"field {i} ({field.name})"):
+                    assert field.metadata is not None
+                    assert field_name in field.metadata
+                    assert field.metadata[field_name].decode(
+                        "utf-8"
+                    ) == query.metadata().tags.metadata_type_name(context, i)
+        else:
+            for i, field in enumerate(schema):
+                with ma(), scoped_trace(f"field {i} ({field.name})"):
+                    if field.metadata is not None:
+                        assert field_name not in field.metadata
 
 
 def generate_tests_by_marks(

tests/test_utils.py

@@ -129,3 +129,85 @@ def test_sort_by(take_table: pyarrow.Table) -> None:
 
     table = utils.sort_by(take_table, [("idx", "descending")])
     assert not table.equals(take_table)
+
+
+def test_multiassert() -> None:
+    with pytest.raises(ExceptionGroup, match="multiple failures") as excinfo:
+        with utils.multiassert() as ma:
+            with ma():
+                assert 1 == 1
+                assert 1 == 2
+                assert 1 == 3
+
+            with ma():
+                assert 2 == 3
+
+            with ma():
+                assert 1 == 1
+
+    assert len(excinfo.value.exceptions) == 2
+    assert "1 == 2" in repr(excinfo.value.exceptions[0])
+    assert "2 == 3" in repr(excinfo.value.exceptions[1])
+
+
+def test_multiassert_single() -> None:
+    with pytest.raises(AssertionError, match="1 == 2"):
+        with utils.multiassert() as ma:
+            with ma():
+                assert 1 == 1
+                assert 1 == 2
+
+    with pytest.raises(AssertionError, match="1 == 2"):
+        with utils.multiassert() as ma:
+            with ma():
+                assert 1 == 1
+            assert 1 == 2
+
+    with pytest.raises(ValueError, match="foobar"):
+        with utils.multiassert() as ma:
+            with ma():
+                raise ValueError("foobar")
+
+    with pytest.raises(ValueError, match="foobar"):
+        with utils.multiassert() as ma:
+            with ma():
+                pass
+            raise ValueError("foobar")
+
+
+def test_multiassert_outside() -> None:
+    with pytest.raises(ExceptionGroup, match="multiple failures") as excinfo:
+        with utils.multiassert() as ma:
+            with ma():
+                assert 1 == 2
+            assert 1 == 3
+
+    assert len(excinfo.value.exceptions) == 2
+    assert "1 == 2" in repr(excinfo.value.exceptions[0])
+    assert "1 == 3" in repr(excinfo.value.exceptions[1])
+
+
+def test_multiassert_nested() -> None:
+    with pytest.raises(ExceptionGroup, match="multiple failures") as excinfo:
+        with utils.multiassert() as ma:
+            with ma():
+                with utils.multiassert() as ma2:
+                    with ma2():
+                        assert 1 == 1
+
+                    with ma2():
+                        assert 1 == 2
+
+                    with ma2():
+                        assert 1 == 3
+
+            with ma():
+                assert 2 == 3
+
+            with ma():
+                assert 1 == 1
+
+    assert len(excinfo.value.exceptions) == 3
+    assert "1 == 2" in repr(excinfo.value.exceptions[0])
+    assert "1 == 3" in repr(excinfo.value.exceptions[1])
+    assert "2 == 3" in repr(excinfo.value.exceptions[2])