test(gateway): wait for periodic memory ticks

Author: sahilm-ai

tests/gateway/test_memory_monitor.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import logging
-import time
+import threading
 
 import pytest
 
@@ -87,21 +87,33 @@ def test_stop_without_start_is_noop():
     assert mm.is_running() is False
 
 
-def test_periodic_timer_fires(caplog):
-    caplog.set_level(logging.INFO, logger="gateway.memory_monitor")
-    # Short interval so we can observe multiple ticks inside the test budget.
-    mm.start_memory_monitoring(interval_seconds=0.1)
-    time.sleep(0.45)
-    mm.stop_memory_monitoring(timeout=1.0)
+def test_periodic_timer_fires(monkeypatch):
+    periodic_ticks = 0
+    periodic_ticks_lock = threading.Lock()
+    periodic_tick_event = threading.Event()
+    original_log_memory_usage = mm.log_memory_usage
+
+    def _instrumented_log_memory_usage(prefix: str = "") -> None:
+        nonlocal periodic_ticks
+        original_log_memory_usage(prefix=prefix)
+        if prefix:
+            return
+        with periodic_ticks_lock:
+            periodic_ticks += 1
+            if periodic_ticks >= 2:
+                periodic_tick_event.set()
+
+    monkeypatch.setattr(mm, "log_memory_usage", _instrumented_log_memory_usage)
+    started = mm.start_memory_monitoring(interval_seconds=0.1)
+    assert started is True
+
+    try:
+        # Behavioral contract: while running, monitor emits repeated periodic ticks.
+        fired = periodic_tick_event.wait(timeout=1.0)
+    finally:
+        mm.stop_memory_monitoring(timeout=1.0)
 
-    periodic = [
-        r for r in caplog.records
-        if r.getMessage().startswith("[MEMORY] rss=") or r.getMessage().startswith("[MEMORY] rss=unavailable")
-    ]
-    # baseline + at least 2 periodic + shutdown — but shutdown has the
-    # "shutdown " prefix so it won't match the strict "[MEMORY] rss=" start.
-    # We expect >= 3 bare "[MEMORY] rss=..." lines.
-    assert len(periodic) >= 3, [r.getMessage() for r in caplog.records]
+    assert fired, f"expected at least two periodic ticks, saw {periodic_ticks}"
 
 
 def test_thread_is_daemon():