better server error capture in test

Author: rmorshea

idom/testing.py

@@ -1,3 +1,5 @@
+import re
+import logging
 from urllib.parse import urlunparse, urlencode
 from typing import (
     Callable,
@@ -8,9 +10,12 @@
     Dict,
     Generic,
     TypeVar,
+    List,
+    Union,
 )
-from weakref import finalize
+from types import TracebackType
 
+from loguru import logger
 from selenium.webdriver.remote.webdriver import WebDriver
 from selenium.webdriver import Chrome
 
@@ -46,13 +51,19 @@ def create_simple_selenium_web_driver(
     return driver
 
 
+_Self = TypeVar("_Self", bound="ServerMountPoint[Any, Any]")
 _Mount = TypeVar("_Mount")
 _Server = TypeVar("_Server", bound=AnyRenderServer)
 
 
 class ServerMountPoint(Generic[_Mount, _Server]):
+    """A context manager for imperatively mounting views to a render server when testing"""
 
-    __slots__ = "server", "host", "port", "mount", "__weakref__"
+    mount: _Mount
+    server: _Server
+
+    _log_handler: "_LogRecordCaptor"
+    _loguru_handler_id: int
 
     def __init__(
         self,
@@ -64,20 +75,60 @@ def __init__(
         mount_and_server_constructor: "Callable[..., Tuple[_Mount, _Server]]" = hotswap_server,  # type: ignore
         app: Optional[Any] = None,
         **other_options: Any,
-    ):
+    ) -> None:
         self.host = host
         self.port = port or find_available_port(host)
-        self.mount, self.server = mount_and_server_constructor(
-            server_type,
-            self.host,
-            self.port,
-            server_config,
-            run_kwargs,
-            app,
-            **other_options,
+        self._mount_and_server_constructor: "Callable[[], Tuple[_Mount, _Server]]" = (
+            lambda: mount_and_server_constructor(
+                server_type,
+                self.host,
+                self.port,
+                server_config,
+                run_kwargs,
+                app,
+                **other_options,
+            )
         )
-        # stop server once mount is done being used
-        finalize(self, self.server.stop)
+
+    @property
+    def log_records(self) -> List[logging.LogRecord]:
+        """A list of captured log records"""
+        return self._log_handler.records
+
+    def assert_logged_exception(
+        self,
+        error_type: Type[Exception],
+        error_pattern: str,
+        clear_after: bool = True,
+    ) -> None:
+        """Assert that a given error type and message were logged"""
+        try:
+            re_pattern = re.compile(error_pattern)
+            for record in self.log_records:
+                if record.exc_info is not None:
+                    error = record.exc_info[1]
+                    if isinstance(error, error_type) and re_pattern.search(str(error)):
+                        break
+            else:  # pragma: no cover
+                assert False, f"did not raise {error_type} matching {error_pattern!r}"
+        finally:
+            if clear_after:
+                self.log_records.clear()
+
+    def raise_first_logged_exception(
+        self,
+        exclude_exc_types: Union[Type[Exception], Tuple[Type[Exception], ...]] = (),
+    ) -> None:
+        """Raise the first logged exception (if any)
+
+        Args:
+            exclude_exc_types: Any exception types to ignore
+        """
+        for record in self._log_handler.records:
+            if record.exc_info is not None:
+                error = record.exc_info[1]
+                if error is not None and not isinstance(error, exclude_exc_types):
+                    raise error
 
     def url(self, path: str = "", query: Optional[Any] = None) -> str:
         return urlunparse(
@@ -90,3 +141,34 @@ def url(self, path: str = "", query: Optional[Any] = None) -> str:
                 "",
             ]
         )
+
+    def __enter__(self: _Self) -> _Self:
+        self._log_handler = _LogRecordCaptor()
+        logging.getLogger().addHandler(self._log_handler)
+        self._loguru_handler_id = logger.add(self._log_handler, format="{message}")
+        self.mount, self.server = self._mount_and_server_constructor()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        self.server.stop()
+
+        logging.getLogger().removeHandler(self._log_handler)
+        logger.remove(self._loguru_handler_id)
+
+        self.raise_first_logged_exception()
+
+        return None
+
+
+class _LogRecordCaptor(logging.NullHandler):
+    def __init__(self) -> None:
+        self.records: List[logging.LogRecord] = []
+        super().__init__()
+
+    def handle(self, record: logging.LogRecord) -> None:
+        self.records.append(record)

noxfile.py

@@ -9,7 +9,7 @@
 
 
 @nox.session
-def test_suite(session: Session) -> None:
+def test_python(session: Session) -> None:
     session.env.update(os.environ)
     session.install("-r", "requirements/test-env.txt")
     session.install(".[all]")
@@ -22,7 +22,8 @@ def test_suite(session: Session) -> None:
 @nox.session
 def check_types(session: Session) -> None:
     session.install("-r", "requirements/check-types.txt")
-    session.install(".[all]")
+    session.install("-r", "requirements/pkg-deps.txt")
+    session.install("-r", "requirements/pkg-extras.txt")
     session.run("mypy", "--strict", "idom")
 
 

tests/conftest.py

@@ -67,7 +67,8 @@ def server_mount_point():
 
     The ``mount`` and ``server`` fixtures use this.
     """
-    return ServerMountPoint(server_config={"cors": True})
+    with ServerMountPoint(server_config={"cors": True}) as mount_point:
+        yield mount_point
 
 
 @pytest.fixture(scope="module")

tests/test_server/test_common/test_per_client_state.py

@@ -20,7 +20,8 @@
     ids=lambda cls: f"{cls.__module__}.{cls.__name__}",
 )
 def server_mount_point(request):
-    return ServerMountPoint(request.param)
+    with ServerMountPoint(request.param) as mount_point:
+        yield mount_point
 
 
 def test_display_simple_hello_world(driver, display):

tests/test_server/test_common/test_shared_state_client.py

@@ -18,7 +18,8 @@
     ids=lambda cls: f"{cls.__module__}.{cls.__name__}",
 )
 def server_mount_point(request):
-    return ServerMountPoint(request.param, sync_views=True)
+    with ServerMountPoint(request.param, sync_views=True) as mount_point:
+        yield mount_point
 
 
 def test_shared_client_state(create_driver, server_mount_point):
@@ -71,17 +72,13 @@ def Counter(count):
 
 
 def test_shared_client_state_server_does_not_support_per_client_parameters(
-    driver_get, caplog
+    driver_get,
+    server_mount_point,
 ):
     driver_get({"per_client_param": 1})
 
-    for record in caplog.records:
-        if record.exc_info and isinstance(record.exc_info[1], ValueError):
-            assert "does not support per-client view parameters" in str(
-                record.exc_info[1]
-            )
-            break
-    else:
-        assert False, "did not log error"
-
-    caplog.clear()
+    server_mount_point.assert_logged_exception(
+        ValueError,
+        "does not support per-client view parameters",
+        clear_after=True,
+    )

tests/test_server/test_prefab.py

@@ -10,10 +10,11 @@
 
 @pytest.fixture
 def server_mount_point():
-    return ServerMountPoint(
+    with ServerMountPoint(
         find_builtin_server_type("PerClientStateServer"),
         mount_and_server_constructor=multiview_server,
-    )
+    ) as mount_point:
+        yield mount_point
 
 
 def test_multiview_server(driver_get, driver, server_mount_point):
@@ -35,3 +36,5 @@ def test_multiview_server(driver_get, driver, server_mount_point):
 
     assert no_such_element(driver, "id", "e1")
     assert no_such_element(driver, "id", "e2")
+
+    server_mount_point.log_records.clear()