import io,json,sqlite3,tarfile,tempfile,unittest
from pathlib import Path
from make_fixtures import create
from restore_check import run,digest
class RecoveryTests(unittest.TestCase):
 def setUp(self):
  self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name);self.bundle=create(self.root/'set');self.work=self.root/'work';self.work.mkdir();(self.work/'unrelated.txt').write_text('keep')
 def tearDown(self):self.temp.cleanup()
 def update_manifest(self,**updates):
  p=self.bundle/'manifest.json';m=json.loads(p.read_text());m.update(updates);p.write_text(json.dumps(m))
 def assert_clean(self):self.assertEqual(sorted(p.name for p in self.work.iterdir()),['unrelated.txt'])
 def result(self,**kw):return run(self.bundle,work_parent=self.work,**kw)
 def test_good_restore_and_readonly_probes(self):
  before=digest(self.bundle/'database.sqlite');r=self.result();self.assertTrue(r['passed']);self.assertTrue(r['restore_succeeded']);self.assertTrue(r['application_checks_passed']);self.assertEqual(before,digest(self.bundle/'database.sqlite'));self.assert_clean()
 def test_truncated_database_hash(self):
  p=self.bundle/'database.sqlite';p.write_bytes(p.read_bytes()[:100]);r=self.result();self.assertFalse(r['passed']);self.assertIn('Checksum mismatch',r['error']);self.assert_clean()
 def test_truncated_database_even_with_updated_hash(self):
  p=self.bundle/'database.sqlite';p.write_bytes(p.read_bytes()[:100]);self.update_manifest(database_sha256=digest(p));self.assertFalse(self.result()['passed']);self.assert_clean()
 def test_snapshot_mismatch_with_valid_hashes(self):
  db=sqlite3.connect(self.bundle/'database.sqlite');db.execute("UPDATE metadata SET value='other' WHERE key='snapshot_id'");db.commit();db.close();self.update_manifest(database_sha256=digest(self.bundle/'database.sqlite'));r=self.result();self.assertTrue(r['restore_succeeded']);self.assertFalse(r['application_checks_passed']);self.assertIn('snapshot mismatch',r['error']);self.assert_clean()
 def make_archive(self,name,kind=None):
  with tarfile.open(self.bundle/'files.tar','w') as tf:
   i=tarfile.TarInfo(name)
   if kind:i.type=kind;i.linkname='/tmp/outside'
   else:i.size=1
   tf.addfile(i,None if kind else io.BytesIO(b'x'))
  self.update_manifest(archive_sha256=digest(self.bundle/'files.tar'))
 def test_path_traversal(self):
  self.make_archive('../outside');r=self.result();self.assertFalse(r['passed']);self.assertIn('Unsafe archive path',r['error']);self.assertFalse((self.root/'outside').exists());self.assert_clean()
 def test_symlink_rejected(self):
  self.make_archive('visits/001.txt',tarfile.SYMTYPE);r=self.result();self.assertFalse(r['passed']);self.assertIn('regular archive',r['error']);self.assert_clean()
 def test_hardlink_rejected(self):
  self.make_archive('visits/001.txt',tarfile.LNKTYPE);self.assertFalse(self.result()['passed']);self.assert_clean()
 def test_missing_file(self):
  with tarfile.open(self.bundle/'files.tar','w'):pass
  self.update_manifest(archive_sha256=digest(self.bundle/'files.tar'));r=self.result();self.assertFalse(r['passed']);self.assertIn('missing manifest files',r['error']);self.assert_clean()
 def test_database_reference_mismatch(self):
  db=sqlite3.connect(self.bundle/'database.sqlite');db.execute("UPDATE documents SET sha256=? WHERE visit_id=1",('0'*64,));db.commit();db.close();self.update_manifest(database_sha256=digest(self.bundle/'database.sqlite'));r=self.result();self.assertFalse(r['passed']);self.assertIn('metadata mismatch',r['error']);self.assert_clean()
 def test_expected_count_mismatch(self):
  self.update_manifest(expected_visits=999);self.assertFalse(self.result()['passed']);self.assert_clean()
 def test_interruption_cleanup_each_stage(self):
  for stage in ['checksums','files','restore','probes']:
   with self.subTest(stage=stage):
    r=self.result(interrupt_at=stage);self.assertFalse(r['passed']);self.assertIn('Injected interruption',r['error']);self.assertTrue(r['cleanup_succeeded']);self.assert_clean()
 def test_observer_failure_cleans_only_owned_restore(self):
  seen=[]
  def observer(stage,path):
   seen.append(stage)
   if stage=='restore':raise RuntimeError('observer failure')
  before=digest(self.bundle/'database.sqlite');r=self.result(stage_hook=observer)
  self.assertEqual(seen,['checksums','files','restore']);self.assertFalse(r['passed']);self.assertTrue(r['cleanup_succeeded']);self.assertEqual(digest(self.bundle/'database.sqlite'),before);self.assert_clean()
 def test_duplicate_manifest_name(self):
  m=json.loads((self.bundle/'manifest.json').read_text());m['files'].append(m['files'][0]);self.update_manifest(files=m['files']);self.assertFalse(self.result()['passed']);self.assert_clean()
if __name__=='__main__':unittest.main()