File size: 1,670 Bytes
81ba775 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 | from src import gpu_timing
def test_callback_timing_records_unavailable_queue_wait_without_cuda(monkeypatch):
monkeypatch.setattr(gpu_timing.torch.cuda, "is_available", lambda: False)
timing = gpu_timing.start_gpu_callback_timing()
snapshot = timing.to_dict()
assert snapshot["zerogpu_queue_allocation_wait_seconds"] is None
assert snapshot["zerogpu_queue_allocation_wait_status"].startswith("unavailable-by-design")
assert snapshot["browser_click_to_callback_seconds"] is None
assert snapshot["cuda_ready_status"] == "cuda-unavailable"
assert snapshot["gpu_callback_body_seconds"] >= 0
def test_callback_timing_synchronizes_once_at_entry(monkeypatch):
calls = []
monkeypatch.setattr(gpu_timing.torch.cuda, "is_available", lambda: True)
monkeypatch.setattr(gpu_timing.torch.cuda, "synchronize", lambda: calls.append("sync"))
timing = gpu_timing.start_gpu_callback_timing()
snapshot = timing.to_dict()
assert calls == ["sync"]
assert snapshot["cuda_ready_status"] == "ready-after-synchronize"
assert snapshot["cuda_initial_sync_seconds"] is not None
assert snapshot["callback_entry_to_cuda_ready_seconds"] is not None
def test_synchronized_wall_time_wraps_operation_with_cuda_sync(monkeypatch):
events = []
monkeypatch.setattr(gpu_timing.torch.cuda, "synchronize", lambda: events.append("sync"))
def operation():
events.append("operation")
return "ok"
result, elapsed, error = gpu_timing.synchronized_wall_time(operation, use_cuda=True)
assert result == "ok"
assert elapsed >= 0
assert error is None
assert events == ["sync", "operation", "sync"]
|