Merge pull request #495 from golemfactory/az/fix-assertion-reporting

Ensure assertion success/failure is reported in EventMonitor

goth/assertions/monitor.py

@@ -8,7 +8,7 @@
 import importlib
 import logging
 import sys
-from typing import Callable, Generic, List, Optional, Sequence, Union
+from typing import Callable, Generic, List, Optional, Sequence, Set, Union
 
 import colors
 
@@ -76,6 +76,9 @@ class EventMonitor(Generic[E]):
     _logger: Union[logging.Logger, MonitorLoggerAdapter]
     """A logger instance for this monitor."""
 
+    _reported: Set[Assertion[E]]
+    """The set of assertions for which acceptance or failure has been reported."""
+
     _worker_task: Optional[asyncio.Task]
     """A worker task that registers events and checks assertions."""
 
@@ -97,6 +100,7 @@ def __init__(
             self._logger = MonitorLoggerAdapter(
                 self._logger, {MonitorLoggerAdapter.EXTRA_MONITOR_NAME: self.name}
             )
+        self._reported = set()
         self._stop_callback = on_stop
         self._worker_task = None
 
@@ -245,30 +249,39 @@ async def _run_worker(self) -> None:
             await self._check_assertions(events_ended)
 
     async def _check_assertions(self, events_ended: bool) -> None:
+        """Notify assertions that a new event has occurred.
+
+        Should be called exactly once after a new event is added
+        to `self._events` or after the monitor is stopped, with
+        `events_ended` set to `True`.
+        """
 
         event_descr = (
             f"#{len(self._events)} ({self._events[-1]})"
             if not events_ended
             else "EndOfEvents"
         )
 
+        # Notify all active (not done) assertions about the new event.
+        for a in list(self.assertions.keys()):
+            if not a.done:
+                await a.update_events(events_ended=events_ended)
+
+        # Report acceptance/failure for all assertions that completed
+        # since last check.
         for a, level in list(self.assertions.items()):
 
-            if a.done:
+            if not a.done or a in self._reported:
                 continue
 
-            await a.update_events(events_ended=events_ended)
-
-            if a.done:
-                self._logger.debug(
-                    "Assertion '%s' finished after event %s", a.name, event_descr
-                )
-
+            self._reported.add(a)
+            self._logger.debug(
+                "Assertion '%s' finished after event %s", a.name, event_descr
+            )
             if a.accepted:
                 result = a.result()
                 msg = colors.green("Assertion '%s' succeeded; result: %s", style="bold")
                 self._logger.log(level, msg, a.name, result)
-
             elif a.failed:
                 await self._report_failure(a)
 

goth/runner/__init__.py

@@ -18,8 +18,10 @@
     TypeVar,
 )
 
+import colors
 import docker
 
+from goth.assertions.monitor import EventMonitor
 from goth.runner.container.compose import (
     ComposeConfig,
     ComposeNetworkManager,
@@ -123,7 +125,7 @@ def get_probes(
         probes = [p for p in probes if isinstance(p, probe_type)]
         return cast(List[ProbeType], probes)
 
-    def check_assertion_errors(self) -> None:
+    def check_assertion_errors(self, *extra_monitors: EventMonitor) -> None:
         """If any monitor reports an assertion error, raise the first error."""
 
         probe_agents = chain(*(probe.agents for probe in self.probes))
@@ -133,6 +135,7 @@ def check_assertion_errors(self) -> None:
                 (probe.container.logs for probe in self.probes),
                 (agent.log_monitor for agent in probe_agents),
                 [self.proxy.monitor] if self.proxy else [],
+                extra_monitors,
             )
         )
         failed = chain.from_iterable(
@@ -257,7 +260,7 @@ async def __call__(
 
     async def _enter(self) -> None:
         self._exit_stack.enter_context(configure_logging_for_test(self.log_dir))
-        logger.info("Running test: %s", self.test_name)
+        logger.info(colors.yellow("Running test: %s"), self.test_name)
 
         await self._exit_stack.enter_async_context(
             run_compose_network(self._compose_manager, self.log_dir)
@@ -274,7 +277,7 @@ async def _enter(self) -> None:
         await self._start_nodes()
 
     async def _exit(self):
-        logger.info("Test finished: %s", self.test_name)
+        logger.info(colors.yellow("Test finished: %s"), self.test_name)
         await self._exit_stack.aclose()
         payment.clean_up()
 

pyproject.toml

@@ -4,7 +4,7 @@ exclude= '/(\.eggs|\.git|\.hg|\.mypy_cache|\.nox|\.tox|\.venv|venv|\.svn|_build|
 
 [tool.poetry]
 name = "goth"
-version = "0.2.3"
+version = "0.2.4"
 description = "Golem Test Harness - integration testing framework"
 authors = ["Golem Factory <[email protected]>"]
 license = "GPL-3.0"

test/goth/assertions/test_monitor.py

@@ -1,5 +1,4 @@
 """Test the `assertions.monitor`."""
-
 import asyncio
 
 import pytest
@@ -220,3 +219,57 @@ async def long_running_assertion_2(stream: Events):
     monitor.add_assertion(assert_all_positive)
 
     await monitor.stop()
+
+
+@pytest.mark.asyncio
+async def test_assertion_results_reported(caplog):
+    """Test that assertion success and failure are logged.
+
+    This used to be a problem for assertions that do not succeed or fail
+    immediately after consuming an event. For example if an assertion
+    contains `asyncio.wait_for()` then it may raise an exception some time
+    after it consumed any event. After the failure, the monitor will not
+    feed new events to the assertion. But it should report the failure
+    (exactly once).
+    """
+
+    monitor = EventMonitor()
+
+    async def never_accept(events):
+        async for _ in events:
+            pass
+
+    async def await_impossible(events):
+        await asyncio.wait_for(never_accept(events), timeout=0.1)
+
+    async def await_inevitable(events):
+        try:
+            await asyncio.wait_for(never_accept(events), timeout=0.1)
+        except asyncio.TimeoutError:
+            return "I'm fine!"
+
+    monitor.add_assertion(await_impossible)
+    monitor.add_assertion(await_inevitable)
+    monitor.start()
+
+    await monitor.add_event(1)
+    # At this point the assertions are still alive
+    assert not monitor.done
+
+    await asyncio.sleep(0.3)
+    # The assertions should be done now
+    assert monitor.failed
+    assert monitor.satisfied
+
+    # Stopping the monitor should trigger logging assertion success and
+    # failure messages
+    await monitor.stop()
+
+    assert any(
+        record.levelname == "ERROR" and "failed" in record.message
+        for record in caplog.records
+    )
+    assert any(
+        record.levelname == "INFO" and "I'm fine!" in record.message
+        for record in caplog.records
+    )

test/goth/runner/test_assertion_monitors.py

@@ -0,0 +1,53 @@
+"""Unit tests related to the use of assertion monitors in Runner."""
+
+import asyncio
+from pathlib import Path
+import tempfile
+from unittest.mock import Mock
+
+import pytest
+
+from goth.assertions.monitor import EventMonitor
+from goth.runner import Runner, TemporalAssertionError
+
+
+@pytest.mark.asyncio
+async def test_check_assertions(caplog):
+    """Test the `Runner.check_assertion_errors()` method."""
+
+    runner = Runner(
+        base_log_dir=Path(tempfile.mkdtemp()),
+        compose_config=Mock(),
+    )
+
+    async def assertion(events):
+        async for _ in events:
+            break
+        async for _ in events:
+            raise AssertionError("Just failing")
+
+    idle_monitor = EventMonitor()
+    idle_monitor.start()
+    busy_monitor = EventMonitor()
+    busy_monitor.add_assertion(assertion)
+    busy_monitor.start()
+
+    await asyncio.sleep(0.1)
+    runner.check_assertion_errors(idle_monitor, busy_monitor)
+
+    await busy_monitor.add_event(1)
+    await asyncio.sleep(0.1)
+    runner.check_assertion_errors(idle_monitor, busy_monitor)
+
+    await busy_monitor.add_event(2)
+    await asyncio.sleep(0.1)
+    # Assertion failure should be logged at this point
+    assert any(record.levelname == "ERROR" for record in caplog.records)
+    # And `check_assertion_errors()` should raise an exception
+    with pytest.raises(TemporalAssertionError):
+        runner.check_assertion_errors(idle_monitor, busy_monitor)
+
+    await busy_monitor.stop()
+    await idle_monitor.stop()
+    with pytest.raises(TemporalAssertionError):
+        runner.check_assertion_errors(idle_monitor, busy_monitor)