| """PF2-LOWKV precheck on CPU with a fake remote: no SSH, HTTP, Spark, x86host or GPU.""" |
| from __future__ import annotations |
|
|
| import json |
| import shutil |
| import subprocess |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from offload.launch.windows import pf1_window as w |
| from offload.launch.windows import pf2_precheck as pp |
| from offload.launch.windows import pf2_window as p |
| from offload.launch.windows.tests import test_pf1_precheck as base |
| from offload.launch.windows.tests import test_sp1_precheck as sp1t |
| from offload.launch.windows.tests.test_pf2_rehearsal import extended |
| from offload.launch.windows.tests.test_pf2_window import PLATFORM, parity |
| from offload.pf2_lowkv import arith |
|
|
|
|
| class FakeRemote(sp1t.FakeRemote): |
| """SP1's read-only fake plus PF2's pins, host identity and extended telemetry probe.""" |
|
|
| def __call__(self, host: str, command: str, stdin: str) -> tuple[int, str]: |
| for word in ('bash offload/launch/start_exp', 'stop.sh', 'window.sh', 'restore.sh', ':8001/', '18001', 'rm ', |
| 'rmdir', 'mkdir', 'git checkout', 'git fetch', 'docker stop', 'docker rm', 'tee ', '> ', 'echo never', |
| 'drop_caches', 'nvidia-smi -lgc', 'nvidia-smi -pl'): |
| assert word not in command, f'precheck must stay read-only: {word}' |
| if command == w.HOST_PARITY: |
| self.calls.append((host, command)) |
| kernel = '6.17.0-1026-nvidia' if self.bad == 'kernel' and host == 'sparkhead' else PLATFORM.kernel[host] |
| return 0, parity(kernel, PLATFORM.driver[host]) |
| if command == p.telemetry_command(1): |
| self.calls.append((host, command)) |
| text = extended() |
| if self.bad == 'psi' and host == 'sparkworker-wired': |
| text = text.replace('"psi": {"some"', '"psi": null, "x": {"some"') |
| return 0, text |
| if 'sha256sum offload/' in command: |
| self.calls.append((host, command)) |
| return 0, '\n'.join(f'{"f" * 64 if self.bad == "pins" else w.digest(self.root / f)} {f}' for f in p.PIN_FILES) + '\n' |
| return super().__call__(host, command, stdin) |
|
|
|
|
| @pytest.fixture |
| def repo(tmp_path: Path) -> tuple[Path, str, Path]: |
| root = tmp_path / 'repo' |
| for f in p.PIN_FILES: |
| (root / f).parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy(w.ROOT / f, root / f) |
| git = ['git', '-C', str(root), '-c', 'user.name=t', '-c', 'user.email=t@t'] |
| subprocess.run([*git, 'init', '-q'], check=True) |
| subprocess.run([*git, 'add', '.'], check=True) |
| subprocess.run([*git, 'commit', '-qm', 'x'], check=True) |
| data = tmp_path / 'data' |
| data.mkdir() |
| for name in ('opinions_cl.jsonl', 'filings_recap.jsonl'): |
| (data / name).write_text(json.dumps({'text': 'public record'}) + '\n') |
| return root, subprocess.check_output([*git, 'rev-parse', 'HEAD'], text=True).strip(), data |
|
|
|
|
| def run(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], tmp_path: Path, bad: str = '', |
| low: int = arith.LOW_KV, platform: p.Platform = PLATFORM, inventory: str | None = None, |
| reference: Path | None = None) -> tuple[int, str, FakeRemote]: |
| root, expected, data = repo |
| campaign = tmp_path / 'campaign' |
| campaign.mkdir(exist_ok=True) |
| if inventory is not None: |
| (campaign / 'FW1').mkdir(exist_ok=True) |
| (campaign / 'FW1' / 'inventory-post.txt').write_text(inventory) |
| fake = FakeRemote(root, expected, bad) |
| rc = pp.precheck(expected, None, fake, low=low, platform=platform, reference=reference, campaign=campaign, root=root, |
| data_dir=data, listening=lambda: '') |
| return rc, capsys.readouterr().out, fake |
|
|
|
|
| INVENTORY = ('7.0.0-1004-nvidia\nDGX_OTA_VERSION="7.6.0"\n580.178.02\nMemTotal: 128000000 kB\n' |
| 'MemAvailable: 120000000 kB\nSwapTotal: 16777212 kB\n') |
|
|
|
|
| def test_all_checks_pass_read_only(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: |
| rc, out, fake = run(repo, capsys, tmp_path) |
| assert rc == 0 and 'PRECHECK PASS: run pf2_window.sh' in out and 'FAIL' not in out, out |
| assert {h for h, _ in fake.calls} == {'sparkhead', 'sparkworker-wired', 'x86host'} |
| assert f'PASS sparkhead staged pins: {len(p.PIN_FILES)} files equal, incl. {p.CONFIG}' in out |
| assert 'PASS controller LOW reservation: 3221225472 B = 23,301 blocks, pool 1,410,348 tokens' in out |
| assert 'PASS sparkhead kernel and driver: 7.0.0-1004-nvidia / 580.178.02' in out |
| assert 'PASS sparkworker-wired kernel and driver: 6.17.0-1026-nvidia / 580.159.03' in out |
| assert 'INFO controller FW1 post-update inventory' in out and 'absent' in out |
| assert 'PASS sparkworker-wired memory and telemetry' in out and 'PSI some/full' in out |
|
|
|
|
| def test_fw1_inventory_must_match_the_parameters(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], |
| tmp_path: Path) -> None: |
| rc, out, _ = run(repo, capsys, tmp_path, inventory=INVENTORY) |
| assert rc == 0 and 'PASS controller FW1 post-update inventory: kernel/driver match; MemTotal 122.07 GiB' in out |
| rc, out, _ = run(repo, capsys, tmp_path, inventory=INVENTORY.replace('580.178.02', '580.159.03')) |
| assert rc == 1 and 'FAIL controller FW1 post-update inventory' in out |
|
|
|
|
| @pytest.mark.parametrize('bad, line', [ |
| ('kernel', 'FAIL sparkhead kernel and driver'), |
| ('psi', 'FAIL sparkworker-wired memory and telemetry'), |
| ('pins', 'FAIL sparkhead staged pins'), |
| ('hold', 'FAIL sparkhead production ledger'), |
| ('profile-dir', 'FAIL sparkworker-wired profile directory and disk'), |
| ('worker', 'FAIL x86host TP2 worker'), |
| ]) |
| def test_each_failure_is_reported_and_every_check_still_runs(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], |
| tmp_path: Path, bad: str, line: str) -> None: |
| rc, out, _ = run(repo, capsys, tmp_path, bad) |
| assert rc == 1 and line in out and 'PRECHECK FAIL: 1 check(s)' in out, out |
| assert 'x86host drafter GPUs' in out |
|
|
|
|
| def test_parameters_are_checked(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: |
| rc, out, _ = run(repo, capsys, tmp_path, low=2 * arith.GIB) |
| assert rc == 1 and 'FAIL controller LOW reservation' in out |
| rc, out, _ = run(repo, capsys, tmp_path, platform=p.Platform({}, {})) |
| assert rc == 1 and 'FAIL controller platform parameters' in out and 'FAIL sparkhead kernel and driver' in out |
|
|
|
|
| def test_reference_is_read(repo: tuple[Path, str, Path], capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: |
| ref = tmp_path / 'SP1-W-post' |
| (ref / 'cells').mkdir(parents=True) |
| (ref / 'cells' / 'ctl-150k-b1.json').write_text(json.dumps({'cell': 'ctl-150k-b1', 'streams': [{'ttft_s': 68.5}]})) |
| rc, out, _ = run(repo, capsys, tmp_path, reference=ref) |
| assert rc == 0 and 'PASS controller 327K reference (analysis): 68.50 s from SP1-W-post/cells/ctl-150k-b1.json' in out |
|
|
|
|
| def test_cli_refuses_short_commit() -> None: |
| assert pp.main(['abc1234']) == 2 |
| assert base.GIB > 0 |
|
|
