Keep runtime test mocks from breaking asynchronous logging

tests/conftest.py

@@ -1,3 +1,4 @@
+import logging
 import os
 from pathlib import Path
 from urllib.parse import urlsplit
@@ -7,6 +8,7 @@
 from alembic.config import Config
 from devfeed_core.config import get_settings
 from devfeed_core.db import get_engine, session_factory
+from devfeed_core.logging import JsonFormatter, QueuedStderrHandler
 from devfeed_core.models import Base, Tag
 from redis import Redis
 from sqlalchemy import text
@@ -15,6 +17,26 @@
 ROOT = Path(__file__).resolve().parents[1]
 
 
+@pytest.fixture
+def queued_runtime_log(capsys):
+    handler = QueuedStderrHandler()
+    handler.setFormatter(JsonFormatter("runtime-fixture"))
+
+    def write():
+        marker = "runtime fixture preserves asynchronous logging"
+        handler.emit(
+            logging.LogRecord("runtime-fixture", logging.INFO, __file__, 0, marker, (), None)
+        )
+        handler.flush()
+        assert marker in capsys.readouterr().err
+        assert handler.worker.is_alive()
+
+    try:
+        yield write
+    finally:
+        handler.close()
+
+
 @pytest.fixture(autouse=True)
 def unit_test_settings(request, monkeypatch):
     # Local credentials and automation settings must never influence tests.

tests/test_scheduler_unit.py

@@ -218,7 +218,9 @@ def lanes(_, __, ___, ____, definitions):
 
 
 @pytest.mark.parametrize("fails", [False, True])
-def test_scheduler_heartbeat_only_advances_after_successful_cycles(monkeypatch, fails):
+def test_scheduler_heartbeat_only_advances_after_successful_cycles(
+    monkeypatch, fails, queued_runtime_log
+):
     stop, health, telemetry = Mock(), Mock(), object()
     stop.is_set.side_effect = [False, True]
     health_context = Mock()
@@ -228,15 +230,19 @@ def test_scheduler_heartbeat_only_advances_after_successful_cycles(monkeypatch,
     monkeypatch.setattr(scheduler, "start_runtime", lambda _: telemetry)
     finish = Mock()
     monkeypatch.setattr(scheduler, "stop_runtime", finish)
-    monkeypatch.setattr(scheduler.threading, "Event", lambda: stop)
+    monkeypatch.setattr(scheduler, "threading", NS(Event=lambda: stop))
     signal_handler = Mock()
     monkeypatch.setattr(scheduler.signal, "signal", signal_handler)
     monkeypatch.setattr(scheduler, "scheduler_health", health_context)
     monkeypatch.setattr(scheduler, "session_factory", Mock())
     monkeypatch.setattr(scheduler, "recommendation_dispatcher", lambda *_: nullcontext())
-    monkeypatch.setattr(
-        scheduler, "tick", Mock(side_effect=RuntimeError("tick")) if fails else Mock()
-    )
+
+    def tick():
+        queued_runtime_log()
+        if fails:
+            raise RuntimeError("tick")
+
+    monkeypatch.setattr(scheduler, "tick", tick)
     scheduler.run()
     assert health.completed.call_count == int(not fails)
     stop.wait.assert_called_once_with(15)

tests/test_search_indexer.py

@@ -42,24 +42,29 @@ def fail(*_):
 
 @pytest.mark.parametrize("once", [False, True])
 @pytest.mark.parametrize("fails", [False, True])
-def test_runtime_stops_telemetry_and_registers_shutdown_signals(monkeypatch, once, fails):
+def test_runtime_stops_telemetry_and_registers_shutdown_signals(
+    monkeypatch, once, fails, queued_runtime_log
+):
     engine, telemetry, stop = Mock(), object(), Mock()
     monkeypatch.setattr(
         runtime, "get_settings", lambda: SimpleNamespace(log_level="INFO", log_format="json")
     )
     monkeypatch.setattr(runtime, "configure_logging", Mock())
-    monkeypatch.setattr(runtime.threading, "Event", lambda: stop)
+    monkeypatch.setattr(runtime, "threading", SimpleNamespace(Event=lambda: stop))
     handlers = Mock()
     monkeypatch.setattr(runtime.signal, "signal", handlers)
     monkeypatch.setattr(runtime, "Typesense", lambda **_: engine)
     start, finish = Mock(return_value=telemetry), Mock()
     monkeypatch.setattr(runtime, "start_runtime", start)
     monkeypatch.setattr(runtime, "stop_runtime", finish)
-    monkeypatch.setattr(
-        runtime,
-        "_consume_index",
-        Mock(side_effect=RuntimeError("sync")) if fails else Mock(return_value=0),
-    )
+
+    def consume(*_):
+        queued_runtime_log()
+        if fails:
+            raise RuntimeError("sync")
+        return 0
+
+    monkeypatch.setattr(runtime, "_consume_index", consume)
     if fails:
         with pytest.raises(RuntimeError):
             runtime.run_indexer(once=once)