Test hygiene: forget the refused study, quiet _Tee on a closed stream, wait for connector recovery

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/tests/flow/test_connector.py

@@ -1,6 +1,7 @@
 """The connector contract: polling, deduplication, health and discovery."""
 
 import asyncio
+from collections.abc import Callable
 from types import SimpleNamespace
 from typing import Any
 
@@ -48,12 +49,28 @@ def a_sensor(readings: list[Any], **params: Any) -> Sensor:
     return node
 
 
-def run_briefly(node: ConnectorNode, seconds: float = 0.12, app: Any = None) -> None:
-    """Start the poll loop, let it tick a few times, stop it."""
+def run_briefly(
+    node: ConnectorNode,
+    seconds: float = 0.12,
+    app: Any = None,
+    until: Callable[[], bool] | None = None,
+) -> None:
+    """Start the poll loop, let it tick a few times, stop it.
+
+    With ``until``, stop once that holds instead, giving up after 5 s: a fixed
+    pause is too short when the machine is busy. The caller's asserts then say
+    what never happened.
+    """
 
     async def cycle() -> None:
         await node.start(app)
-        await asyncio.sleep(seconds)
+        if until is None:
+            await asyncio.sleep(seconds)
+        else:
+            for _ in range(500):
+                if until():
+                    break
+                await asyncio.sleep(0.01)
         await node.stop()
 
     asyncio.run(cycle())
@@ -97,7 +114,8 @@ def test_a_failing_poll_reports_down_and_keeps_going():
     node._on_health = lambda _node, status, detail: health.append((status, detail))
     Pipeline(nodes=[node])
 
-    run_briefly(node)
+    # The first poll fails, so an "ok" after it is the recovery.
+    run_briefly(node, until=lambda: len(health) > 1 and health[-1][0] == "ok")
 
     assert ("down", "RuntimeError: device unplugged") in health
     # It recovered rather than giving up.