File size: 5,156 Bytes
1e251f3 | 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 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """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:
# DEP1-W-20261027T020407Z: an exited 10-22 production container blocked W8 and left production down.
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)
|