Make epics demo test use log capture

ophyd/v2/tests/test_core.py

@@ -4,7 +4,7 @@
 import time
 import traceback
 from enum import Enum
-from typing import Any, Callable, Sequence, Tuple, Type
+from typing import Any, Callable, Sequence, Tuple, Type, cast
 from unittest.mock import Mock
 
 import bluesky.plan_stubs as bps
@@ -323,6 +323,11 @@ async def test_device_collector_logs_exceptions_for_raised_errors(
     await _assert_failing_device_does_not_connect(DummyDeviceGroupThatErrors)
     assert caplog.records[0].message == "1 Devices raised an error:"
     assert caplog.records[1].message == "  should_fail:"
+    assert_exception_type_and_message(
+        caplog.records[1],
+        OSError,
+        "Connection failed",
+    )
 
 
 async def test_device_collector_logs_exceptions_for_timeouts(
@@ -332,6 +337,11 @@ async def test_device_collector_logs_exceptions_for_timeouts(
     await _assert_failing_device_does_not_connect(DummyDeviceGroupThatTimesOut)
     assert caplog.records[0].message == "1 Devices did not connect:"
     assert caplog.records[1].message == "  should_fail:"
+    assert_exception_type_and_message(
+        caplog.records[1],
+        NotConnected,
+        "child1: source: foo",
+    )
 
 
 async def test_device_collector_logs_exceptions_for_multiple_devices(
@@ -343,8 +353,18 @@ async def test_device_collector_logs_exceptions_for_multiple_devices(
     )
     assert caplog.records[0].message == "1 Devices did not connect:"
     assert caplog.records[1].message == "  should_fail_1:"
+    assert_exception_type_and_message(
+        caplog.records[1],
+        OSError,
+        "Connection failed",
+    )
     assert caplog.records[2].message == "1 Devices raised an error:"
     assert caplog.records[3].message == "  should_fail_2:"
+    assert_exception_type_and_message(
+        caplog.records[3],
+        OSError,
+        "Connection failed",
+    )
 
 
 async def _assert_failing_device_does_not_connect(
@@ -373,6 +393,19 @@ async def _assert_failing_devices_do_not_connect(
     return excepton_info
 
 
+def assert_exception_type_and_message(
+    record: logging.LogRecord,
+    expected_type: Type[Exception],
+    expected_message: str,
+):
+    exception_type, exception, _ = cast(
+        Tuple[Type[Exception], Exception, str],
+        record.exc_info,
+    )
+    assert expected_type is exception_type
+    assert (expected_message,) == exception.args
+
+
 async def normal_coroutine(time: float):
     await asyncio.sleep(time)
 

ophyd/v2/tests/test_epicsdemo.py

@@ -1,6 +1,7 @@
 import asyncio
-from typing import Dict
-from unittest.mock import Mock, call, patch
+import logging
+from typing import Dict, Tuple, Type, cast
+from unittest.mock import Mock, call
 
 import pytest
 from bluesky.protocols import Reading
@@ -130,18 +131,25 @@ async def test_mover_disconncted():
     assert m.name == "mover"
 
 
-async def test_sensor_disconncted():
-    with patch("ophyd.v2.core.logging") as mock_logging:
-        with pytest.raises(NotConnected, match="Not all Devices connected"):
-            async with DeviceCollector(timeout=0.1):
-                s = epicsdemo.Sensor("ca://PRE:", name="sensor")
-        mock_logging.error.assert_called_once_with(
-            """\
-1 Devices did not connect:
-  s: NotConnected
-    value: ca://PRE:Value
-    mode: ca://PRE:Mode"""
-        )
+async def test_sensor_disconncted(caplog: pytest.LogCaptureFixture):
+    caplog.set_level(logging.INFO)
+    with pytest.raises(NotConnected, match="Not all Devices connected"):
+        async with DeviceCollector(timeout=0.1):
+            s = epicsdemo.Sensor("ca://PRE:", name="sensor")
+
+    # Check log messages
+    assert caplog.records[0].message == "1 Devices did not connect:"
+    assert caplog.records[1].message == "  s:"
+
+    # Check logged exception
+    exception_type, exception, _ = cast(
+        Tuple[Type[Exception], Exception, str],
+        caplog.records[1].exc_info,
+    )
+    assert NotConnected is exception_type
+    assert ("value: ca://PRE:Value", "mode: ca://PRE:Mode") == exception.args
+
+    # Ensure correct device
     assert s.name == "sensor"