key-cape/scripts/test_factor_recovery.py

import copy,unittest
from factor_recovery import recover,fingerprint,RecoveryError
class Store:
 realm='coulomb'
 def __init__(self):
  self.state={'id':1,'serial':'T1','user_id':'u1','resolver':'directory','realms':['coulomb'],'active':True,'rollout_state':'enrolled','tokentype':'totp'};self.rows={};self.writes=0;self.fail_audit=False;self.fail_complete=False;self.race=False
 def snapshot(self,user,serial):return copy.deepcopy(self.state) if user=='alice' and serial=='T1' else None
 def receipt(self,ref):return self.rows.get(ref)
 def record(self,request,version,complete):
  if self.fail_audit or self.fail_complete and complete:raise RecoveryError('audit_unavailable')
  self.rows[request['reference']]=dict(request,version=version,complete=complete)
 def disable(self,user,serial,expected):
  if self.race:self.state['id']=2
  if fingerprint(self.state)!=expected:raise RecoveryError('stale_preview')
  self.state['active']=False;self.writes+=1
class RecoveryTests(unittest.TestCase):
 def setUp(self):self.store=Store();self.request=dict(user='alice',serial='T1',realm='coulomb',actor='operator-entity',reference='support-1')
 def approval(self):return dict(self.request,apply=True,identity_verified=True,expected_version=recover(self.store,self.request)['version'])
 def test_preview_never_mutates_and_names_global_scope(self):
  p=recover(self.store,self.request);self.assertFalse(p['changes_applied']);self.assertEqual(p['scope'],'shared_identity_across_applications');self.assertEqual(self.store.writes,0)
 def test_verified_recovery_and_replay_disable_exactly_once(self):
  r=self.approval();self.assertTrue(recover(self.store,r)['changes_applied']);self.assertTrue(recover(self.store,r)['replayed']);self.assertEqual(self.store.writes,1)
 def test_verification_and_ownership_denied_before_mutation(self):
  for r in [dict(self.approval(),identity_verified=False),dict(self.request,user='bob'),dict(self.request,realm='other')]:
   with self.assertRaises(RecoveryError):recover(self.store,r)
  self.assertEqual(self.store.writes,0)
 def test_stale_confirmation_and_concurrent_replacement_rejected(self):
  r=self.approval();self.store.state['id']=2
  with self.assertRaises(RecoveryError):recover(self.store,r)
  self.store.state['id']=1;self.store.race=True
  with self.assertRaises(RecoveryError):recover(self.store,r)
  self.assertEqual(self.store.writes,0)
 def test_audit_failure_prevents_mutation(self):
  r=self.approval();self.store.fail_audit=True
  with self.assertRaises(RecoveryError):recover(self.store,r)
  self.assertEqual(self.store.writes,0)
 def test_completion_audit_retry_does_not_repeat_mutation(self):
  r=self.approval();self.store.fail_complete=True
  with self.assertRaises(RecoveryError):recover(self.store,r)
  self.store.fail_complete=False;self.assertTrue(recover(self.store,r)['replayed']);self.assertEqual(self.store.writes,1)
 def test_conflicting_reference_rejected(self):
  r=self.approval();recover(self.store,r)
  with self.assertRaises(RecoveryError):recover(self.store,dict(r,actor='different-operator'))

 def test_replaced_inactive_factor_cannot_replay_old_recovery(self):
  r=self.approval();recover(self.store,r);self.store.state['id']=2
  with self.assertRaises(RecoveryError):recover(self.store,r)

 def test_replacement_after_disable_cannot_report_success(self):
  original=self.store.disable
  def replace_after_disable(*args):
   original(*args);self.store.state['id']=2
  self.store.disable=replace_after_disable
  with self.assertRaisesRegex(RecoveryError,'readback_failed'):recover(self.store,self.approval())
  self.assertFalse(self.store.rows['support-1']['complete'])