"""Native recovery tests use only synthetic temporary SQLite databases."""
import hashlib,json,sqlite3,subprocess,sys,tempfile,threading,unittest
from pathlib import Path
from test_backup_recovery import execute_non_destructive_backup_test,REQUIRED_TABLES

ROOT=Path(__file__).resolve().parents[1]

def schema(connection):
    for table in REQUIRED_TABLES:connection.execute('CREATE TABLE '+table+'(id INTEGER PRIMARY KEY,value TEXT NOT NULL)')
    connection.commit()

class BackupDrillTests(unittest.TestCase):
    def test_committed_wal_data_is_recovered_without_source_checkpoint(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'synthetic-wal.db'
            with sqlite3.connect(path) as writer:
                writer.execute('PRAGMA journal_mode=WAL');writer.execute('PRAGMA wal_autocheckpoint=0');schema(writer)
                writer.execute('INSERT INTO enterprise_portfolio VALUES(1,?)',('SYNTHETIC_WAL_TEST',));writer.commit()
                wal=Path(str(path)+'-wal');self.assertTrue(wal.exists());self.assertGreater(wal.stat().st_size,0)
                before=path.read_bytes();before_wal=wal.read_bytes()
                r=execute_non_destructive_backup_test(path)
                self.assertEqual(r['status'],'NATIVE_SNAPSHOT_REOPEN_AND_SCHEMA_COUNTS_VERIFIED',r)
                self.assertEqual(r['required_table_row_counts']['enterprise_portfolio'],1)
                self.assertEqual(path.read_bytes(),before);self.assertEqual(wal.read_bytes(),before_wal)
                self.assertTrue(r['snapshot_reopened_read_only']);self.assertTrue(r['temporary_workspace_removed'])
                self.assertFalse(r['source_checkpoint_requested'])
    def test_missing_source_not_created(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'missing.db';r=execute_non_destructive_backup_test(path)
            self.assertEqual(r['status'],'BACKUP_RECOVERY_BLOCKED');self.assertFalse(path.exists())
    def test_missing_required_schema_blocks_without_creating_tables(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'different-ledger.db'
            with sqlite3.connect(path) as c:c.execute('CREATE TABLE actual_trade_ledger(id INTEGER)')
            before=path.read_bytes();r=execute_non_destructive_backup_test(path)
            self.assertEqual(r['status'],'BACKUP_RECOVERY_BLOCKED');self.assertIn('Required schema missing',r['reason'])
            self.assertEqual(path.read_bytes(),before);self.assertTrue(r['temporary_workspace_removed'])
    def test_foreign_key_violations_block_even_if_integrity_is_ok(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'invalid-reference.db'
            with sqlite3.connect(path) as c:
                schema(c);c.execute('CREATE TABLE p(id INTEGER PRIMARY KEY)')
                c.execute('CREATE TABLE child(id INTEGER PRIMARY KEY,parent INTEGER REFERENCES p(id))')
                c.execute('INSERT INTO child VALUES(1,999)');c.commit()
                self.assertEqual(c.execute('PRAGMA integrity_check').fetchone()[0],'ok')
            r=execute_non_destructive_backup_test(path)
            self.assertEqual(r['status'],'BACKUP_RECOVERY_BLOCKED');self.assertIn('foreign-key',r['reason'])
    def test_corruption_blocks_and_cleans_temporary_workspace(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'corrupt.db';path.write_bytes(b'not-a-database')
            r=execute_non_destructive_backup_test(path)
            self.assertEqual(r['status'],'BACKUP_RECOVERY_BLOCKED');self.assertTrue(r['temporary_workspace_removed'])
            self.assertEqual(path.read_bytes(),b'not-a-database')
    def test_invalid_deadline_rejected(self):
        for deadline in (0,True,float('nan'),float('inf'),61):
            r=execute_non_destructive_backup_test('/does/not/exist',timeout_seconds=deadline)
            self.assertEqual(r['status'],'BACKUP_RECOVERY_BLOCKED')
    def test_standalone_cli_uses_no_package_dependencies(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'synthetic-standalone.db'
            with sqlite3.connect(path) as c:schema(c)
            p=subprocess.run([sys.executable,str(ROOT/'test_backup_recovery.py'),'--db-path',str(path)],capture_output=True,text=True)
            self.assertEqual(p.returncode,0,p.stderr);self.assertEqual(json.loads(p.stdout)['product'],'InFinBot')
    def test_keychain_wrapper_missing_evidence_never_passes(self):
        p=subprocess.run([sys.executable,str(ROOT/'verify_keychain_token.py'),'--token-file','/does/not/exist'],capture_output=True,text=True)
        self.assertEqual(p.returncode,1);self.assertFalse(json.loads(p.stdout)['external_token_loaded'])
    def test_live_writer_can_commit_during_read_only_drill(self):
        with tempfile.TemporaryDirectory() as tmp:
            path=Path(tmp)/'synthetic-concurrent.db'
            with sqlite3.connect(path) as c:
                c.execute('PRAGMA journal_mode=WAL');schema(c)
                c.executemany('INSERT INTO enterprise_portfolio(value) VALUES(?)',[('SYNTHETIC_PAGE_'+'x'*1000,) for _ in range(5000)]);c.commit()
            stop=threading.Event();started=threading.Event();count=[0];errors=[]
            def write():
                try:
                    with sqlite3.connect(path,timeout=1) as c:
                        c.execute('PRAGMA wal_autocheckpoint=0')
                        while not stop.is_set():
                            c.execute('INSERT INTO notification_outbox(value) VALUES(?)',('SYNTHETIC_CONCURRENT_WRITER',));c.commit();count[0]+=1;started.set()
                except Exception as e:errors.append(type(e).__name__)
            thread=threading.Thread(target=write);thread.start()
            try:
                self.assertTrue(started.wait(2));r=execute_non_destructive_backup_test(path,timeout_seconds=10)
                self.assertEqual(r['status'],'NATIVE_SNAPSHOT_REOPEN_AND_SCHEMA_COUNTS_VERIFIED',r)
                self.assertEqual(r['required_table_row_counts']['enterprise_portfolio'],5000)
            finally:stop.set();thread.join(2)
            self.assertFalse(thread.is_alive());self.assertFalse(errors);self.assertGreater(count[0],0)

if __name__=='__main__':unittest.main()