| """A100 opt-in protocol/budget checks; no GPU, model, network, or engine execution.""" |
| import copy |
| import hashlib |
| import json |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from types import SimpleNamespace |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) |
| import benchmark_batch |
| import evaluate_pro as pro |
|
|
|
|
| class ProEvaluationTests(unittest.TestCase): |
| def setUp(self): |
| self.temp = tempfile.TemporaryDirectory() |
| self.root = Path(self.temp.name) |
| self.run = self.root / 'run' |
| self.data = self.root / 'data' |
| self.adapter = self.run / 'adapter' |
| self.output = self.run / pro.NAME |
| self.t4 = self.run / 'evaluation' |
| for path in (self.data, self.adapter, self.t4): |
| path.mkdir(parents=True) |
| rows = [{'instruction': f'original training prompt {index}', 'output': '{}'} for index in range(4)] |
| (self.data / 'train.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows)) |
| for name in ('dev.jsonl', 'eval_prompts.jsonl', 'validation_manifest.jsonl'): |
| (self.data / name).write_text('original fixture ' + name) |
| signature = {'base_model': pro.BASE_MODEL, 'base_revision': 'pinned-revision', |
| 'system_prompt_sha256': hashlib.sha256(pro.SYSTEM_PROMPT.encode()).hexdigest()} |
| for key, filename in [('train_sha256', 'train.jsonl'), ('dev_sha256', 'dev.jsonl'), |
| ('blind_sha256', 'eval_prompts.jsonl'), ('validation_manifest_sha256', 'validation_manifest.jsonl')]: |
| signature[key] = pro.file_hash(self.data / filename) |
| for path in (self.run / 'run_manifest.json', self.adapter / 'run_manifest.json'): |
| path.write_text(json.dumps({'signature': signature})) |
| (self.adapter / 'TRAINING_COMPLETE.json').write_text(json.dumps({'status': 'complete', 'target_steps': 14, 'global_step': 14})) |
| (self.adapter / 'adapter_model.safetensors').write_bytes(b'fixture-adapter-no-real-tensors') |
| (self.adapter / 'tokenizer.json').write_text('{}') |
| self.godot = self.root / 'godot-not-an-executable' |
| self.godot.write_bytes(b'fixed-engine-fixture') |
| self.runtime = {'cuda_available': True, 'device_count': 1, 'gpu': 'NVIDIA A100-SXM4-40GB', |
| 'torch_version': '2.11.0+cu130', 'cuda_version': '13.0', |
| 'packages': {'unsloth': 'fixture', 'transformers': 'fixture', 'peft': 'fixture', 'bitsandbytes': 'fixture'}} |
| self.prior = {'backend': 'unsloth', 'adapter': pro.directory_fingerprint(self.adapter), |
| 'base_model': signature['base_model'], 'base_revision': signature['base_revision'], |
| 'system_prompt_sha256': signature['system_prompt_sha256'], |
| 'gpu': 'Tesla T4', 'base_artifacts': {'sha256': 'fixture-pinned-base', 'files': {}}, |
| **{key: self.runtime[key] for key in ('torch_version', 'cuda_version', 'packages')}} |
| self.prior_manifest = {'expected_count': 500, 'max_new_tokens': 3072, 'run_seconds': 60, |
| 'sample_indices': [0, 200, 400], 'smoke_test': False, 'batch_size': 8, |
| 'workers': 4, 'decoding': {'temperature': 0, 'seed': 42}, 'latency_version': 2, |
| 'prompts_sha256': signature['blind_sha256'], 'runtime_sha256': pro.file_hash(self.godot), |
| 'generation': self.prior} |
| (self.t4 / 'manifest.json').write_text(json.dumps(self.prior_manifest)) |
| (self.t4 / 'result-fixture.json').write_text('immutable prior result') |
| self.prompts = [row['instruction'] for row in rows] |
|
|
| def tearDown(self): |
| self.temp.cleanup() |
|
|
| def protocol(self): |
| return pro.validate_sources(self.run, self.data, self.output, self.godot, self.runtime) |
|
|
| def capacity(self): |
| protocol = self.protocol() |
| estimate = {'within_limit': True, 'assessed_batch_size': 8, 'target_context_tokens': 4096, |
| 'cache_growth_allowance_tokens': 512, 'reserve_bytes': 2_000_000_000, |
| 'estimated_peak_bytes': 10_000_000_000, 'effective_limit_bytes': 12_500_000_000} |
| full = {'status': 'success', 'batch_size': 8, 'max_new_tokens': 3072, 'within_memory_limit': True, |
| 'full_context_memory_estimate': estimate, 'peak_allocated_bytes': 8_000_000_000, |
| 'peak_reserved_bytes': 9_000_000_000, 'per_prompt_tokens': [1536] * 8, |
| 'end_to_end_call_seconds': 20.0} |
| return {'status': 'complete', 'evaluation_profile': 'pro-a100-opt-in', 'blind_prompts_read': False, |
| 'batch_sizes': [8], 'recommended_batch_size': 8, 'full_max_new_tokens': 3072, |
| 'memory_estimate_policy': 'kv_full_context_v1', 'training_jsonl_sha256': protocol['train_sha256'], |
| 'training_prompts_sha256': hashlib.sha256(json.dumps(self.prompts, ensure_ascii=False).encode()).hexdigest(), |
| 'gpu': self.runtime['gpu'], 'gpu_memory_bytes': 40_000_000_000, |
| 'generation': {**self.prior, 'gpu': self.runtime['gpu']}, |
| 'full_check': full, 'measurements': [{**full, 'max_new_tokens': 256}]} |
|
|
| def test_valid_predeclaration_is_immutable_and_t4_evidence_untouched(self): |
| original = {path.name: path.read_bytes() for path in self.t4.iterdir()} |
| protocol = self.protocol() |
| pro.declare_protocol(self.output, protocol) |
| first = (self.output / 'pro-protocol.json').read_bytes() |
| pro.declare_protocol(self.output, protocol) |
| self.assertEqual(first, (self.output / 'pro-protocol.json').read_bytes()) |
| self.assertEqual(original, {path.name: path.read_bytes() for path in self.t4.iterdir()}) |
| self.assertFalse((self.output / 'items').exists()) |
| with self.assertRaisesRegex(ValueError, 'protocol changed'): |
| pro.declare_protocol(self.output, {**protocol, 'batch_size': 16}) |
| self.assertEqual(first, (self.output / 'pro-protocol.json').read_bytes()) |
|
|
| def test_t4_and_other_output_paths_and_symlink_are_rejected(self): |
| for output in (self.t4, self.run / 'evaluation-a100-v2', self.root / pro.NAME): |
| with self.subTest(output=output), self.assertRaisesRegex(ValueError, 'separate'): |
| pro.validate_sources(self.run, self.data, output, self.godot, self.runtime) |
| self.output.symlink_to(self.t4, target_is_directory=True) |
| with self.assertRaisesRegex(ValueError, 'separate'): |
| self.protocol() |
|
|
| def test_no_unowned_results_are_adopted(self): |
| self.output.mkdir() |
| (self.output / 'manifest.json').write_text('previously started') |
| with self.assertRaisesRegex(ValueError, 'without the A100 predeclaration'): |
| pro.declare_protocol(self.output, self.protocol()) |
|
|
| def test_actual_a100_and_unchanged_runtime_are_required(self): |
| for changes in ({'gpu': 'Tesla T4'}, {'cuda_available': False}, {'device_count': 2}, |
| {'torch_version': 'changed'}, {'packages': {}}): |
| with self.subTest(changes=changes), self.assertRaises(ValueError): |
| pro.validate_sources(self.run, self.data, self.output, self.godot, {**self.runtime, **changes}) |
|
|
| def test_changed_tokenizer_data_or_incomplete_training_is_rejected(self): |
| for path, content in [(self.adapter / 'tokenizer.json', '{"changed":true}'), |
| (self.data / 'train.jsonl', 'changed'), |
| (self.adapter / 'TRAINING_COMPLETE.json', '{"status":"partial"}')]: |
| old = path.read_bytes() |
| path.write_text(content) |
| with self.subTest(path=path), self.assertRaises(ValueError): |
| self.protocol() |
| path.write_bytes(old) |
|
|
| def test_capacity_accepts_only_matching_completed_training_probe(self): |
| report = self.capacity() |
| self.assertTrue(pro.capacity_valid(report, self.protocol(), self.prior, self.prompts)) |
| for change in ({'status': 'full_length_check'}, {'recommended_batch_size': 16}, {'blind_prompts_read': True}, |
| {'evaluation_profile': 't4'}, {'training_jsonl_sha256': 'other'}, {'gpu': 'Tesla T4'}): |
| with self.subTest(change=change): |
| self.assertFalse(pro.capacity_valid({**report, **change}, self.protocol(), self.prior, self.prompts)) |
| changed = copy.deepcopy(report) |
| changed['full_check']['full_context_memory_estimate']['within_limit'] = False |
| self.assertFalse(pro.capacity_valid(changed, self.protocol(), self.prior, self.prompts)) |
| changed = copy.deepcopy(report) |
| changed['generation']['base_revision'] = 'different-base' |
| self.assertFalse(pro.capacity_valid(changed, self.protocol(), self.prior, self.prompts)) |
|
|
| def test_budget_normalizes_early_eos_and_never_credits_overlap(self): |
| budget = pro.budget_projection(self.capacity()['full_check'], 10, 100, 360) |
| self.assertEqual(budget['normalized_3072_batch_seconds'], 40) |
| self.assertEqual(budget['generation_seconds_with_margin'], 63 * 40 * 1.5) |
| self.assertEqual(budget['validation_seconds_without_overlap_credit'], 125 * 90) |
| self.assertEqual(budget['projected_remaining_seconds'], 63 * 40 * 1.5 + 125 * 90 + 1800) |
| self.assertEqual(budget['estimated_units_used_since_start'], 1) |
| self.assertEqual(budget['usable_remaining_units'], 89) |
| self.assertTrue(budget['fits_observed_balance']) |
| self.assertFalse(pro.budget_projection(self.capacity()['full_check'], 10, 40, 360)['fits_observed_balance']) |
| self.assertFalse(pro.budget_projection(self.capacity()['full_check'], 10, 100, 40000)['fits_observed_balance']) |
|
|
| def test_invalid_budget_observations_or_measurements_fail_closed(self): |
| for rate, balance in [(0, 100), (10, 0), (float('nan'), 100), (10, float('inf')), (-1, 100)]: |
| with self.subTest(rate=rate, balance=balance), self.assertRaises(ValueError): |
| pro.budget_projection(self.capacity()['full_check'], rate, balance, 0) |
| for changed in ({'per_prompt_tokens': [0] * 8}, {'end_to_end_call_seconds': float('nan')}): |
| with self.subTest(changed=changed), self.assertRaises(ValueError): |
| pro.budget_projection({**self.capacity()['full_check'], **changed}, 10, 100, 0) |
|
|
| def test_opt_in_capacity_preserves_t4_hardware_default(self): |
| def fake(name): |
| return SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True, device_count=lambda: 1, |
| get_device_name=lambda: name)) |
| benchmark_batch.require_gpu(fake('Tesla T4')) |
| with self.assertRaises(RuntimeError): |
| benchmark_batch.require_gpu(fake('NVIDIA A100-SXM4-40GB')) |
| benchmark_batch.require_gpu(fake('NVIDIA A100-SXM4-40GB'), pro_a100=True) |
| with self.assertRaises(RuntimeError): |
| benchmark_batch.require_gpu(fake('Tesla T4'), pro_a100=True) |
|
|
| def test_capacity_reads_only_hash_matched_training_split(self): |
| rows = benchmark_batch.capacity_prompts(self.adapter, self.data / 'train.jsonl') |
| self.assertEqual([row['instruction'] for row in rows], self.prompts) |
| with self.assertRaisesRegex(ValueError, 'training split'): |
| benchmark_batch.capacity_prompts(self.adapter, self.data / 'eval_prompts.jsonl') |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
