test: add helper for parameterized outcome tests

Add an abstract class, `test.until.outcome.OutcomeChecker`, and
implementations that make it easier to check for specific outcomes in
parameterized tests.

Also changed existing tests that use similar patterns to use
`OutcomeChecker` instead.

Author: WhiteGobo

test/test_graph/test_graph.py

@@ -1,13 +1,12 @@
 # -*- coding: utf-8 -*-
 import logging
 import os
-from contextlib import ExitStack
 from pathlib import Path
 from test.data import TEST_DATA_DIR, bob, cheese, hates, likes, michel, pizza, tarek
 from test.utils import GraphHelper, get_unique_plugin_names
-from test.utils.exceptions import ExceptionChecker
 from test.utils.httpfileserver import HTTPFileServer, ProtoFileResource
-from typing import Callable, Optional, Set, Tuple, Union
+from test.utils.outcome import ExceptionChecker, OutcomeChecker, OutcomePrimitive
+from typing import Callable, Optional, Set, Tuple
 from urllib.error import HTTPError, URLError
 
 import pytest
@@ -373,7 +372,7 @@ def test_guess_format_for_parse_http(
     http_file_server: HTTPFileServer,
     file: Path,
     content_type: Optional[str],
-    expected_result: Union[int, ExceptionChecker],
+    expected_result: OutcomePrimitive[int],
 ) -> None:
     graph = make_graph()
     headers: Tuple[Tuple[str, str], ...] = tuple()
@@ -384,21 +383,11 @@ def test_guess_format_for_parse_http(
         ProtoFileResource(headers, file),
         suffix=f"/{file.name}",
     )
-    catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-
+    checker = OutcomeChecker.from_primitive(expected_result)
     assert 0 == len(graph)
-    with ExitStack() as exit_stack:
-        if isinstance(expected_result, ExceptionChecker):
-            catcher = exit_stack.enter_context(pytest.raises(expected_result.type))
+    with checker.context():
         graph.parse(location=file_info.request_url)
-
-    if catcher is not None:
-        # assert catcher.value is not None
-        assert isinstance(expected_result, ExceptionChecker)
-        logging.debug("graph = %s", list(graph.triples((None, None, None))))
-    else:
-        assert isinstance(expected_result, int)
-        assert expected_result == len(graph)
+        checker.check(len(graph))
 
 
 def test_parse_file_uri(make_graph: GraphFactory):

test/test_literal/test_literal.py

@@ -9,11 +9,11 @@
 
 import datetime
 import logging
-from contextlib import ExitStack
 from decimal import Decimal
 from test.utils import affix_tuples
 from test.utils.literal import LiteralChecker
-from typing import Any, Callable, Generator, Iterable, Optional, Type, Union
+from test.utils.outcome import OutcomeChecker, OutcomePrimitive, OutcomePrimitives
+from typing import Any, Callable, Generator, Optional, Type, Union
 
 import isodate
 import pytest
@@ -614,16 +614,10 @@ def test_literal_addsub(
     a: Literal,
     b: Literal,
     op: str,
-    expected_result: Union[Literal, Type[Exception], Exception],
+    expected_result: OutcomePrimitive[Literal],
 ) -> None:
-    catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-    expected_exception: Optional[Exception] = None
-    with ExitStack() as xstack:
-        if isinstance(expected_result, type) and issubclass(expected_result, Exception):
-            catcher = xstack.enter_context(pytest.raises(expected_result))
-        elif isinstance(expected_result, Exception):
-            expected_exception = expected_result
-            catcher = xstack.enter_context(pytest.raises(type(expected_exception)))
+    checker = OutcomeChecker[Literal].from_primitive(expected_result)
+    with checker.context():
         if op == "aplusb":
             result = a + b
 
@@ -636,14 +630,7 @@ def test_literal_addsub(
         else:
             raise ValueError(f"invalid operation {op}")
         logging.debug("result = %r", result)
-    if catcher is not None or expected_exception is not None:
-        assert catcher is not None
-        assert catcher.value is not None
-        if expected_exception is not None:
-            assert catcher.match(expected_exception.args[0])
-    else:
-        assert isinstance(expected_result, Literal)
-        assert expected_result == result
+        checker.check(result)
 
 
 @pytest.mark.parametrize(
@@ -930,7 +917,7 @@ def unlexify(s: str) -> str:
 
 
 @pytest.mark.parametrize(
-    ["literal_maker", "checks"],
+    ["literal_maker", "outcome"],
     [
         (
             lambda: Literal("foo"),
@@ -969,32 +956,9 @@ def unlexify(s: str) -> str:
 )
 def test_literal_construction(
     literal_maker: Callable[[], Literal],
-    checks: Union[
-        Iterable[Union[LiteralChecker, Literal]],
-        LiteralChecker,
-        Literal,
-        Type[Exception],
-    ],
+    outcome: OutcomePrimitives[Literal],
 ) -> None:
-    check_error: Optional[Type[Exception]] = None
-    if isinstance(checks, type) and issubclass(checks, Exception):
-        check_error = checks
-        checks = []
-    elif not isinstance(checks, Iterable):
-        checks = [checks]
-
-    catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-    with ExitStack() as xstack:
-        if check_error is not None:
-            catcher = xstack.enter_context(pytest.raises(check_error))
-        literal = literal_maker()
-
-    if check_error is not None:
-        assert catcher is not None
-        assert catcher.value is not None
-
-    for check in checks:
-        if isinstance(check, LiteralChecker):
-            check.check(literal)
-        else:
-            check = literal
+    checker = OutcomeChecker[Literal].from_primitives(outcome)
+    with checker.context():
+        actual_outcome = literal_maker()
+        checker.check(actual_outcome)

test/test_misc/test_input_source.py

@@ -7,18 +7,16 @@
 import re
 from contextlib import ExitStack, contextmanager
 from dataclasses import dataclass
-
-# from itertools import product
 from pathlib import Path
 from test.utils import GraphHelper
-from test.utils.exceptions import ExceptionChecker
 from test.utils.httpfileserver import (
     HTTPFileInfo,
     HTTPFileServer,
     LocationType,
     ProtoFileResource,
     ProtoRedirectResource,
 )
+from test.utils.outcome import ExceptionChecker
 from typing import (  # Callable,
     IO,
     BinaryIO,
@@ -648,9 +646,7 @@ def test_create_input_source(
     input_source: Optional[InputSource] = None
     with ExitStack() as xstack:
         if isinstance(test_params.expected_result, ExceptionChecker):
-            catcher = xstack.enter_context(
-                pytest.raises(test_params.expected_result.type)
-            )
+            catcher = xstack.enter_context(test_params.expected_result.context())
 
         input_source = xstack.enter_context(
             call_create_input_source(
@@ -670,8 +666,3 @@ def test_create_input_source(
             )
 
     logging.debug("input_source = %s, catcher = %s", input_source, catcher)
-
-    if isinstance(test_params.expected_result, ExceptionChecker):
-        assert catcher is not None
-        assert input_source is None
-        test_params.expected_result.check(catcher.value)

test/test_misc/test_networking_redirect.py

@@ -1,7 +1,7 @@
 from contextlib import ExitStack
 from copy import deepcopy
-from test.utils.exceptions import ExceptionChecker
 from test.utils.http import headers_as_message as headers_as_message
+from test.utils.outcome import ExceptionChecker
 from typing import Any, Dict, Iterable, Optional, Type, TypeVar, Union
 from urllib.error import HTTPError
 from urllib.request import HTTPRedirectHandler, Request
@@ -197,14 +197,13 @@ def test_make_redirect_request(
     result: Optional[Request] = None
     with ExitStack() as stack:
         if isinstance(expected_result, ExceptionChecker):
-            catcher = stack.enter_context(pytest.raises(expected_result.type))
+            catcher = stack.enter_context(expected_result.context())
         elif expected_result is RaisesIdentity:
             catcher = stack.enter_context(pytest.raises(HTTPError))
         result = _make_redirect_request(http_request, http_error)
 
     if isinstance(expected_result, ExceptionChecker):
         assert catcher is not None
-        expected_result.check(catcher.value)
     elif isinstance(expected_result, type):
         assert catcher is not None
         assert http_error is catcher.value

test/test_namespace/test_namespace.py

@@ -1,5 +1,5 @@
-from contextlib import ExitStack
-from typing import Any, Optional, Type, Union
+from test.utils.outcome import OutcomeChecker, OutcomePrimitive
+from typing import Any, Optional
 from warnings import warn
 
 import pytest
@@ -306,22 +306,15 @@ def test_expand_curie_exception_messages(self) -> None:
         ],
     )
     def test_expand_curie(
-        self, curie: Any, expected_result: Union[Type[Exception], URIRef, None]
+        self, curie: Any, expected_result: OutcomePrimitive[URIRef]
     ) -> None:
         g = Graph(bind_namespaces="none")
         nsm = g.namespace_manager
         nsm.bind("ex", "urn:example:")
+
+        checker = OutcomeChecker.from_primitive(expected_result)
+
         result: Optional[URIRef] = None
-        catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-        with ExitStack() as xstack:
-            if isinstance(expected_result, type) and issubclass(
-                expected_result, Exception
-            ):
-                catcher = xstack.enter_context(pytest.raises(expected_result))
+        with checker.context():
             result = g.namespace_manager.expand_curie(curie)
-
-        if catcher is not None:
-            assert result is None
-            assert catcher.value is not None
-        else:
-            assert expected_result == result
+            checker.check(result)

test/test_namespace/test_namespacemanager.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
 import logging
-import re
 import sys
 from contextlib import ExitStack
 from pathlib import Path
-from test.utils.exceptions import ExceptionChecker
+from test.utils.outcome import ExceptionChecker, OutcomeChecker, OutcomePrimitive
 from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Set, Tuple, Type, Union
 
 import pytest
@@ -374,7 +373,7 @@ def test_compute_qname(
     manager_prefixes: Optional[Mapping[str, Namespace]],
     graph_prefixes: Optional[Mapping[str, Namespace]],
     store_prefixes: Optional[Mapping[str, Namespace]],
-    expected_result: Union[Tuple[str, URIRef, str], Type[Exception], Exception],
+    expected_result: OutcomePrimitive[Tuple[str, URIRef, str]],
 ) -> None:
     """
     :param uri: argument to compute_qname()
@@ -403,25 +402,13 @@ def test_compute_qname(
             nm.bind(prefix, ns)
 
     def check() -> None:
-        catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-        with ExitStack() as xstack:
-            if isinstance(expected_result, type) and issubclass(
-                expected_result, Exception
-            ):
-                catcher = xstack.enter_context(pytest.raises(expected_result))
-            if isinstance(expected_result, Exception):
-                catcher = xstack.enter_context(pytest.raises(type(expected_result)))
+        checker = OutcomeChecker[Tuple[str, URIRef, str]].from_primitive(
+            expected_result
+        )
+        with checker.context():
             actual_result = nm.compute_qname(uri, generate)
             logging.debug("actual_result = %s", actual_result)
-        if catcher is not None:
-            assert catcher is not None
-            assert catcher.value is not None
-            if isinstance(expected_result, Exception):
-                assert re.match(expected_result.args[0], f"{catcher.value}")
-        else:
-            assert isinstance(expected_result, tuple)
-            assert isinstance(actual_result, tuple)
-            assert actual_result == expected_result
+            checker.check(actual_result)
 
     check()
     # Run a second time to check caching
@@ -452,7 +439,7 @@ def test_compute_qname_strict(
     generate: bool,
     bind_namespaces: _NamespaceSetString,
     additional_prefixes: Optional[Mapping[str, Namespace]],
-    expected_result: Union[Tuple[str, URIRef, str], Type[Exception], Exception],
+    expected_result: OutcomePrimitive[Tuple[str, str, str]],
 ) -> None:
     graph = Graph(bind_namespaces=bind_namespaces)
     nm = graph.namespace_manager
@@ -462,25 +449,11 @@ def test_compute_qname_strict(
             nm.bind(prefix, ns)
 
     def check() -> None:
-        catcher: Optional[pytest.ExceptionInfo[Exception]] = None
-        with ExitStack() as xstack:
-            if isinstance(expected_result, type) and issubclass(
-                expected_result, Exception
-            ):
-                catcher = xstack.enter_context(pytest.raises(expected_result))
-            if isinstance(expected_result, Exception):
-                catcher = xstack.enter_context(pytest.raises(type(expected_result)))
+        checker = OutcomeChecker[Tuple[str, str, str]].from_primitive(expected_result)
+        with checker.context():
             actual_result = nm.compute_qname_strict(uri, generate)
             logging.debug("actual_result = %s", actual_result)
-        if catcher is not None:
-            assert catcher is not None
-            assert catcher.value is not None
-            if isinstance(expected_result, Exception):
-                assert re.match(expected_result.args[0], f"{catcher.value}")
-        else:
-            assert isinstance(expected_result, tuple)
-            assert isinstance(actual_result, tuple)
-            assert actual_result == expected_result
+            checker.check(actual_result)
 
     check()
     # Run a second time to check caching
@@ -538,16 +511,15 @@ def test_nsm_function() -> NamespaceManager:
 def test_expand_curie(
     test_nsm_session: NamespaceManager,
     curie: str,
-    expected_result: Union[ExceptionChecker, str],
+    expected_result: OutcomePrimitive[str],
 ) -> None:
     nsm = test_nsm_session
-    with ExitStack() as xstack:
-        if isinstance(expected_result, ExceptionChecker):
-            xstack.enter_context(expected_result)
-        result = nsm.expand_curie(curie)
-
-    if not isinstance(expected_result, ExceptionChecker):
-        assert URIRef(expected_result) == result
+    if isinstance(expected_result, str):
+        expected_result = URIRef(expected_result)
+    checker = OutcomeChecker[str].from_primitive(expected_result)
+    with checker.context():
+        actual_result = nsm.expand_curie(curie)
+        checker.check(actual_result)
 
 
 @pytest.mark.parametrize(
@@ -578,7 +550,7 @@ def test_generate_curie(
     test_nsm_function: NamespaceManager,
     uri: str,
     generate: Optional[bool],
-    expected_result: Union[ExceptionChecker, str],
+    expected_result: OutcomePrimitive[str],
 ) -> None:
     """
     .. note::
@@ -587,13 +559,10 @@ def test_generate_curie(
         effects and will modify the namespace manager.
     """
     nsm = test_nsm_function
-    with ExitStack() as xstack:
-        if isinstance(expected_result, ExceptionChecker):
-            xstack.enter_context(expected_result)
+    checker = OutcomeChecker[str].from_primitive(expected_result)
+    with checker.context():
         if generate is None:
-            result = nsm.curie(uri)
+            actual_result = nsm.curie(uri)
         else:
-            result = nsm.curie(uri, generate=generate)
-
-    if not isinstance(expected_result, ExceptionChecker):
-        assert expected_result == result
+            actual_result = nsm.curie(uri, generate=generate)
+        checker.check(actual_result)