| """Meaningful parser regressions on recorded aggregate logs plus malformed inputs.""" |
| from __future__ import annotations |
|
|
| import contextlib |
| import io |
| import importlib.util |
| import json |
| from pathlib import Path |
| import subprocess |
| import sys |
| import tempfile |
| import unittest |
| from typing import Any |
|
|
| HERE = Path(__file__).resolve().parents[1] |
| ROOT = HERE.parents[2] |
| spec = importlib.util.spec_from_file_location('window_checks', HERE / 'checks.py') |
| assert spec and spec.loader |
| checks = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(checks) |
|
|
|
|
| class Checks(unittest.TestCase): |
| def test_recorded_bench(self) -> None: |
| rows = checks.bench(ROOT / 'offload/results/E/E-tp2-short.jsonl') |
| self.assertGreater(len(rows), 0) |
| self.assertTrue(all(r['accepted_per_round'] > 0 and r['ttft_s'] > 0 for r in rows)) |
|
|
| def test_recorded_kv(self) -> None: |
| path = ROOT / 'offload/results/C/startup-C-shadow-rank0.txt' |
| text = path.read_text() |
| values = checks.re.findall(r'GPU KV cache size: ([\d,]+)', text) |
| self.assertTrue(values) |
| expected = int(values[0].replace(',', '')) |
| self.assertEqual(checks.kv(text, expected), expected) |
| with self.assertRaises(ValueError): checks.kv(text, expected + 1) |
| with self.assertRaises(ValueError): checks.kv('startup succeeded', expected) |
|
|
| def test_recorded_counters_and_failure(self) -> None: |
| root = ROOT / 'offload/results/W4-E0' |
| before = checks.load(root / 'rank0-stats-before.json') |
| after = checks.load(root / 'rank0-stats-after.json') |
| with self.assertRaises(ValueError): checks.counter_delta(before, after) |
| before = checks.load(root / 'rank0-stats-after.json') |
| checks.counter_delta(before, after) |
| after['counters']['fallback_internal_error'] = 1 |
| with self.assertRaises(ValueError): checks.counter_delta(before, after) |
|
|
| def test_strict_memory_and_prefix(self) -> None: |
| checks.memory('MemAvailable: 5242880 kB\nSwapFree: 100 kB\n') |
| for bad in ('MemAvailable: 5242879 kB\nSwapFree: 100 kB', 'MemAvailable: 9999999 kB', ''): |
| with self.assertRaises(ValueError): checks.memory(bad) |
| for rank in (0, 1): |
| line = f'EXL3 RDMA all-reduce: rank {rank} of 2 on rocep1s0f1,roceP2p1s0f1, gid 3, proxy on cpu 5' |
| checks.dual(line, rank) |
| with self.assertRaises(ValueError): checks.dual('prefix ' + line, rank) |
| with self.assertRaises(ValueError): checks.dual(line.replace('roceP2p1s0f1', 'roceP2p1s0f10'), rank) |
| with self.assertRaises(ValueError): checks.dual(line + '\nRDMA all-reduce off x NCCL stays', rank) |
|
|
| def test_replay_stop_strings(self) -> None: |
| for text in ('invalid replay tap provenance', 'replay skip would remove taps'): |
| with self.assertRaises(ValueError): checks.clean_logs(text) |
|
|
| def test_receipt_fail_closed(self) -> None: |
| with tempfile.TemporaryDirectory() as td: |
| root = Path(td) |
| (root / '.run-start').touch() |
| source = root / 'source.txt'; source.write_text('aggregate diagnostic evidence') |
| data: dict[str, Any] = {'schema': 'wr1-gate-v1', 'gate': 'idle', 'arm': 'test', 'run_dir': str(root), |
| 'checks': dict.fromkeys(checks.REQUIRED['idle'], True), |
| 'evidence': [{'path': str(source), 'sha256': checks.hashlib.sha256(source.read_bytes()).hexdigest()}]} |
| target = root / 'receipt.json'; target.write_text(json.dumps(data)) |
| checks.receipt(target, 'idle', 'test', root) |
| source.write_text('changed') |
| with self.assertRaises(ValueError): checks.receipt(target, 'idle', 'test', root) |
| data['checks']['experiment_idle'] = 'true' |
| target.write_text(json.dumps(data)) |
| with self.assertRaises(ValueError): checks.receipt(target, 'idle', 'test', root) |
|
|
| def test_native_dry_run_no_transport_or_writes(self) -> None: |
| with tempfile.TemporaryDirectory() as td: |
| env = dict(checks.os.environ, WINDOW_OUT=td + '/must-not-exist') |
| fake = Path(td) / 'ssh' |
| fake.write_text('#!/bin/sh\nexit 99\n'); fake.chmod(0o755) |
| env['PATH'] = td + ':' + env['PATH'] |
| for script, args in [('native_pair.sh', ['r000008_b0']), ('native.sh', ['control.env', '100', '100', '/missing-timings.json']), ('up1.sh', ['2502600']), ('rp1.sh', ['2502600', '2502600', 'a' * 64, '0-4,10-13']), ('depth.sh', ['2502600', 'control.env', 'sp2-fixed.env', 'sp2-adaptive.env'])]: |
| proc = subprocess.run(['bash', str(HERE / script), '--dry-run', *args], env=env, capture_output=True, text=True) |
| self.assertEqual(proc.returncode, 0, proc.stderr) |
| self.assertIn('STEP summary', proc.stdout) |
| self.assertFalse(Path(env['WINDOW_OUT']).exists()) |
|
|
| def test_order_and_required_dry_run_steps(self) -> None: |
| env = dict(checks.os.environ, WINDOW_OUT='/tmp/wr1-uncreated-test-plan') |
| proc = subprocess.run(['bash', str(HERE / 'up1.sh'), '--dry-run', '2502600'], env=env, capture_output=True, text=True, check=True) |
| labels = ['p0-r1-K1-start', 'p0-r1-C-start', 'p0-r1-K2-start', 'up1-split-r1-X-start', |
| 'up1-split-pin-r1-X-start', 'up1-cpu-pin-r1-X-start', 'up1-replay-r1-X-start', |
| 'up1-k5-r1-X-start', 'up1-dual-ring-r1-X-start', 'up1-vor-copy-r1-X-start', |
| 'up1-vor-swa-r1-X-start', 'up1-vor-both-r1-X-start'] |
| positions = [proc.stdout.index('STEP ' + label + ':') for label in labels] |
| self.assertEqual(positions, sorted(positions)) |
| for rank in (0, 1): |
| self.assertIn(f'up1-split-r1-X-runtime-hash-rank{rank}', proc.stdout) |
| self.assertIn(f'up1-dual-ring-r1-X-dual-rank{rank}', proc.stdout) |
| self.assertIn('UP1_SPLIT=63', proc.stdout) |
| self.assertIn('up1-replay-r1-X-stats-before', proc.stdout) |
| self.assertIn('up1-replay-r1-X-stats-after', proc.stdout) |
| self.assertNotIn('up1-all-', proc.stdout) |
|
|
| def test_invalid_arguments_stop_before_transport(self) -> None: |
| for script, args in [('native.sh', ['control.env', 'bad', '100', '/missing']), |
| ('rp1.sh', ['100', '100', 'bad-hash', '0-4']), |
| ('depth.sh', ['100', 'same.env', 'same.env'])]: |
| proc = subprocess.run(['bash', str(HERE / script), '--dry-run', *args], capture_output=True, text=True) |
| self.assertNotEqual(proc.returncode, 0) |
| self.assertNotIn('STEP hold:', proc.stdout) |
|
|
| def test_timing_requires_evidence_and_cost_semantics(self) -> None: |
| with tempfile.TemporaryDirectory() as td: |
| source = Path(td) / 'costs.json'; source.write_text('aggregate timing evidence') |
| data: dict[str, Any] = {'classification': 'MEASURED', 'arm': 'native-hybrid-kx2', 'draft_ms': 6.5, |
| 'round_ms': [10, 11, 12, 13, 14, 15], 'overlap_audited': True, 'verify_only': True, |
| 'sources': [{'path': str(source), 'sha256': checks.hashlib.sha256(source.read_bytes()).hexdigest()}]} |
| path = Path(td) / 'timing.json'; path.write_text(json.dumps(data)) |
| with contextlib.redirect_stdout(io.StringIO()): checks.timing(path) |
| data['round_ms'] = [10, 11, 12, 10, 14, 15]; path.write_text(json.dumps(data)) |
| with self.assertRaises(ValueError): checks.timing(path) |
| data['round_ms'] = [10, 11, 12, 13, 14, 15] |
| data['sources'] = []; path.write_text(json.dumps(data)) |
| with self.assertRaises(ValueError): checks.timing(path) |
|
|
| def test_step_propagates_failure(self) -> None: |
| with tempfile.TemporaryDirectory() as td: |
| command = f'source "{HERE}/common.sh"; W_DRY=0; W_OUT="{td}"; w_step failure bash -c "exit 23"; touch "{td}/unreachable"' |
| proc = subprocess.run(['bash', '-c', command], capture_output=True) |
| self.assertNotEqual(proc.returncode, 0) |
| self.assertFalse((Path(td) / 'unreachable').exists()) |
| self.assertIn('rc=23', (Path(td) / 'steps.log').read_text()) |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
