| """S54 FIX1: file/in-memory Docker and GPU fixtures; no real daemon or CUDA.""" |
| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| import time |
| from typing import cast |
|
|
| import pytest |
|
|
| from offload.launch.windows import spark_quiet as q |
| from offload.launch.windows.tests.test_post_close_watch import Clock |
|
|
| CID = 'a' * 64 |
|
|
|
|
| @pytest.mark.parametrize('marker', ['NBW EXPERIMENT ABSENT', 'WATCH SPARK QUIET']) |
| @pytest.mark.parametrize('command', [ |
| "from vllm_exl3_moe import tf_experts as t, mrow; t._ext(); mrow.ext()", |
| "import vllm.models.deepseek_v41.common.engram as e; e._load_engram_remote_ext()", |
| ]) |
| @pytest.mark.parametrize('pause', ['import', 'build-lock', 'active-CUDA']) |
| def test_anonymous_jit_waits_until_original_finishes( |
| monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], |
| marker: str, command: str, pause: str, |
| ) -> None: |
| clock = Clock() |
| monkeypatch.setattr(time, 'time', clock) |
| monkeypatch.setattr(time, 'sleep', clock.sleep) |
| calls: list[list[str]] = [] |
|
|
| def checked(argv: list[str], timeout: float) -> str: |
| calls.append(argv) |
| assert timeout > 0 |
| if argv[1] == 'ps': |
| return CID if clock() < 10 else '' |
| if argv[1] == 'inspect': |
| assert argv[2] == CID |
| return json.dumps([{'Id': CID, 'Name': '/patient_babbage', 'Path': 'python3', |
| 'Args': ['-c', command], 'Config': {'Cmd': ['-c', command]}, |
| 'State': {'Status': 'running'}}]) |
| assert argv[0] == 'nvidia-smi' |
| return '123, python3' if pause == 'active-CUDA' and clock() < 10 else '' |
|
|
| monkeypatch.setattr(q, 'checked', checked) |
| q.wait(20, marker) |
| assert clock() == 10 |
| assert capsys.readouterr().out.strip() == marker |
| assert all(argv[1] in ('ps', 'inspect', '--query-compute-apps=pid,process_name') for argv in calls) |
|
|
|
|
| @pytest.mark.parametrize('marker', ['NBW EXPERIMENT ABSENT', 'WATCH SPARK QUIET']) |
| @pytest.mark.parametrize('failure', ['enumerate', 'inspect', 'gpu', 'malformed', 'identity']) |
| def test_unproved_docker_or_gpu_state_keeps_hold( |
| monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], marker: str, failure: str, |
| ) -> None: |
| def checked(argv: list[str], timeout: float) -> str: |
| if argv[1] == 'ps': |
| if failure == 'enumerate': |
| raise subprocess.CalledProcessError(1, argv) |
| return CID |
| if argv[1] == 'inspect': |
| if failure == 'inspect': |
| raise subprocess.CalledProcessError(1, argv) |
| if failure == 'malformed': |
| return '{}' |
| return json.dumps([{'Id': 'b' * 64 if failure == 'identity' else CID, |
| 'Name': '/unrelated', 'Path': 'sleep', 'Args': ['1'], 'Config': {}, |
| 'State': {'Status': 'running'}}]) |
| raise subprocess.CalledProcessError(1, argv) |
| monkeypatch.setattr(q, 'checked', checked) |
| with pytest.raises((ValueError, subprocess.CalledProcessError)): |
| q.wait(time.time() + 10, marker) |
| assert not capsys.readouterr().out |
|
|
|
|
| def test_unrelated_container_untouched_and_compiler_never_killed(monkeypatch: pytest.MonkeyPatch) -> None: |
| clock = Clock() |
| monkeypatch.setattr(time, 'time', clock) |
| monkeypatch.setattr(time, 'sleep', clock.sleep) |
| row: dict[str, object] = {'Id': CID, 'Name': '/unrelated', 'Path': 'sleep', 'Args': ['300'], |
| 'Config': {}, 'State': {'Status': 'running'}} |
| def checked(argv: list[str], timeout: float) -> str: |
| if argv[1] == 'ps': |
| return CID |
| if argv[1] == 'inspect': |
| return json.dumps([row]) |
| assert argv[0] == 'nvidia-smi' |
| return '' |
| monkeypatch.setattr(q, 'checked', checked) |
| assert q.snapshot(10) |
| cast(dict[str, object], row['Config'])['Env'] = ['DSV41_ENGRAM_OVERLAY=/reviewed/engram.py'] |
| with pytest.raises(ValueError, match='exhausted'): |
| q.wait(12, 'must not pass') |
| assert clock() == 12 and row['State'] == {'Status': 'running'} |
|
|
|
|
| @pytest.mark.parametrize('status,quiet', [('exited', True), ('dead', True), ('running', False), ('created', False), |
| ('paused', False), ('restarting', False)]) |
| @pytest.mark.parametrize('name', ['/dsv41-serve-stopped-20261022', '/dsv41-serve', '/dsv41-exp-nbw']) |
| def test_only_live_experiment_or_serve_containers_block_restore( |
| monkeypatch: pytest.MonkeyPatch, status: str, quiet: bool, name: str, |
| ) -> None: |
| |
| def checked(argv: list[str], timeout: float) -> str: |
| if argv[1] == 'ps': |
| return CID |
| if argv[1] == 'inspect': |
| return json.dumps([{'Id': CID, 'Name': name, 'Path': 'python3', 'Args': ['-m', 'vllm'], |
| 'Config': {}, 'State': {'Status': status}}]) |
| return '' |
| monkeypatch.setattr(q, 'checked', checked) |
| if quiet: |
| assert q.snapshot(10) |
| else: |
| with pytest.raises(ValueError, match='experiment/serve container remains'): |
| q.snapshot(10) |
|
|
|
|
