File size: 4,644 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 | """WIN2-R3 smoke start: shared sequence, verdicts, always-stop, chain-head and x-first modes (fake backend, CPU only)."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from offload.launch.windows import smoke_start as sm
from offload.prefill_maxlen.evidence import Json, record
class Fake:
def __init__(self, fail: dict[str, str] | None = None, *, rounds: float = 30.0, tokens: int = sm.SMOKE_TOKENS) -> None:
self.fail = fail or {}
self.rounds, self.tokens = rounds, tokens
self.drafts = 0.0
self.calls: list[tuple[str, str]] = []
def _maybe(self, stage: str, arm: str) -> None:
self.calls.append((stage, arm))
if self.fail.get(arm) == stage:
raise RuntimeError(f'{stage} failed for {arm}')
def start(self, arm: str) -> None:
self._maybe('start', arm)
def healthy(self, arm: str) -> dict[str, Json]:
self._maybe('healthy', arm)
return {'id': 'fixture', 'max_model_len': 327680}
def greedy(self, arm: str) -> dict[str, Json]:
self._maybe('greedy', arm)
return {'sha256': 'a' * 64, 'tokens': 300}
def seeded(self, arm: str, body: dict[str, Json]) -> dict[str, Json]:
self._maybe('seeded', arm)
assert body['seed'] == sm.SMOKE_SEED and body['max_tokens'] == sm.SMOKE_TOKENS and body['ignore_eos'] is True
self.drafts += self.rounds
return {'usage': {'completion_tokens': self.tokens}, 'output_sha256': 'b' * 64}
def metrics(self, arm: str) -> str:
return f'vllm:spec_decode_num_drafts_total{{m="x"}} {self.drafts}\n'
def stop(self, arm: str) -> None:
self._maybe('stop', arm)
def test_pass_runs_every_stage_and_stops() -> None:
fake = Fake()
r = sm.smoke(fake, 'X')
assert r['status'] == 'PASS' and r['stage'] == 'done' and record(r['seeded'])['rounds'] == 30.0
assert [c[0] for c in fake.calls] == ['start', 'healthy', 'greedy', 'seeded', 'stop']
@pytest.mark.parametrize('stage', ['start', 'healthy', 'greedy', 'seeded'])
def test_any_stage_failure_is_fail_and_still_stops(stage: str) -> None:
fake = Fake({'X': stage})
r = sm.smoke(fake, 'X')
assert r['status'] == 'FAIL' and r['stage'] == stage and stage in str(r['error'])
assert fake.calls[-1] == ('stop', 'X')
@pytest.mark.parametrize('fake', [Fake(rounds=0.0), Fake(tokens=12)])
def test_seeded_request_must_run_speculation_to_length(fake: Fake) -> None:
r = sm.smoke(fake, 'X')
assert r['status'] == 'FAIL' and r['stage'] == 'seeded'
def test_stop_failure_raises_with_the_result() -> None:
with pytest.raises(RuntimeError, match='arm stop failed; HOLD stays'):
sm.smoke(Fake({'X': 'stop'}), 'X')
def test_budget_exhaustion_fails_and_stops() -> None:
times = iter([0.0, 0.0, 1000.0, 1000.0, 1000.0, 1000.0])
fake = Fake()
r = sm.smoke(fake, 'X', clock=lambda: next(times), budget_s=10.0)
assert r['status'] == 'FAIL' and 'budget' in str(r['error']) and fake.calls[-1] == ('stop', 'X')
def test_chain_head_drops_failing_candidates() -> None:
fake = Fake({'B': 'healthy'})
r = sm.chain_head(fake, ['A', 'B', 'C'])
assert r['passed'] == ['A', 'C'] and r['dropped'] == ['B']
assert [c for c in fake.calls if c[0] == 'stop'] == [('stop', 'A'), ('stop', 'B'), ('stop', 'C')]
with pytest.raises(ValueError, match='distinct'):
sm.chain_head(fake, ['A', 'A'])
def test_x_first_ends_the_window_on_failure() -> None:
assert sm.x_first(Fake(), 'X')['window'] == 'CONTINUE'
assert str(sm.x_first(Fake({'X': 'greedy'}), 'X')['window']).startswith('END')
def factory(experiment: str, arms: list[str], out: Path) -> Fake:
return Fake({'bad': 'start'})
def test_cli_dry_plan_and_coordinator_execution(tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str]) -> None:
assert sm.main(['PC1', 'chunk16384']) == 0 and 'DRY RUN' in capsys.readouterr().out
out = tmp_path / 'smoke'
args = ['PC1', 'chunk16384', 'bad', '--mode', 'chain-head', '--execute', '--out', str(out),
'--backend', f'{__name__}:factory']
monkeypatch.delenv('WINDOW_COORDINATOR', raising=False)
assert sm.main(args) == 2
monkeypatch.setenv('WINDOW_COORDINATOR', sm.COORDINATOR)
assert sm.main(args) == 0
saved = json.loads((out / 'smoke.json').read_text())
assert saved['passed'] == ['chunk16384'] and saved['dropped'] == ['bad']
with pytest.raises(SystemExit):
sm.main(['PC1', 'a', 'b']) # x-first smokes exactly one arm
|