File size: 5,318 Bytes
e1acc15 | 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 | """CPU checks of the PP4 #61161 verifier-RNG gate receipt (the real tests run under pp4_rng_gate --execute)."""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from typing import cast
import pytest
from offload.dspark.probabilistic.pp3 import prepare_verifier as pp3
from offload.launch.windows import pf1_window as pf
from offload.launch.windows import pp4_rng_gate as g
from offload.prefill_maxlen.evidence import Json
HEAD = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=pf.ROOT, text=True).strip()
needs_image = pytest.mark.skipif(not (g.IMAGE_SRC / pp3.UTILS).is_file(), reason='image verifier source absent')
def junit(path: Path, counts: dict[str, int], *, fail: str = '') -> None:
cases = ''.join(f'<testcase classname="offload.audit.verifier_rng.{module}" name="t{i}">'
+ ('<failure/>' if module == fail and i == 0 else '') + '</testcase>'
for module, n in counts.items() for i in range(n))
path.write_text(f'<testsuites><testsuite>{cases}</testsuite></testsuites>')
def fake_runner(counts: dict[str, int], rc: int = 0, fail: str = '') -> object:
def run(argv: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
assert argv[1:4] == ['-m', 'pytest', '-q'] and list(g.TESTS) == argv[-2:]
env = cast(dict[str, str], kwargs['env'])
assert env['TRITON_INTERPRET'] == '1' and env['CUDA_VISIBLE_DEVICES'] == ''
junit(Path(next(a for a in argv if a.startswith('--junitxml=')).split('=', 1)[1]), counts, fail=fail)
return subprocess.CompletedProcess(argv, rc, '', '')
return run
@needs_image
def test_static_checks_on_the_generated_pp4_kernel() -> None:
result = g.static_checks(g.pp4_kernel())
assert result == {'acceptance_draws': 1, 'block_refusal': True, 'step_keyed': False, 'pass': True}
broken = g.pp4_kernel().replace(g.BLOCK_REFUSAL, 'if False:')
assert g.static_checks(broken)['pass'] is False
@pytest.mark.parametrize('spec,ok', [('{"method":"dspark","num_speculative_tokens":5}', True),
('{"method":"dspark","num_speculative_tokens":5,"num_speculative_tokens_per_batch_size":[[1,2,5],[3,4,4]]}', True),
('{"method":"dspark","num_speculative_tokens":5,"rejection_sample_method":"block"}', False),
('{"method":"dspark","num_speculative_tokens":5,"enable_adaptive_verification":true}', False),
('{"method":"mtp","num_speculative_tokens":5}', False)])
def test_standard_spec(spec: str, ok: bool) -> None:
if ok:
g.standard_spec(spec)
else:
with pytest.raises(ValueError):
g.standard_spec(spec)
@needs_image
@pytest.mark.parametrize('change', ['', 'commit', 'kernel', 'count', 'source', 'status'])
def test_receipt_binds_commit_tests_sources_and_admitted_kernel(tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
change: str) -> None:
monkeypatch.setattr(subprocess, 'check_output', lambda argv, **kw: HEAD if argv[1] == 'rev-parse' else '')
out = tmp_path / 'gate.json'
receipt = g.run(out, tmp_path / 'base', runner=fake_runner(dict(g.EXPECTED))) # type: ignore[arg-type]
assert receipt['status'] == 'PASS' and receipt['commit'] == HEAD
admission: dict[str, Json] = {'verifier_sha256': {pp3.UTILS: receipt['pp4_kernel_sha256']}}
data = json.loads(out.read_text())
head = HEAD
if change == 'commit':
head = 'f' * 40
elif change == 'kernel':
admission = {'verifier_sha256': {pp3.UTILS: 'e' * 64}}
elif change == 'count':
data['tests']['test_pp4_sampled_rng']['passed'] = 3
elif change == 'source':
data['source_sha256'][g.TESTS[0]] = '0' * 64
elif change == 'status':
data['status'] = 'FAIL'
out.write_text(json.dumps(data))
if change:
with pytest.raises(ValueError):
g.validate(out, head, admission)
else:
assert g.validate(out, head, admission)['status'] == 'PASS'
@needs_image
@pytest.mark.parametrize('counts,rc,fail', [({'test_verifier_rng': 6, 'test_pp4_sampled_rng': 3}, 0, ''),
(dict(g.EXPECTED), 1, 'test_pp4_sampled_rng')])
def test_failed_or_short_run_is_fail(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, counts: dict[str, int],
rc: int, fail: str) -> None:
monkeypatch.setattr(subprocess, 'check_output', lambda argv, **kw: HEAD if argv[1] == 'rev-parse' else '')
receipt = g.run(tmp_path / 'gate.json', tmp_path / 'base', runner=fake_runner(counts, rc, fail)) # type: ignore[arg-type]
assert receipt['status'] == 'FAIL'
with pytest.raises(ValueError, match='did not PASS'):
g.validate(tmp_path / 'gate.json', HEAD, {'verifier_sha256': {pp3.UTILS: receipt['pp4_kernel_sha256']}})
def test_dirty_tree_or_existing_receipt_refuses(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
(tmp_path / 'gate.json').write_text('{}')
with pytest.raises(ValueError, match='fresh'):
g.run(tmp_path / 'gate.json', tmp_path)
monkeypatch.setattr(subprocess, 'check_output', lambda argv, **kw: HEAD if argv[1] == 'rev-parse' else ' M x.py')
with pytest.raises(ValueError, match='dirty'):
g.run(tmp_path / 'new.json', tmp_path)
|