Test that _trace is thread-safe
(self, tmp_path)
| 408 | obj._next_replay_event(step_id="step2", iteration=0) |
| 409 | |
| 410 | def test_trace_thread_safe(self, tmp_path): |
| 411 | """Test that _trace is thread-safe""" |
| 412 | obj = self.TracingTestClass() |
| 413 | obj._trace_writer = TraceWriter(tmp_path / "trace.jsonl") |
| 414 | |
| 415 | results = [] |
| 416 | |
| 417 | def write_events(prefix, count): |
| 418 | for i in range(count): |
| 419 | obj._trace({"prefix": prefix, "id": i}) |
| 420 | results.append(prefix) |
| 421 | |
| 422 | threads = [ |
| 423 | threading.Thread(target=write_events, args=("a", 10)), |
| 424 | threading.Thread(target=write_events, args=("b", 10)), |
| 425 | threading.Thread(target=write_events, args=("c", 10)), |
| 426 | ] |
| 427 | |
| 428 | for t in threads: |
| 429 | t.start() |
| 430 | for t in threads: |
| 431 | t.join() |
| 432 | |
| 433 | events = list(iter_jsonl(tmp_path / "trace.jsonl")) |
| 434 | assert len(events) == 30 |
nothing calls this directly
no test coverage detected