test: batch fractional transcript deadline cases

tests/lib/live/test_transcript_grouping.py

@@ -297,60 +297,71 @@ async def test_zero_disables_suppression(make_transcript: Factory) -> None:
 
 
 @pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
-@pytest.mark.parametrize(
-    "options,fragments,expected",
-    [
-        pytest.param(
+def test_fractional_source_deadlines(async_mode: bool) -> None:
+    scenarios = [
+        (
+            "assistant-silence",
             {"assistant_silence_ms": 0.1},
             [("assistant", "First", 0, 200), ("assistant", "Second", 201, 300)],
             [("assistant", "First", "inactivity"), ("assistant", "Second", "manual")],
-            id="assistant-silence",
         ),
-        pytest.param(
+        (
+            "speaker-separation",
             {"min_turn_separation_ms": 0.1},
             [("user", "Question", 0, 200), ("assistant", "Answer", 200, 400), ("user", "More", 800, 1000)],
             [
                 ("user", "Question", "speaker_change"),
                 ("assistant", "Answer", "speaker_change"),
                 ("user", "More", "manual"),
             ],
-            id="speaker-separation",
         ),
-        pytest.param(
+        (
+            "backchannel-isolation",
             {"min_turn_separation_ms": 0, "backchannel_isolation_ms": 0.1},
             [("user", "Question", 0, 150), ("assistant", "okay", 100, 200), ("user", " More", 800, 1000)],
             [("user", "Question More", "manual")],
-            id="backchannel-isolation",
         ),
-    ],
-)
-def test_fractional_source_deadlines(
-    async_mode: bool,
-    options: dict[str, float],
-    fragments: list[tuple[str, str, int, int]],
-    expected: list[tuple[str, str, str]],
-) -> None:
+    ]
     # A synchronous non-progressing loop also blocks asyncio timeouts.
     code = f"""
 import asyncio
+import sys
+import traceback
+from tests.lib.live.helpers import close
 from tests.lib.live.test_transcript_grouping import Transcript, fragment
 
-async def run():
-    t = Transcript({async_mode!r}, **{options!r})
-    for speaker, value, start, end in {fragments!r}:
-        await t.feed(fragment(speaker, value, start, end))
-    await t.finish(*[(speaker, value) for speaker, value, _ in {expected!r}])
-    assert [(c.segment.speaker, c.segment.text, c.reason) for c in t.recording.closed] == {expected!r}
-
-asyncio.run(run())
+async def run(options, fragments, expected):
+    t = Transcript({async_mode!r}, **options)
+    try:
+        for speaker, value, start, end in fragments:
+            await t.feed(fragment(speaker, value, start, end))
+        await t.finish(*[(speaker, value) for speaker, value, _ in expected])
+        assert [(c.segment.speaker, c.segment.text, c.reason) for c in t.recording.closed] == expected
+    finally:
+        await close(t.grouper)
+
+failed = False
+for scenario, options, fragments, expected in {scenarios!r}:
+    print("scenario:", scenario, file=sys.stderr, flush=True)
+    try:
+        asyncio.run(run(options, fragments, expected))
+    except Exception:
+        traceback.print_exc()
+        failed = True
+if failed:
+    raise SystemExit(1)
 """
-    result = subprocess.run(
-        [sys.executable, "-c", code],
-        cwd=Path(__file__).resolve().parents[3],
-        capture_output=True,
-        text=True,
-        timeout=5,
-    )
+    try:
+        result = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd=Path(__file__).resolve().parents[3],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except subprocess.TimeoutExpired as exc:
+        stderr = exc.stderr.decode(errors="replace") if isinstance(exc.stderr, bytes) else exc.stderr
+        pytest.fail(f"fractional source deadlines timed out:\n{stderr or 'no scenario started'}", pytrace=False)
     assert result.returncode == 0, result.stdout + result.stderr