tegwick
c96b0c6d88
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
80 lines
5 KiB
Python
80 lines
5 KiB
Python
"""Synthetic recovery guards; no actual private journal/DB/native command."""
|
|
import copy
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from authority_journal import Journal, authority, original_anchor
|
|
from test_execution import accepted, NOW, REVISION
|
|
|
|
ROOT=Path(__file__).parent
|
|
PLAN=json.loads((ROOT/'execution-plan.json').read_text())
|
|
CASE=ROOT.parent
|
|
BACKUP=json.loads((CASE/'original-approved-drill-backup.json').read_text())
|
|
|
|
class RescueGuards(unittest.TestCase):
|
|
def original(self,parent):
|
|
original={'identity':PLAN['recovery']['original_identity'],'phases':{}}
|
|
for name,metadata in PLAN['recovery']['original_anchor']['phase_metadata'].items():
|
|
original['phases'][name]={**metadata,'result':BACKUP if name=='drill-backup' else {'all_checks_pass':False}}
|
|
p=parent/'execution-journal.json';p.write_text(json.dumps(original));os.chmod(p,0o600)
|
|
anchor=copy.deepcopy(PLAN['recovery']['original_anchor']);anchor['original_journal_sha256']=hashlib.sha256(p.read_bytes()).hexdigest()
|
|
return p,original,anchor
|
|
|
|
def test_exact_original_failed_journal_is_linked_without_mutation(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent=Path(directory);os.chmod(parent,0o700);p,original,expected=self.original(parent);before=p.read_bytes()
|
|
anchor=original_anchor(parent,expected,BACKUP,original['identity'])
|
|
j=Journal(parent,{'fixture':True,'recovery_anchor':anchor});j.start('drill-copy-rescue-r1',{})
|
|
self.assertEqual(p.read_bytes(),before);self.assertEqual(j.path.name,'rescue-execution-journal-r1.json')
|
|
j.finish('drill-copy-rescue-r1',{'all_checks_pass':True});j.start('offline-verify',{})
|
|
self.assertEqual(p.read_bytes(),before);j.close()
|
|
|
|
def test_original_missing_uncertain_recreated_attempt_or_rollout_refuse(self):
|
|
for defect in ('missing','complete','attempt','rollout','hash'):
|
|
with self.subTest(defect=defect),tempfile.TemporaryDirectory() as directory:
|
|
parent=Path(directory);os.chmod(parent,0o700);p,o,e=self.original(parent)
|
|
if defect=='missing':o['phases'].pop('drill-copy')
|
|
elif defect=='complete':o['phases']['drill-copy']['state']='COMPLETE'
|
|
elif defect=='attempt':o['phases']['drill-copy']['attempt_id']='different'
|
|
elif defect=='rollout':o['phases']['release']={'state':'RUNNING'}
|
|
if defect!='hash':
|
|
p.write_text(json.dumps(o));e['original_journal_sha256']=hashlib.sha256(p.read_bytes()).hexdigest()
|
|
else:e['original_journal_sha256']='0'*64
|
|
with self.assertRaises((ValueError,KeyError)):original_anchor(parent,e,BACKUP,PLAN['recovery']['original_identity'])
|
|
|
|
def test_rescue_replay_refused_and_old_phase_never_admitted(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
os.chmod(directory,0o700);j=Journal(directory,{'fixture':True})
|
|
with self.assertRaises(ValueError):j.start('drill-copy',{})
|
|
with self.assertRaises(ValueError):j.start('drill-backup',{})
|
|
with self.assertRaises(ValueError):j.start('offline-verify',{})
|
|
j.start('drill-copy-rescue-r1',{});j.finish('drill-copy-rescue-r1',{'all_checks_pass':False})
|
|
with self.assertRaises(ValueError):j.start('drill-copy-rescue-r1',{})
|
|
with self.assertRaises(ValueError):j.start('offline-verify',{})
|
|
j.close()
|
|
|
|
def test_new_authority_requires_version2_old_approval_refuses(self):
|
|
receipt,outcome=accepted();receipt['memo_version']=2;outcome['dispositions'][0]['disposition']['memo_version']=1
|
|
authority(receipt,outcome,'a'*64,REVISION,'drill-copy-rescue-r1',NOW)
|
|
receipt['memo_version']=1
|
|
with self.assertRaises(ValueError):authority(receipt,outcome,'a'*64,REVISION,'drill-copy-rescue-r1',NOW)
|
|
with self.assertRaises(ValueError):authority(receipt,outcome,'a'*64,REVISION,'drill-copy',NOW)
|
|
receipt['memo_version']=2;receipt['outcome_query']='/lifecycle/memos/infd-20261011-purpose-overlay-r1/outcomes?version=1'
|
|
with self.assertRaises(ValueError):authority(receipt,outcome,'a'*64,REVISION,'drill-copy-rescue-r1',NOW)
|
|
|
|
def test_explicit_offline_input_mapping_and_absence_guards(self):
|
|
source=(ROOT/'actuator.py').read_text()
|
|
self.assertIn("previous = journal.data['phases']['drill-copy-rescue-r1']['result']",source)
|
|
self.assertIn("not os.path.lexists(CUSTODY / 'drill.sqlite')",source)
|
|
self.assertIn("VOLUME not in listed.stdout.splitlines()",source)
|
|
self.assertNotIn("os.mkdir(CUSTODY",source)
|
|
import ast
|
|
program=ast.literal_eval(next(n.value for n in ast.parse((ROOT/'native_program.py').read_text()).body if isinstance(n,ast.Assign)))
|
|
compile(program,'native-rescue-fixture','exec')
|
|
self.assertIn("phase=='drill-copy-rescue-r1':require(pod['metadata']['uid']=='e8561ae0-4b09-4455-924d-64c27b625fc7'",program)
|
|
|
|
if __name__=='__main__':unittest.main()
|