Speed up behavior-focused test fixtures

Author: kstonekuan

tests/test_catalog_curation.py

@@ -172,11 +172,17 @@ def recorded_data_root(tmp_path_factory: pytest.TempPathFactory) -> Path:
     episode_specs = {
         # ~12.5% blackout: passes the gate below.
         "fold_napkin": SyntheticEpisodeSpec(
-            duration_s=4.0, task="fold_napkin", black_segment=(1.0, 1.5)
+            duration_s=1.0,
+            cameras=("wrist_cam",),
+            task="fold_napkin",
+            black_segment=(0.2, 0.35),
         ),
-        # 75% blackout: quarantined by the gate below.
+        # ~80% blackout: quarantined by the gate below.
         "pour_water": SyntheticEpisodeSpec(
-            duration_s=4.0, task="pour_water", black_segment=(0.5, 3.5)
+            duration_s=1.0,
+            cameras=("wrist_cam",),
+            task="pour_water",
+            black_segment=(0.1, 0.9),
         ),
     }
 

tests/test_checks.py

@@ -131,11 +131,31 @@ def test_action_rate_matches_the_synthesized_rate(jittery_episode: hflow.Episode
 
 
 def test_content_digest_identifies_duplicate_content(tmp_path: Path) -> None:
-    spec = SyntheticEpisodeSpec(duration_s=2.0)
+    # Digest behavior is independent of camera encoding. A tiny state stream
+    # keeps this contract focused on message content instead of fixture cost.
+    spec = SyntheticEpisodeSpec(
+        duration_s=0.2,
+        cameras=(),
+        joint_hz=10.0,
+        joint_count=1,
+        black_segment=None,
+        joint_jump_at_s=None,
+        timestamp_offset_segment=None,
+    )
     first = synthesize_episode(tmp_path / "a.mcap", spec)
     duplicate = synthesize_episode(tmp_path / "b.mcap", spec)
     different = synthesize_episode(
-        tmp_path / "c.mcap", SyntheticEpisodeSpec(duration_s=2.0, joint_jump_at_s=1.0)
+        tmp_path / "c.mcap",
+        SyntheticEpisodeSpec(
+            duration_s=0.2,
+            cameras=(),
+            joint_hz=10.0,
+            joint_count=1,
+            black_segment=None,
+            joint_jump_at_s=None,
+            timestamp_offset_segment=None,
+            seed=1,
+        ),
     )
     with (
         hflow.Episode(first) as ep_a,

tests/test_end_to_end.py

@@ -8,7 +8,7 @@
 import pytest
 
 import hflow
-from hflow.testing import synthesize_episode
+from hflow.testing import SyntheticEpisodeSpec, synthesize_episode
 
 
 def check_joint_smoothness(joints: np.ndarray, rate_hz: float) -> dict[str, float]:
@@ -21,7 +21,23 @@ def check_joint_smoothness(joints: np.ndarray, rate_hz: float) -> dict[str, floa
 
 @pytest.fixture(scope="module")
 def source_episode(tmp_path_factory: pytest.TempPathFactory) -> Path:
-    return synthesize_episode(tmp_path_factory.mktemp("e2e-source") / "episode_0001.mcap")
+    return synthesize_episode(
+        tmp_path_factory.mktemp("e2e-source") / "episode_0001.mcap",
+        SyntheticEpisodeSpec(
+            duration_s=2.0,
+            black_segment=(0.5, 0.75),
+            joint_jump_at_s=1.0,
+            timestamp_offset_segment=(1.4, 1.7),
+        ),
+    )
+
+
+@pytest.fixture(scope="module")
+def state_only_source_episode(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    return synthesize_episode(
+        tmp_path_factory.mktemp("e2e-state-only-source") / "episode_0001.mcap",
+        SyntheticEpisodeSpec(duration_s=2.0, cameras=(), joint_jump_at_s=1.0),
+    )
 
 
 @pytest.fixture(scope="module")
@@ -109,7 +125,7 @@ def test_builtin_checks_found_the_injected_defects(
     assert blackout is not None
     black_pct = blackout.measurements["black_pct"]
     assert isinstance(black_pct, float)
-    # The fixture blacks out 1s of the 8s wrist stream: ~12.5% of frames.
+    # The fixture blacks out 0.25s of the 2s wrist stream: ~12.5% of frames.
     assert 5.0 < black_pct < 25.0
 
 
@@ -129,7 +145,7 @@ def test_canonical_episode_accessors(
         assert mp4.is_file() and mp4.stat().st_size > 0
 
         frames = episode.frames("overhead_cam", fps=2.0)
-        assert 12 <= len(frames) <= 17  # ~8s at 2 fps
+        assert 3 <= len(frames) <= 5  # ~2s at 2 fps
         assert frames[0].path.read_bytes()[:2] == b"\xff\xd8"
         # Frame log times map to SOURCE messages: at t=0.5s the fps filter
         # emits the frame visible then -- source frame floor(0.5 * 15) = 7,
@@ -141,7 +157,7 @@ def test_arrow_export(report_and_app: tuple[hflow.TestReport, hflow.App]) -> Non
     report, _app = report_and_app
     with hflow.Episode(report.canonical_path) as episode:
         table: Any = episode.channel("/joint_states").to_arrow()
-    assert table.num_rows == 800
+    assert table.num_rows == 200
     assert "log_time_ns" in table.column_names
     assert "position" in table.column_names
 
@@ -171,7 +187,9 @@ def never_reached(ep: hflow.Episode) -> hflow.CheckResult:
     assert by_name["camera_blackout"].status == "failed"
 
 
-def test_crashing_check_is_infrastructure_not_data(source_episode: Path, tmp_path: Path) -> None:
+def test_crashing_check_is_infrastructure_not_data(
+    state_only_source_episode: Path, tmp_path: Path
+) -> None:
     app = hflow.App("crashy-pipeline", data_root=tmp_path)
 
     @app.check()
@@ -182,7 +200,7 @@ def exploding(ep: hflow.Episode) -> hflow.CheckResult:
     def still_runs(ep: hflow.Episode) -> hflow.CheckResult:
         return hflow.CheckResult(measurements={"ran": True})
 
-    report = app.test(source_episode, verbose=False)
+    report = app.test(state_only_source_episode, verbose=False)
     assert not report.quarantined
     by_name = {run.check.name: run for run in report.checks}
     assert by_name["exploding"].status == "error"
@@ -191,7 +209,7 @@ def still_runs(ep: hflow.Episode) -> hflow.CheckResult:
 
 
 def test_resource_declaring_checks_run_after_plain_ones(
-    source_episode: Path, tmp_path: Path
+    state_only_source_episode: Path, tmp_path: Path
 ) -> None:
     app = hflow.App(
         "ordered-pipeline", data_root=tmp_path, endpoints={"judge": "http://localhost:9"}
@@ -208,16 +226,18 @@ def cheap(ep: hflow.Episode) -> hflow.CheckResult:
         execution_order.append("cheap")
         return hflow.CheckResult()
 
-    app.test(source_episode, verbose=False)
+    app.test(state_only_source_episode, verbose=False)
     assert execution_order == ["cheap", "expensive"]
 
 
-def test_missing_provider_alias_fails_preflight(source_episode: Path, tmp_path: Path) -> None:
+def test_missing_provider_alias_fails_preflight(
+    state_only_source_episode: Path, tmp_path: Path
+) -> None:
     app = hflow.App("misconfigured", data_root=tmp_path)
 
     @app.check(uses="judge")
     def needs_endpoint(ep: hflow.Episode) -> hflow.CheckResult:
         return hflow.CheckResult()
 
     with pytest.raises(ValueError, match="judge"):
-        app.test(source_episode, verbose=False)
+        app.test(state_only_source_episode, verbose=False)

tests/test_runtime_client.py

@@ -99,19 +99,34 @@ def log_message(self, format: str, *args: Any) -> None:
         pass
 
 
-@pytest.fixture()
-def stub_server() -> Iterator[str]:
+def _reset_stub_airflow_state() -> None:
     _StubAirflowHandler.issued_tokens = []
     _StubAirflowHandler.requests_seen = []
     _StubAirflowHandler.healthy = True
     _StubAirflowHandler.expire_first_token = False
+
+
+@pytest.fixture(scope="module")
+def stub_server_base_url() -> Iterator[str]:
     server = ThreadingHTTPServer(("127.0.0.1", 0), _StubAirflowHandler)
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
-    thread.start()
+    server_thread = threading.Thread(
+        target=server.serve_forever,
+        kwargs={"poll_interval": 0.01},
+        daemon=True,
+    )
+    server_thread.start()
     try:
         yield f"http://127.0.0.1:{server.server_port}"
     finally:
         server.shutdown()
+        server.server_close()
+        server_thread.join()
+
+
+@pytest.fixture()
+def stub_server(stub_server_base_url: str) -> str:
+    _reset_stub_airflow_state()
+    return stub_server_base_url
 
 
 def test_trigger_fetches_token_once_and_sends_bearer(stub_server: str) -> None:
@@ -176,7 +191,22 @@ def test_health_parses_body_not_status(stub_server: str) -> None:
 
 
 def test_wait_until_healthy_times_out_with_last_status(stub_server: str) -> None:
+    class InstantlyAdvancingClock:
+        def __init__(self) -> None:
+            self.current_time_s = 0.0
+
+        def monotonic(self) -> float:
+            return self.current_time_s
+
+        def sleep(self, duration_s: float) -> None:
+            self.current_time_s += duration_s
+
     _StubAirflowHandler.healthy = False
     client = AirflowClient(stub_server, "airflow", "right-password")
-    with pytest.raises(TimeoutError, match="scheduler=unhealthy"):
+    instantly_advancing_clock = InstantlyAdvancingClock()
+    with (
+        pytest.MonkeyPatch.context() as monkeypatch,
+        pytest.raises(TimeoutError, match="scheduler=unhealthy"),
+    ):
+        monkeypatch.setattr("hflow.runtime._client.time", instantly_advancing_clock)
         client.wait_until_healthy(timeout_s=0.3, poll_interval_s=0.1)