| 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"] | |