| | |
| | | |
| | | import importlib.util |
| | | import io |
| | | import subprocess |
| | | import sys |
| | | import tempfile |
| | | import unittest |
| | | import json |
| | | from unittest import mock |
| | | from http import HTTPStatus |
| | | from urllib.parse import quote |
| | | from pathlib import Path |
| | |
| | | with self.assertRaisesRegex(MODULE.ApiError, "Processing failed"): |
| | | handler.run_command([sys.executable, "-c", "raise SystemExit(2)"], timeout=10) |
| | | |
| | | def test_pointcloud_execution_uses_fixed_gpu_environment_when_cuda_probe_succeeds(self) -> None: |
| | | with tempfile.TemporaryDirectory() as temp_dir: |
| | | root = Path(temp_dir) |
| | | gpu_python = root / ".venvs" / MODULE.POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe" |
| | | gpu_python.parent.mkdir(parents=True) |
| | | gpu_python.write_bytes(b"fixed-interpreter") |
| | | with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": true, "torch": "2.11.0+cu128"}\n', "")): |
| | | execution = MODULE.pointcloud_execution_environment(root) |
| | | self.assertEqual(execution["device"], "cuda") |
| | | self.assertEqual(execution["environment"], MODULE.POINTCLOUD_GPU_ENVIRONMENT) |
| | | self.assertEqual(execution["torchVersion"], "2.11.0+cu128") |
| | | self.assertEqual(Path(execution["python"]), gpu_python) |
| | | |
| | | def test_pointcloud_execution_falls_back_to_fixed_cpu_environment(self) -> None: |
| | | with tempfile.TemporaryDirectory() as temp_dir: |
| | | root = Path(temp_dir) |
| | | cpu_python = root / ".venvs" / MODULE.POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe" |
| | | gpu_python = root / ".venvs" / MODULE.POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe" |
| | | cpu_python.parent.mkdir(parents=True) |
| | | gpu_python.parent.mkdir(parents=True) |
| | | cpu_python.write_bytes(b"fixed-cpu-interpreter") |
| | | gpu_python.write_bytes(b"fixed-gpu-interpreter") |
| | | with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")): |
| | | execution = MODULE.pointcloud_execution_environment(root) |
| | | self.assertEqual(execution["device"], "cpu") |
| | | self.assertEqual(execution["environment"], MODULE.POINTCLOUD_CPU_ENVIRONMENT) |
| | | with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")): |
| | | with self.assertRaisesRegex(MODULE.ApiError, "CUDA was requested"): |
| | | MODULE.pointcloud_execution_environment(root, "cuda") |
| | | |
| | | def test_semantic_validation_run_is_discovered(self) -> None: |
| | | runs = MODULE.semantic_runs(ROOT) |
| | | self.assertTrue(any(item["id"] == "validation-20260817" for item in runs)) |