test: make TestListVideosOffloading responsiveness assertion timing-relative

tests/unit/test_v1_router_extended.py

@@ -2488,30 +2488,52 @@ async def _run():
         )
 
     def test_event_loop_stays_responsive_while_scan_is_in_flight(self):
+        """Ticks must complete *while* the scan is still running.
+
+        Counting ticks alone is not enough: a blocking call with a timeout
+        eventually returns, after which the loop is free and the ticks run
+        anyway. So each tick is timestamped and compared against the moment the
+        scan actually finished. If the scan runs inline it pins the loop, and
+        every tick necessarily lands *after* it -- giving zero qualifying ticks.
+        """
+        import time
+
         release = threading.Event()
-        svc = self._service(on_count=lambda: release.wait(timeout=2.0))
+        finished_at: dict[str, float] = {}
+
+        def _block():
+            release.wait(timeout=5.0)
+            finished_at["scan"] = time.monotonic()
+
+        svc = self._service(on_count=_block)
 
         async def _run():
-            ticks = 0
             task = asyncio.create_task(
                 router_module.list_videos_v1(limit=10, offset=0, data_service=svc)
             )
-            # While the scan is parked in a worker thread the loop must remain
-            # free to schedule unrelated coroutines.
+            tick_times: list[float] = []
             for _ in range(20):
-                if task.done():
-                    break
-                ticks += 1
                 await asyncio.sleep(0.005)
+                tick_times.append(time.monotonic())
+                if len(tick_times) >= 3:
+                    break
             release.set()
-            return ticks, await task
+            return tick_times, await task
 
-        ticks, result = asyncio.run(_run())
+        tick_times, result = asyncio.run(_run())
+
+        # Anti-vacuity: the endpoint still returned its real payload, and the
+        # blocking work really ran to completion.
+        assert result["total"] == 1
+        assert "scan" in finished_at, "count_videos never completed"
 
-        assert result["total"] == 1  # anti-vacuity
-        assert ticks >= 3, (
-            f"event loop only advanced {ticks} time(s) while the scan was "
-            "running; the blocking work is starving the loop"
+        scan_end = finished_at["scan"]
+        concurrent = [t for t in tick_times if t < scan_end]
+        assert len(concurrent) >= 3, (
+            "no loop ticks were observed while the scan was in flight; the "
+            "scan is running inline on the event loop "
+            f"({len(concurrent)} of {len(tick_times)} tick(s) completed "
+            "before the scan finished)"
         )
 
     def test_offset_beyond_total_skips_the_page_read(self):