test: refactor tests for readability and maintainability

- Convert repetitive test methods to parametrized tests
- Split test_reporter.py into focused modules:
  test_reporter_report.py, test_reporter_rollup.py, test_reporter_stop_perf.py
- Consolidate duplicate test patterns across test suites
- Improve test organization in transforms, extractor, types, record tests

Author: nv-alicheng

tests/unit/async_utils/services/event_logger/test_event_logger.py

@@ -118,17 +118,17 @@ def _make_stub(*args, **kwargs) -> tuple[StubEventLoggerService, list[FakeWriter
 
 @pytest.mark.unit
 class TestIsErrorEvent:
-    def test_error_event_types(self):
-        for et in ErrorEventType:
-            assert _is_error_event(_record(et)) is True
-
-    def test_session_events_are_not_errors(self):
-        for et in SessionEventType:
-            assert _is_error_event(_record(et)) is False
-
-    def test_sample_events_are_not_errors(self):
-        for et in SampleEventType:
-            assert _is_error_event(_record(et)) is False
+    @pytest.mark.parametrize(
+        "case_desc, event_type_class, expected",
+        [
+            ("error events", ErrorEventType, True),
+            ("session events", SessionEventType, False),
+            ("sample events", SampleEventType, False),
+        ],
+    )
+    def test_is_error_event(self, case_desc, event_type_class, expected):
+        for et in event_type_class:
+            assert _is_error_event(_record(et)) is expected
 
 
 # ---------------------------------------------------------------------------
@@ -181,7 +181,7 @@ async def test_multiple_batches_accumulate(self):
 @pytest.mark.unit
 class TestShutdownBehavior:
     @pytest.mark.asyncio
-    async def test_session_ended_triggers_flush_and_close(self):
+    async def test_ended_triggers_flush_close(self):
         service, writers = _make_stub()
         await service.process([_record(SessionEventType.ENDED, ts=100)])
         for writer in writers:
@@ -216,7 +216,7 @@ async def test_events_after_ended_are_dropped(self):
         assert len(new_writer.written) == 0
 
     @pytest.mark.asyncio
-    async def test_non_error_events_after_ended_in_same_batch_dropped(self):
+    async def test_non_errors_after_ended_same_batch(self):
         service, writers = _make_stub()
         await service.process(
             [
@@ -230,7 +230,7 @@ async def test_non_error_events_after_ended_in_same_batch_dropped(self):
             assert writer.written[0].event_type == SessionEventType.ENDED
 
     @pytest.mark.asyncio
-    async def test_error_events_after_ended_in_same_batch_still_written(self):
+    async def test_errors_after_ended_same_batch_kept(self):
         service, writers = _make_stub()
         err_data = ErrorData(error_type="TestError", error_message="boom")
         await service.process(
@@ -244,7 +244,7 @@ async def test_error_events_after_ended_in_same_batch_still_written(self):
             assert writer.written[1].event_type == ErrorEventType.GENERIC
 
     @pytest.mark.asyncio
-    async def test_error_events_after_ended_in_later_batch_dropped(self):
+    async def test_errors_after_ended_later_batch(self):
         """Error events are only kept in the same batch as ENDED.
 
         After the batch containing ENDED completes, writers are closed and
@@ -519,47 +519,42 @@ async def test_full_lifecycle(self, tmp_path):
 @pytest.mark.unit
 class TestEdgeCases:
     @pytest.mark.asyncio
-    async def test_all_error_event_types_are_recognized(self):
-        service, writers = _make_stub()
-        error_records = [_record(et, ts=i) for i, et in enumerate(ErrorEventType)]
-        await service.process(error_records)
-        for writer in writers:
-            assert len(writer.written) == len(list(ErrorEventType))
-
-    @pytest.mark.asyncio
-    async def test_all_session_event_types_written(self):
-        service, writers = _make_stub()
-        session_records = [_record(et, ts=i) for i, et in enumerate(SessionEventType)]
-        await service.process(session_records)
-        for writer in writers:
-            # All session events should be written
-            # (ENDED is among them but everything in the batch up to and including ENDED is written)
-            assert len(writer.written) == len(list(SessionEventType))
-
-    @pytest.mark.asyncio
-    async def test_all_sample_event_types_written(self):
+    @pytest.mark.parametrize(
+        "case_desc, event_type_class, use_uuid",
+        [
+            ("error events", ErrorEventType, False),
+            ("session events", SessionEventType, False),
+            ("sample events", SampleEventType, True),
+        ],
+    )
+    async def test_all_event_types_written(self, case_desc, event_type_class, use_uuid):
         service, writers = _make_stub()
-        sample_records = [
-            _record(et, uuid="s1", ts=i) for i, et in enumerate(SampleEventType)
+        records = [
+            _record(et, uuid="s1" if use_uuid else "", ts=i)
+            for i, et in enumerate(event_type_class)
         ]
-        await service.process(sample_records)
-        for writer in writers:
-            assert len(writer.written) == len(list(SampleEventType))
-
-    @pytest.mark.asyncio
-    async def test_record_with_no_data(self):
-        service, writers = _make_stub()
-        await service.process([_record(SampleEventType.ISSUED, uuid="s1")])
+        await service.process(records)
         for writer in writers:
-            assert writer.written[0].data is None
+            assert len(writer.written) == len(list(event_type_class))
 
     @pytest.mark.asyncio
-    async def test_record_with_error_data(self):
+    @pytest.mark.parametrize(
+        "case_desc, event_type, data, expected_data",
+        [
+            ("no data", SampleEventType.ISSUED, None, None),
+            (
+                "error data",
+                ErrorEventType.CLIENT,
+                ErrorData(error_type="SomeError", error_message="detail"),
+                ErrorData(error_type="SomeError", error_message="detail"),
+            ),
+        ],
+    )
+    async def test_record_data(self, case_desc, event_type, data, expected_data):
         service, writers = _make_stub()
-        err = ErrorData(error_type="SomeError", error_message="detail")
-        await service.process([_record(ErrorEventType.CLIENT, data=err)])
+        await service.process([_record(event_type, uuid="s1", data=data)])
         for writer in writers:
-            assert writer.written[0].data == err
+            assert writer.written[0].data == expected_data
 
     @pytest.mark.asyncio
     async def test_large_batch(self):

tests/unit/async_utils/services/event_logger/test_sql_writer.py

@@ -47,32 +47,38 @@ def _record(event_type, uuid="", ts=0, data=None):
 
 @pytest.mark.unit
 class TestRecordToRow:
-    def test_sample_event_topic(self):
-        row = _record_to_row(_record(SampleEventType.ISSUED, uuid="s1", ts=1000))
-        assert row.event_type == "sample.issued"
-        assert row.sample_uuid == "s1"
-        assert row.timestamp_ns == 1000
-
-    def test_session_event_topic(self):
-        row = _record_to_row(_record(SessionEventType.ENDED, ts=42))
-        assert row.event_type == "session.ended"
-        assert row.sample_uuid == ""
-        assert row.timestamp_ns == 42
-
-    def test_error_event_topic(self):
-        row = _record_to_row(_record(ErrorEventType.GENERIC, ts=99))
-        assert row.event_type == "error.generic"
-
-    def test_data_is_json_encoded(self):
-        err = ErrorData(error_type="TestError", error_message="boom")
-        row = _record_to_row(_record(SampleEventType.COMPLETE, data=err))
-        decoded = msgspec.json.decode(row.data)
-        assert "TestError" in str(decoded)
-
-    def test_none_data_encodes_to_null(self):
-        row = _record_to_row(_record(SampleEventType.ISSUED))
+    @pytest.mark.parametrize(
+        "case_desc, event_type, uuid, ts, expected_topic",
+        [
+            ("sample event", SampleEventType.ISSUED, "s1", 1000, "sample.issued"),
+            ("session event", SessionEventType.ENDED, "", 42, "session.ended"),
+            ("error event", ErrorEventType.GENERIC, "", 99, "error.generic"),
+        ],
+    )
+    def test_topic_and_fields(self, case_desc, event_type, uuid, ts, expected_topic):
+        row = _record_to_row(_record(event_type, uuid=uuid, ts=ts))
+        assert row.event_type == expected_topic
+        assert row.sample_uuid == uuid
+        assert row.timestamp_ns == ts
+
+    @pytest.mark.parametrize(
+        "case_desc, data, check_str",
+        [
+            (
+                "error data",
+                ErrorData(error_type="TestError", error_message="boom"),
+                "TestError",
+            ),
+            ("none data", None, None),
+        ],
+    )
+    def test_data_encoding(self, case_desc, data, check_str):
+        row = _record_to_row(_record(SampleEventType.COMPLETE, data=data))
         decoded = msgspec.json.decode(row.data)
-        assert decoded is None
+        if check_str is not None:
+            assert check_str in str(decoded)
+        else:
+            assert decoded is None
 
 
 # ---------------------------------------------------------------------------

tests/unit/async_utils/services/metrics_aggregator/test_aggregator.py

@@ -270,7 +270,7 @@ async def test_all_timing_metrics_full_lifecycle(self):
         assert m["chunk_delta_ns"] == 1000
 
     @pytest.mark.asyncio
-    async def test_chunk_delta_not_emitted_without_last_recv(self):
+    async def test_chunk_delta_needs_last_recv(self):
         """RECV_NON_FIRST without prior RECV_FIRST: no chunk_delta emitted."""
         emitter = FakeEmitter()
         agg = StubAggregator(emitter)
@@ -285,7 +285,7 @@ async def test_chunk_delta_not_emitted_without_last_recv(self):
         assert row.last_recv_ns is None  # No recv events yet
 
     @pytest.mark.asyncio
-    async def test_request_duration_not_emitted_without_client_send(self):
+    async def test_req_duration_needs_client_send(self):
         """CLIENT_RESP_DONE without CLIENT_SEND: no request_duration."""
         emitter = FakeEmitter()
         agg = StubAggregator(emitter)
@@ -326,7 +326,7 @@ async def test_issued_stores_prompt_text(self):
         assert row.prompt_text == "What is AI?"
 
     @pytest.mark.asyncio
-    async def test_issued_with_token_ids_emits_isl_directly(self):
+    async def test_token_ids_emit_isl_directly(self):
         """SGLang path: PromptData with token_ids emits ISL = len(token_ids)
         without tokenization."""
         emitter = FakeEmitter()

tests/unit/async_utils/services/metrics_aggregator/test_metrics_table.py

@@ -39,53 +39,47 @@ def test_is_msgspec_struct(self):
         row = SampleRow("s1")
         assert isinstance(row, msgspec.Struct)
 
-    def test_ttft(self):
-        row = SampleRow("s1")
-        row.issued_ns = 1000
-        row.recv_first_ns = 2500
-        assert row.ttft_ns() == 1500
-
-    def test_ttft_returns_none_without_issued(self):
-        row = SampleRow("s1")
-        row.recv_first_ns = 2500
-        assert row.ttft_ns() is None
-
-    def test_ttft_returns_none_without_recv_first(self):
-        row = SampleRow("s1")
-        row.issued_ns = 1000
-        assert row.ttft_ns() is None
-
-    def test_sample_latency(self):
-        row = SampleRow("s1")
-        row.issued_ns = 1000
-        row.complete_ns = 5000
-        assert row.sample_latency_ns() == 4000
-
-    def test_sample_latency_returns_none_without_issued(self):
-        row = SampleRow("s1")
-        row.complete_ns = 5000
-        assert row.sample_latency_ns() is None
-
-    def test_sample_latency_returns_none_without_complete(self):
-        row = SampleRow("s1")
-        row.issued_ns = 1000
-        assert row.sample_latency_ns() is None
-
-    def test_request_duration(self):
-        row = SampleRow("s1")
-        row.client_send_ns = 100
-        row.client_resp_done_ns = 600
-        assert row.request_duration_ns() == 500
-
-    def test_request_duration_returns_none_without_send(self):
-        row = SampleRow("s1")
-        row.client_resp_done_ns = 600
-        assert row.request_duration_ns() is None
-
-    def test_request_duration_returns_none_without_resp_done(self):
-        row = SampleRow("s1")
-        row.client_send_ns = 100
-        assert row.request_duration_ns() is None
+    @pytest.mark.parametrize(
+        "case_desc, issued, recv_first, expected",
+        [
+            ("both set", 1000, 2500, 1500),
+            ("no issued", None, 2500, None),
+            ("no recv_first", 1000, None, None),
+        ],
+    )
+    def test_ttft(self, case_desc, issued, recv_first, expected):
+        row = SampleRow("s1")
+        row.issued_ns = issued
+        row.recv_first_ns = recv_first
+        assert row.ttft_ns() == expected
+
+    @pytest.mark.parametrize(
+        "case_desc, issued, complete, expected",
+        [
+            ("both set", 1000, 5000, 4000),
+            ("no issued", None, 5000, None),
+            ("no complete", 1000, None, None),
+        ],
+    )
+    def test_sample_latency(self, case_desc, issued, complete, expected):
+        row = SampleRow("s1")
+        row.issued_ns = issued
+        row.complete_ns = complete
+        assert row.sample_latency_ns() == expected
+
+    @pytest.mark.parametrize(
+        "case_desc, send, resp_done, expected",
+        [
+            ("both set", 100, 600, 500),
+            ("no send", None, 600, None),
+            ("no resp_done", 100, None, None),
+        ],
+    )
+    def test_request_duration(self, case_desc, send, resp_done, expected):
+        row = SampleRow("s1")
+        row.client_send_ns = send
+        row.client_resp_done_ns = resp_done
+        assert row.request_duration_ns() == expected
 
     def test_output_text_empty(self):
         row = SampleRow("s1")