bootstrap_retention_test.py (11649B)
1 #!/usr/bin/env python3 2 """Filesystem safety controls for the opt-in checkpoint retention command.""" 3 import copy 4 import fcntl 5 import json 6 import os 7 from pathlib import Path 8 import subprocess 9 import sys 10 import tempfile 11 import unittest 12 from unittest.mock import patch 13 14 import bootstrap_retention as r 15 from bootstrap_retention_probe import portable_probe 16 from bootstrap_retention_lock import acquire, protect_generations 17 18 19 class RetentionTests(unittest.TestCase): 20 def setUp(self): 21 self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) 22 self.root = Path(self.temp.name) / '.bootstrap-run'; self.root.mkdir() 23 self.keep = self.root / 'current'; self.keep.mkdir() 24 (self.keep / 'compiler').write_bytes(b'protected compiler') 25 self.old = self.root / 'old/cas/objects/abc'; self.old.mkdir(parents=True) 26 (self.old / 'tree').write_bytes(b'old tree') 27 (self.old / 'receipt').write_bytes(b'arboricx.build-receipt.v1\nsource\t123\n') 28 self.out = self.root / 'checkpoint'; self.out.mkdir() 29 self.config = dict(protected=['current']) 30 31 def plan(self): 32 plan = r.make_plan(self.root, self.config, self.out) 33 r.save(self.out / 'plan.json', plan) 34 return plan 35 36 def test_explicit_preparation_sequence(self): 37 self.assertEqual(r.support_preparations({'preparation':'first'}), ['first']) 38 self.assertEqual(r.support_preparations({'preparation':'first','additional_preparations':['second']}), ['first','second']) 39 for extra in [['first'], ['second','second'], '', [None], ['']]: 40 with self.assertRaises(ValueError): 41 r.support_preparations({'preparation':'first','additional_preparations':extra}) 42 43 def test_current_cli_oracle_is_v2_only(self): 44 oracle, flags = portable_probe(self.keep, 2) 45 self.assertEqual(oracle, r.REPO / 'test/portable_cli.py') 46 self.assertEqual(flags, ['--compiler-io-version', '2']) 47 48 def test_historical_cli_oracle_requires_frozen_source(self): 49 with self.assertRaisesRegex(ValueError, 'frozen portable CLI oracle'): 50 portable_probe(self.keep, 1) 51 oracle = self.keep / 'test/portable_cli.py' 52 oracle.parent.mkdir(); oracle.write_text('frozen historical oracle') 53 self.assertEqual(portable_probe(self.keep, 1), (oracle, [])) 54 plan = self.plan() 55 oracle.write_text('changed historical oracle') 56 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 57 58 def test_unknown_cli_version_has_no_oracle_fallback(self): 59 with self.assertRaisesRegex(ValueError, 'unsupported retained'): 60 portable_probe(self.keep, 3) 61 62 def test_selected_support_source_is_checked_and_fingerprinted(self): 63 source = self.keep / 'source'; (source/'bootstrap').mkdir(parents=True) 64 (source/'bootstrap/compiler-source-map.v1').write_text('map') 65 (source/'module.tri').write_text('source') 66 rows = [('compiler-library', 'module', 'module', 'module.tri', 'pin')] 67 with patch('fixture_sources.verify', return_value=rows), patch('fixture_sources.current_mismatches', return_value=[]): 68 self.assertEqual(r.check_support_source(self.root, {'source':'current/source'}), source) 69 plan = self.plan() 70 (source/'module.tri').write_text('drift') 71 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 72 (source/'module.tri').unlink() 73 (source/'module.tri').symlink_to(self.keep/'compiler') 74 with self.assertRaises(ValueError): r.check_support_source(self.root, {'source':'current/source'}) 75 with patch('fixture_sources.verify', return_value=[]), patch('fixture_sources.current_mismatches', return_value=['module.tri']): 76 with self.assertRaises(ValueError): r.check_support_source(self.root, {'source':'current/source'}) 77 78 def test_only_objects_and_preserved_evidence(self): 79 (self.root / 'old/status.json').write_text('{"complete": true}') 80 (self.root / 'old/source.tri').write_text('t') 81 plan = self.plan() 82 self.assertEqual([x['objects'] for x in plan['stores']], ['old/cas/objects']) 83 r.prune(self.root, plan, self.out) 84 self.assertFalse(self.old.exists()) 85 self.assertTrue((self.keep / 'compiler').exists()) 86 self.assertTrue((self.root / 'old/status.json').exists()) 87 self.assertTrue((self.root / 'old/source.tri').exists()) 88 self.assertFalse(plan['complete']) # postchecks are mandatory 89 marker = json.loads((self.root / 'old/cas/PRUNED-CAS.json').read_text()) 90 self.assertFalse(marker['resumable']) 91 self.assertEqual(len(list((self.out / 'historical-metadata').iterdir())), 1) 92 93 def test_symlink_target_rejected(self): 94 (self.root / 'bad').mkdir() 95 (self.root / 'bad/objects').symlink_to(self.old.parent, target_is_directory=True) 96 with self.assertRaises(ValueError): self.plan() 97 98 def test_symlink_object_rejected(self): 99 (self.old / 'link').symlink_to(self.keep / 'compiler') 100 with self.assertRaises(ValueError): self.plan() 101 102 def test_path_escape_and_normalization(self): 103 for relative in ['../outside', '/tmp/objects', 'old/../current', 'old//cas', '']: 104 with self.assertRaises(ValueError): r.within(self.root, relative) 105 106 def test_mutated_candidate_blocks_all_deletion(self): 107 plan = self.plan(); (self.old / 'tree').write_bytes(b'changed tree') 108 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 109 self.assertTrue((self.old / 'receipt').exists()) 110 111 def test_mutated_protected_input_blocks_deletion(self): 112 plan = self.plan(); (self.keep / 'compiler').write_bytes(b'changed') 113 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 114 self.assertTrue(self.old.exists()) 115 116 def test_corrupt_metadata_blocks_deletion(self): 117 plan = self.plan() 118 next((self.out / 'historical-metadata').iterdir()).write_bytes(b'corrupt') 119 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 120 self.assertTrue(self.old.exists()) 121 122 def test_protected_target_injected_in_plan_rejected(self): 123 (self.keep / 'objects').mkdir() 124 plan = self.plan() 125 plan['stores'].append(dict(objects='current/objects', files={}, metadata=[], state='pending')) 126 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 127 self.assertTrue(self.old.exists()) 128 129 def test_hardlink_reclaim_and_retained_identity(self): 130 os.link(self.keep / 'compiler', self.old / 'shared') 131 external = Path(self.temp.name) / 'external'; os.link(self.old / 'tree', external) 132 plan = self.plan() 133 expected = (self.old / 'receipt').stat().st_blocks * 512 134 self.assertEqual(plan['reclaim_bytes'], expected) 135 r.prune(self.root, plan, self.out) 136 self.assertEqual((self.keep / 'compiler').read_bytes(), b'protected compiler') 137 self.assertEqual(external.read_bytes(), b'old tree') 138 139 def test_interrupted_directory_resumption(self): 140 plan = self.plan() 141 plan['stores'][0]['state'] = 'deleting' 142 (self.old / 'tree').unlink() # interrupted shutil.rmtree 143 r.prune(self.root, plan, self.out) 144 self.assertEqual(plan['stores'][0]['state'], 'removed') 145 r.prune(self.root, plan, self.out) # idempotent before postchecks 146 147 def test_missing_pending_directory_not_resumption(self): 148 plan = self.plan() 149 import shutil 150 shutil.rmtree(self.old.parent) 151 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 152 153 def test_removed_directory_reappearance_rejected(self): 154 plan = self.plan(); r.prune(self.root, plan, self.out) 155 self.old.mkdir(parents=True) 156 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 157 158 def test_tool_changed_plan_rejected(self): 159 plan = self.plan(); plan['tool_sha256'] = '0' * 64 160 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 161 162 def test_generation_cleanup_lock_exclusion(self): 163 handles = protect_generations([self.keep, self.old]) 164 self.addCleanup(lambda: [h.close() for h in handles]) 165 with self.assertRaises(ValueError): acquire(self.root, exclusive=True) 166 for h in handles: h.close() 167 lock = acquire(self.root, exclusive=True); self.addCleanup(lock.close) 168 with self.assertRaises(ValueError): protect_generations([self.keep]) 169 170 def test_generation_runner_rejects_pruned_input_and_output(self): 171 plan = self.plan(); r.prune(self.root, plan, self.out) 172 script = r.REPO / 'test/bootstrap_generations.py' 173 for args in [[str(self.root / 'old'), '--bootstrap-kit', str(self.root / 'kit')], 174 [str(self.root / 'old'), str(self.root / 'fresh')]]: 175 reply = subprocess.run([sys.executable, str(script), *args], 176 capture_output=True, text=True, timeout=30) 177 self.assertNotEqual(reply.returncode, 0) 178 self.assertIn('pruned generations are historical evidence', reply.stderr) 179 self.assertFalse((self.root / 'old/status.json').exists()) 180 self.assertFalse((self.root / 'fresh').exists()) 181 182 def test_pending_checkpoint_cannot_pass_completion_gate(self): 183 plan = self.plan() 184 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 185 r.prune(self.root, plan, self.out) 186 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 187 188 def test_completed_checkpoint_requires_both_check_reports(self): 189 plan = self.plan(); r.prune(self.root, plan, self.out) 190 plan.update(complete=True, phase='complete', prechecks='missing-before', postchecks='missing-after') 191 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 192 193 def test_plan_cannot_create_evidence_inside_protected_inputs(self): 194 evidence = self.keep / 'must-not-create' 195 args = ['bootstrap_retention.py', 'plan', '--root', str(self.root), 196 '--directory', str(evidence), '--config', str(self.root / 'config.json')] 197 with patch('sys.argv', args), patch.object(r, 'load_config', return_value=self.config): 198 with self.assertRaises(ValueError): r.main() 199 self.assertFalse(evidence.exists()) 200 201 def test_archived_harness_allows_non_destructive_new_version_check(self): 202 plan = self.plan() 203 original_sha = self.original_sha 204 with patch.object(r, 'sha', side_effect=lambda p: 'changed' if Path(p) == Path(r.__file__) else original_sha(p)): 205 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 206 r.validate_plan(self.root, plan, self.out, same_tools=False) 207 (self.out / 'harness/bootstrap_retention.py').write_bytes(b'corrupt') 208 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out, same_tools=False) 209 210 original_sha = staticmethod(r.sha) 211 212 def test_symlink_retained_directory_is_fingerprinted(self): 213 (self.keep / 'runtime').symlink_to('/nix/store/one', target_is_directory=True) 214 plan = self.plan() 215 (self.keep / 'runtime').unlink() 216 (self.keep / 'runtime').symlink_to('/nix/store/two', target_is_directory=True) 217 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 218 219 220 if __name__ == '__main__': 221 unittest.main()