Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 63 additions & 51 deletions tests/training/test_cuda_mps_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import socket
import stat
import tempfile
from collections.abc import Iterator
from pathlib import Path
from types import SimpleNamespace
from typing import Any
Expand Down Expand Up @@ -56,11 +57,18 @@ def torch_with_uuid(uuid: str = GPU_A) -> Any:


@pytest.fixture
def linux_host(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
def short_temp_path() -> Iterator[Path]:
temporary_root = Path("/tmp").resolve()
with tempfile.TemporaryDirectory(prefix="ucmps-", dir=temporary_root) as raw:
yield Path(raw)


@pytest.fixture
def linux_host(monkeypatch: pytest.MonkeyPatch, short_temp_path: Path) -> Path:
fake_platform(monkeypatch)
root = tmp_path / "registry"
monkeypatch.setenv("HOME", str(tmp_path / "home"))
(tmp_path / "home").mkdir()
root = short_temp_path / "registry"
monkeypatch.setenv("HOME", str(short_temp_path / "home"))
(short_temp_path / "home").mkdir()
return root


Expand Down Expand Up @@ -137,9 +145,11 @@ def test_missing_selector_rejects_ambiguous_multi_gpu_host() -> None:
mps.resolve_topology(None, run_command=fake_run())


def test_start_rejects_task_per_gpu_until_lease_contract_lands() -> None:
def test_start_rejects_task_per_gpu_until_lease_contract_lands(
linux_host: Path,
) -> None:
with pytest.raises(mps.CudaMpsError, match="current support is single_gpu"):
mps.start_daemon("all", root=Path("/unused"), run_command=fake_run())
mps.start_daemon("all", root=linux_host, run_command=fake_run())


def test_status_is_read_only_and_reports_compatible_topology_modes(
Expand Down Expand Up @@ -366,7 +376,7 @@ def test_default_daemon_name_uses_complete_uuid() -> None:


def test_start_daemon_rejects_long_unix_socket_path(linux_host: Path) -> None:
long_dir = linux_host / ("x" * 96)
long_dir = linux_host / ("x" * 200)

with pytest.raises(mps.CudaMpsError, match="MPS control socket path is too long"):
mps.start_daemon(
Expand All @@ -380,9 +390,9 @@ def test_start_daemon_rejects_long_unix_socket_path(linux_host: Path) -> None:


def test_start_daemon_recovers_only_inert_failed_control_artifacts(
linux_host: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
linux_host: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
short_root = tmp_path.parent / "u"
short_root = linux_host.parent / "u"
monkeypatch.setattr(mps.tempfile, "gettempdir", lambda: str(short_root))
monkeypatch.setattr(mps.shutil, "which", lambda _name: "/fake/bin/nvidia-cuda-mps-control")
pipe_dir = short_root / "p" / "prod" / "pipe"
Expand Down Expand Up @@ -410,10 +420,10 @@ def test_start_daemon_recovers_only_inert_failed_control_artifacts(


def test_start_daemon_failure_reports_control_log_and_cleans_artifacts(
linux_host: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
linux_host: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(mps.shutil, "which", lambda _name: "/fake/bin/nvidia-cuda-mps-control")
short_root = tmp_path.parent / "u2"
short_root = linux_host.parent / "u2"
monkeypatch.setattr(mps.tempfile, "gettempdir", lambda: str(short_root))
pipe_dir = short_root / "p" / "prod" / "pipe"
log_dir = short_root / "p" / "prod" / "log"
Expand Down Expand Up @@ -442,58 +452,60 @@ def run(command: list[str], **kwargs: Any) -> Completed:
assert not (pipe_dir / "control_lock").exists()


def test_default_runtime_recovers_orphan_control_sockets() -> None:
with tempfile.TemporaryDirectory() as raw:
root = Path(raw)
monkeypatch_root = root / "tmp"
runtime = monkeypatch_root / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
control = runtime / "control"
# Create actual socket nodes like NVIDIA leaves behind.
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(runtime / "control_privileged"))
def test_default_runtime_recovers_orphan_control_sockets(
short_temp_path: Path,
) -> None:
root = short_temp_path
monkeypatch_root = root / "tmp"
runtime = monkeypatch_root / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
control = runtime / "control"
# Create actual socket nodes like NVIDIA leaves behind.
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(runtime / "control_privileged"))

with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setattr(mps.tempfile, "gettempdir", lambda: str(monkeypatch_root))
root_path = mps._default_runtime_root("gpu-test")
mps._remove_orphan_default_control_sockets(root_path / "pipe")
with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setattr(mps.tempfile, "gettempdir", lambda: str(monkeypatch_root))
root_path = mps._default_runtime_root("gpu-test")
mps._remove_orphan_default_control_sockets(root_path / "pipe")

assert not control.exists()
assert not (runtime / "control_privileged").exists()
assert not control.exists()
assert not (runtime / "control_privileged").exists()


def test_default_runtime_preserves_live_control_with_stale_pid_file(
monkeypatch: pytest.MonkeyPatch,
short_temp_path: Path,
) -> None:
with tempfile.TemporaryDirectory() as raw:
runtime = Path(raw) / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
(runtime / "nvidia-cuda-mps-control.pid").write_text("4242\n", encoding="utf-8")
control = runtime / "control"
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))

monkeypatch.setattr(mps, "_proc_stat", live_proc_stat())
monkeypatch.setattr(
mps,
"_process_name_is_alive",
lambda pid, name: pid == 4242 and name.startswith("nvidia"),
)
mps._remove_orphan_default_control_sockets(runtime)
runtime = short_temp_path / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
(runtime / "nvidia-cuda-mps-control.pid").write_text("4242\n", encoding="utf-8")
control = runtime / "control"
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))

assert control.exists()
monkeypatch.setattr(mps, "_proc_stat", live_proc_stat())
monkeypatch.setattr(
mps,
"_process_name_is_alive",
lambda pid, name: pid == 4242 and name.startswith("nvidia"),
)
mps._remove_orphan_default_control_sockets(runtime)

assert control.exists()


def test_default_runtime_removes_orphan_socket_with_stale_pid_file() -> None:
with tempfile.TemporaryDirectory() as raw:
runtime = Path(raw) / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
(runtime / "nvidia-cuda-mps-control.pid").write_text("99999999\n", encoding="utf-8")
control = runtime / "control"
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))
def test_default_runtime_removes_orphan_socket_with_stale_pid_file(
short_temp_path: Path,
) -> None:
runtime = short_temp_path / "uni-cumps" / "gpu-test" / "pipe"
runtime.mkdir(parents=True)
(runtime / "nvidia-cuda-mps-control.pid").write_text("99999999\n", encoding="utf-8")
control = runtime / "control"
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).bind(str(control))

mps._remove_orphan_default_control_sockets(runtime)
mps._remove_orphan_default_control_sockets(runtime)

assert not control.exists()
assert not control.exists()


def test_start_daemon_refuses_existing_unmanaged_control_path(
Expand Down
Loading