"""Real local crypt/archive rehearsal tests; Docker DB isolation has a separate opt-in test."""
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tarfile
import tempfile
import time
from types import SimpleNamespace
import unittest
from email.parser import BytesParser
from email import policy

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / 'lib'))
import backup_set
import restore_test


@unittest.skipUnless(os.geteuid() == 0 and shutil.which('rclone'), 'run in test container')
class RestoreTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='restore-rehearsal-')
        self.addCleanup(self.temp.cleanup)
        self.base = Path(self.temp.name)
        self.work = self.base/'work'
        self.work.mkdir()
        self.root = self.work/'restore-test'
        self.sentinel = self.work/'keep-backup.txt'
        self.sentinel.write_text('real backup must survive')
        self.config = self.base/'config'
        self.crypt = self.base/'crypt.conf'
        password = subprocess.check_output(['rclone','obscure','restore-test-only'],text=True).strip()
        self.crypt.write_text(f'[crypt]\ntype=crypt\nremote={self.base}/encrypted\npassword={password}\n')
        self.crypt.chmod(0o600)
        self.sendmail = self.base/'sendmail'
        self.sendmail.write_text(f'#!/bin/sh\ncat > "{self.base}/mail.eml"\n')
        self.sendmail.chmod(0o700)
        self.config.write_text(f'''WORKDIR="{self.work}"
LOG_DIR="{self.base}/logs"
LOCKFILE="{self.base}/lock"
ENABLE_RESTORE_TEST=true
ENABLE_WORDPRESS=true
WP_DIR="{self.base}/source"
ENABLE_DB_DUMPS=false
ENABLE_MAIL=false
ENABLE_NEXTCLOUD=false
RCLONE_CONFIG="{self.crypt}"
RCLONE_REMOTE_BASE="crypt:sets"
RCLONE_RETRIES=1
RCLONE_LOW_LEVEL_RETRIES=1
RCLONE_CHECK_RETRY_DELAY=0
RESTORE_TEST_MIN_FREE_GB=0
RESTORE_TEST_MAX_WORK_GB=1
RESTORE_TEST_TIMEOUT=30
MAIL_HOSTNAME="gemini.jr-it-services.de"
MAIL_TO="backup@example.invalid"
MAIL_FROM="app-backup@example.invalid"
MAIL_SENDMAIL_BIN="{self.sendmail}"
''')
        self.config.chmod(0o600)
        self.run = 'appbackup_'+time.strftime('%Y-%m-%d_%H-%M-%S',time.gmtime())+'_12345678'
        self.folder = self.base/self.run
        self.folder.mkdir()
        self.make_archive('meta', 'meta', b'{}')
        self.make_archive('wordpress', 'files/wordpress', b'expected data\n')
        self.manifest()
        self.upload()

    def make_archive(self, label, root, content):
        with tarfile.open(self.folder/(label+'.tar.gz'),'w:gz') as tar:
            if not root.endswith('.sql'):
                entry = tarfile.TarInfo(root)
                entry.type = tarfile.DIRTYPE
                entry.mode = 0o700
                tar.addfile(entry)
                root += '/payload.txt'
            entry = tarfile.TarInfo(root)
            entry.mode = 0o600
            entry.size = len(content)
            tar.addfile(entry,io.BytesIO(content))

    def manifest(self):
        entries = []
        for archive in sorted(self.folder.glob('*.tar.gz')):
            label = archive.name.removesuffix('.tar.gz')
            entries.append(dict(label=label, source='/source', root=backup_set.expected_root(label),
                                file=archive.name,size=archive.stat().st_size,sha256=backup_set.digest(archive)))
        data = dict(format=2,run_id=self.run,created_epoch=int(time.time()),archives=entries)
        self.write_manifest(data)

    def write_manifest(self, data):
        (self.folder/'manifest.json').write_text(json.dumps(data))
        (self.folder/'COMPLETE').write_text('APP_BACKUP_V2 '+backup_set.digest(self.folder/'manifest.json')+'\n')

    def rc(self,*args):
        return subprocess.run(['rclone','--config',str(self.crypt),*args],check=True,capture_output=True)

    def upload(self):
        self.rc('copy',str(self.folder),'crypt:sets/'+self.run,'--ignore-times')

    def launch(self,*args,**kw):
        return subprocess.Popen(['bash',str(REPO/'usr-local-sbin/app-restore-test.sh'),'--config',str(self.config),*args],
                                stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,**kw)

    def execute(self,*args,ok=True):
        p = self.launch(*args)
        output = p.communicate(timeout=60)[0]
        self.assertEqual(p.returncode,0 if ok else 1,output)
        return output

    def clean(self):
        self.assertFalse((self.root/'scratch').exists())
        self.assertFalse((self.root/'db-created').exists())
        self.assertFalse((self.root/'child.json').exists())
        self.assertEqual(self.sentinel.read_text(),'real backup must survive')

    def state(self):
        return json.loads((self.root/'state.json').read_text())

    def test_full_crypt_download_extract_report_and_cleanup(self):
        self.execute()
        state = self.state()
        self.assertEqual(state['status'],'SUCCESS')
        self.assertEqual(state['cleanup'],'OK')
        self.assertEqual([c['label'] for c in state['components']],['meta','wordpress'])
        self.assertTrue(all(c['status']=='SUCCESS' for c in state['components']))
        self.clean()
        mail = BytesParser(policy=policy.default).parsebytes((self.base/'mail.eml').read_bytes())
        self.assertIn('[app-restore-test] SUCCESS gemini.jr-it-services.de',str(mail['Subject']))
        self.assertIn('Bereinigung: OK',mail.get_content())
        self.assertNotIn('expected data',mail.get_content())
        self.assertEqual((self.root/'last-success.json').stat().st_mode & 0o777,0o600)
        self.execute('--cleanup')
        self.clean()

    def test_stale_completed_backup_fails_and_cleans(self):
        data=json.loads((self.folder/'manifest.json').read_text())
        data['created_epoch']-=4*86400
        self.write_manifest(data)
        self.upload()
        self.execute(ok=False)
        self.assertIn('stale',self.state()['error'])
        self.clean()

    def test_corrupt_archive_is_not_success_and_cleans(self):
        (self.folder/'wordpress.tar.gz').write_bytes(b'corrupt')
        self.upload()
        self.execute(ok=False)
        self.assertEqual(self.state()['status'],'FAIL')
        self.clean()

    def download_failure(self, mode, target='wordpress.tar.gz'):
        # Real crypt transfers surround injected stream failures. stderr includes
        # a fake credential to prove that diagnostics never copy raw messages.
        wrapper = self.base/'test-rclone'
        counter = self.base/'attempts'
        wrapper.write_text(f'''#!/usr/bin/env python3
import os, pathlib, subprocess, sys, time
args = sys.argv[1:]
if 'cat' in args and any(a.endswith('/' + {target!r}) for a in args):
 counter = pathlib.Path({str(counter)!r})
 attempt = int(counter.read_text()) + 1 if counter.exists() else 1
 counter.write_text(str(attempt))
 assert args[args.index('--timeout')+1] == '300s', args
 assert args[args.index('--low-level-retries')+1] == '3', args
 mode = {mode!r}
 if mode == 'signal': sys.exit(143)
 if mode == 'oversize':
  sys.stdout.buffer.write(b'x'*1048576)
  sys.exit(0)
 if mode == 'hang' or (mode == 'hang-once' and attempt == 1):
  sys.stdout.buffer.write(b'partial')
  sys.stdout.buffer.flush()
  time.sleep(20)
 if mode in ('always-fail', 'stderr-flood') or (mode == 'fail-once' and attempt == 1):
  # Larger than the pipe capacity; the supervisor must drain stderr concurrently.
  sys.stderr.write(('https://example.invalid/?token=SECRET-DO-NOT-LOG HTTP 503 bad magic string\\n') * (2000 if mode == 'stderr-flood' else 1))
  sys.stdout.buffer.write(b'partial')
  sys.exit(5)
os.execv('/usr/bin/rclone', ['/usr/bin/rclone', *args])
''')
        wrapper.chmod(0o700)
        with self.config.open('a') as f: f.write(f'RCLONE_BIN="{wrapper}"\n')
        return counter

    def test_partial_download_restarts_cleanly_and_checks_real_content(self):
        counter = self.download_failure('fail-once')
        output = self.execute()
        self.assertEqual(counter.read_text(),'2')
        self.assertIn('crypt-header',output)
        self.assertIn('http-503',output)
        self.assertIn('exit 5',output)
        self.assertNotIn('SECRET-DO-NOT-LOG',output)
        self.assertEqual(self.state()['status'],'SUCCESS')
        self.assertTrue(all(c['phase']=='complete' for c in self.state()['components']))
        self.clean()

    def test_partial_metadata_read_is_not_appended_on_retry(self):
        counter = self.download_failure('fail-once', 'manifest.json')
        self.execute()
        self.assertEqual(counter.read_text(),'2')
        self.assertEqual(self.state()['status'],'SUCCESS')
        self.clean()

    def test_database_stderr_stays_private_with_operation_and_exit_code(self):
        rehearsal = restore_test.Rehearsal(SimpleNamespace(workdir=str(self.work), mode='run', timeout=10))
        with self.assertRaises(restore_test.CommandError) as result:
            rehearsal.command([sys.executable, '-c', "import sys; sys.stderr.write('SQL-SECRET'); sys.exit(2)"],
                              operation='Database import')
        self.assertIn('Database import failed (exit 2', str(result.exception))
        self.assertNotIn('SQL-SECRET', str(result.exception))
        self.assertNotIn('SQL-SECRET', (self.root/'restore-test.log').read_text())
        self.clean()

    def test_persistent_download_error_reports_phase_and_preserves_previous_success(self):
        self.execute()
        previous = (self.root/'last-success.json').read_bytes()
        counter = self.download_failure('always-fail')
        output = self.execute(ok=False)
        self.assertEqual(counter.read_text(),'3')
        self.assertIn('Component wordpress / download: Remote cat failed (exit 5',output)
        self.assertIn('crypt-header',self.state()['error'])
        self.assertEqual((self.root/'last-success.json').read_bytes(),previous)
        for path in ['state.json','restore-test.log','report.eml']:
            self.assertNotIn('SECRET-DO-NOT-LOG',(self.root/path).read_text())
        self.clean()

    def test_remote_stderr_flood_does_not_deadlock_or_leak(self):
        counter = self.download_failure('stderr-flood')
        output = self.execute(ok=False)
        self.assertEqual(counter.read_text(),'3')
        self.assertIn('http-503',output)
        self.assertNotIn('SECRET-DO-NOT-LOG',output)
        self.assertLess(len(output),12000)
        self.clean()

    def test_stalled_download_retries_from_empty_file(self):
        counter = self.download_failure('hang-once')
        with self.config.open('a') as f: f.write('RCLONE_CHECK_DEADLINE=2\n')
        output = self.execute()
        self.assertIn('deadline exceeded',output)
        self.assertEqual(counter.read_text(),'2')
        self.clean()

    def test_stalled_download_exhausts_attempts_and_cleans(self):
        counter = self.download_failure('hang')
        with self.config.open('a') as f: f.write('RCLONE_CHECK_DEADLINE=2\nRCLONE_CHECK_ATTEMPTS=2\n')
        self.execute(ok=False)
        self.assertEqual(counter.read_text(),'2')
        self.assertIn('deadline exceeded',self.state()['error'])
        self.assertEqual(self.state()['cleanup'],'OK')
        self.clean()

    def test_download_signal_does_not_retry(self):
        counter = self.download_failure('signal')
        self.execute(ok=False)
        self.assertEqual(counter.read_text(),'1')
        self.assertIn('exit 143',self.state()['error'])
        self.clean()

    def test_oversized_download_does_not_retry(self):
        counter = self.download_failure('oversize')
        self.execute(ok=False)
        self.assertEqual(counter.read_text(),'1')
        self.assertIn('exceeds expected size',self.state()['error'])
        self.clean()

    def test_global_timeout_stops_download_retry_backoff(self):
        counter = self.download_failure('always-fail')
        with self.config.open('a') as f:
            f.write('RESTORE_TEST_TIMEOUT=5\nRCLONE_CHECK_RETRY_DELAY=10\n')
        self.execute(ok=False)
        self.assertEqual(counter.read_text(),'1')
        self.assertIn('time limit exceeded',self.state()['error'])
        self.clean()

    def test_disk_budget_refuses_work_and_cleans(self):
        with self.config.open('a') as f: f.write('RESTORE_TEST_MAX_WORK_GB=0.000001\n')
        self.execute(ok=False)
        self.assertIn('budget',self.state()['error'])
        self.clean()

    def test_reserve_refuses_work_and_cleans(self):
        with self.config.open('a') as f: f.write('RESTORE_TEST_MIN_FREE_GB=100000\n')
        self.execute(ok=False)
        self.assertIn('reserve',self.state()['error'])
        self.clean()

    def test_missing_configured_component_fails(self):
        data=json.loads((self.folder/'manifest.json').read_text())
        data['archives']=[a for a in data['archives'] if a['label']=='meta']
        self.write_manifest(data)
        self.upload()
        self.execute(ok=False)
        self.assertIn('missing configured',self.state()['error'])
        self.clean()

    def test_latest_incomplete_set_is_skipped(self):
        self.rc('mkdir','crypt:sets/appbackup_2099-01-01_00-00-00_abcdef12')
        self.execute()
        self.assertEqual(self.state()['run'],self.run)
        self.clean()

    def test_scratch_symlink_is_never_followed(self):
        self.root.mkdir(mode=0o700)
        (self.root/'.owner').write_text(restore_test.MARKER)
        (self.root/'scratch').symlink_to(self.work,target_is_directory=True)
        self.execute('--cleanup',ok=False)
        self.assertEqual(self.sentinel.read_text(),'real backup must survive')
        self.assertTrue((self.root/'scratch').is_symlink())

    def test_unknown_directory_is_never_deleted(self):
        self.root.mkdir(mode=0o700)
        (self.root/'precious').write_text('keep')
        self.execute('--cleanup',ok=False)
        self.assertEqual((self.root/'precious').read_text(),'keep')

    def test_check_preserves_interrupted_cleanup_markers(self):
        self.root.mkdir(mode=0o700)
        (self.root/'.owner').write_text(restore_test.MARKER)
        (self.root/'scratch').mkdir()
        (self.root/'child.json').write_text('{"pid": 99999999, "start": "old"}')
        self.execute('--check',ok=False)
        self.assertTrue((self.root/'child.json').exists())
        self.assertTrue((self.root/'scratch').exists())

    def test_corrupt_state_does_not_block_cleanup(self):
        self.root.mkdir(mode=0o700)
        (self.root/'.owner').write_text(restore_test.MARKER)
        (self.root/'scratch').mkdir()
        (self.root/'scratch/old-data').write_text('remove')
        (self.root/'state.json').write_text('incomplete json')
        self.execute('--cleanup',ok=False)
        self.clean()
        self.assertEqual(self.state()['cleanup'],'OK')

    def test_unreachable_docker_does_not_claim_cleanup_success(self):
        self.root.mkdir(mode=0o700)
        (self.root/'.owner').write_text(restore_test.MARKER)
        (self.root/'scratch').mkdir()
        (self.root/'scratch/database-file').write_text('potentially still in use')
        (self.root/'db-created').write_text('pending')
        (self.root/'state.json').write_text(json.dumps(dict(status='RUNNING',cleanup='PENDING',started_epoch=int(time.time()),components=[],run=self.run)))
        binpath=self.base/'bin'
        binpath.mkdir()
        docker=binpath/'docker'
        docker.write_text('#!/bin/sh\nexit 42\n')
        docker.chmod(0o700)
        env=dict(os.environ,PATH=str(binpath)+':'+os.environ['PATH'])
        p=self.launch('--cleanup',env=env)
        output=p.communicate(timeout=20)[0]
        self.assertEqual(p.returncode,1,output)
        self.assertEqual(self.state()['cleanup'],'FAILED')
        self.assertTrue((self.root/'scratch/database-file').exists())
        self.assertTrue((self.root/'db-created').exists())
        docker.write_text('#!/bin/sh\nexit 0\n')
        p=self.launch('--cleanup',env=env)
        output=p.communicate(timeout=20)[0]
        self.assertEqual(p.returncode,1,output)  # Previous run remains failed.
        self.clean()
        self.assertEqual(self.state()['cleanup'],'OK')

    def slow_download(self):
        wrapper = self.base/'slow-rclone'
        wrapper.write_text('''#!/bin/bash
for arg in "$@"; do
  if [[ $arg == *wordpress.tar.gz ]]; then sleep 20; fi
done
exec /usr/bin/rclone "$@"
''')
        wrapper.chmod(0o700)
        with self.config.open('a') as f: f.write(f'RCLONE_BIN="{wrapper}"\n')

    def wait_download(self,p):
        end=time.monotonic()+20
        while time.monotonic()<end:
            if p.poll() is not None:
                self.fail(p.communicate()[0])
            if (self.root/'state.json').exists():
                state=self.state()
                if len(state['components'])==2 and (self.root/'child.json').exists():
                    return
            time.sleep(0.05)
        self.fail('test did not reach download')

    def test_term_cleans_partial_download(self):
        self.slow_download()
        p=self.launch()
        self.wait_download(p)
        p.send_signal(signal.SIGTERM)
        output=p.communicate(timeout=20)[0]
        self.assertEqual(p.returncode,1,output)
        self.clean()
        self.assertEqual(self.state()['status'],'FAIL')

    def test_sigkill_recovery_cleans_orphan_and_reports_failure(self):
        self.slow_download()
        p=self.launch()
        self.wait_download(p)
        p.kill()
        p.communicate(timeout=30)
        self.execute('--cleanup',ok=False)
        self.clean()
        self.assertEqual(self.state()['status'],'FAIL')
        self.assertIn('interrupted',self.state()['error'])

    def test_shared_lock_excludes_restore_test(self):
        lock=self.base/'lock'
        p=subprocess.Popen(['flock',str(lock),'sleep','20'])
        try:
            time.sleep(.2)
            test=self.launch()
            output=test.communicate(timeout=10)[0]
            self.assertEqual(test.returncode,75,output)
            self.assertFalse(self.root.exists())
        finally:
            p.terminate()
            p.wait()

    def test_unsafe_lock_is_not_treated_as_busy(self):
        (self.base/'lock').symlink_to(self.sentinel)
        self.execute(ok=False)
        self.assertFalse(self.root.exists())
        self.assertEqual(self.sentinel.read_text(),'real backup must survive')

    def prepare_database(self, sql):
        image=os.environ['RESTORE_TEST_REAL_DOCKER_IMAGE']
        self.make_archive('db-wordpress','db/wordpress.sql',sql)
        self.manifest()
        self.upload()
        with self.config.open('a') as f:
            f.write(f'ENABLE_DB_DUMPS=true\nWP_DB_NAME=production\nWP_DB_CNF=/unused\nRESTORE_TEST_DB_IMAGE="{image}"\nRESTORE_TEST_TIMEOUT=120\n')

    @unittest.skipUnless(os.environ.get('RESTORE_TEST_REAL_DOCKER_IMAGE'), 'opt-in local Docker integration')
    def test_installer_detects_mariadb_and_probes_real_image(self):
        self.prepare_database(b'CREATE TABLE example (id INT);')
        cnf=self.base/'backup.cnf'
        cnf.write_text('[client]\n')
        cnf.chmod(0o600)
        client=self.base/'version-client'
        client.write_text('#!/bin/sh\necho 10.11.18-MariaDB\n')
        client.chmod(0o700)
        with self.config.open('a') as f:
            f.write(f'WP_DB_CNF="{cnf}"\nDB_CLIENT_BIN="{client}"\nRESTORE_TEST_DB_IMAGE=""\n')
        result=subprocess.run(['bash',str(REPO/'lib/prepare-restore-test.sh'),str(self.config)],capture_output=True,text=True,timeout=120)
        self.assertEqual(result.returncode,0,result.stdout+result.stderr)
        self.assertIn('RESTORE_TEST_DB_IMAGE=mariadb:10.11',self.config.read_text())
        self.assertIn('initialization, import, check and cleanup passed',result.stdout)
        self.assertFalse((self.root/'last-success.json').exists())
        self.no_container()

    @unittest.skipUnless(os.environ.get('RESTORE_TEST_REAL_DOCKER_IMAGE'), 'opt-in local Docker integration')
    def test_installer_rejects_wrong_mariadb_series_without_config_change(self):
        self.prepare_database(b'CREATE TABLE example (id INT);')
        cnf=self.base/'backup.cnf'
        cnf.write_text('[client]\n')
        cnf.chmod(0o600)
        client=self.base/'version-client'
        client.write_text('#!/bin/sh\necho 11.4.1-MariaDB\n')
        client.chmod(0o700)
        with self.config.open('a') as f: f.write(f'WP_DB_CNF="{cnf}"\nDB_CLIENT_BIN="{client}"\n')
        original=self.config.read_bytes()
        result=subprocess.run(['bash',str(REPO/'lib/prepare-restore-test.sh'),str(self.config)],capture_output=True,text=True,timeout=60)
        self.assertNotEqual(result.returncode,0,result.stdout+result.stderr)
        self.assertIn('does not match production MariaDB series',result.stdout+result.stderr)
        self.assertEqual(self.config.read_bytes(),original)
        self.assertFalse(self.root.exists())

    def test_space_recovery_uses_parent_lock_and_preserves_three_local_sets(self):
        self.work.chmod(0o700)
        for name in ('archives','staging','restore'): (self.work/name).mkdir(mode=0o700)
        old=[]
        for day in range(1,5):
            run=f'appbackup_2020-01-0{day}_00-00-00_aaaaaaaa'
            path=self.work/'archives'/run
            shutil.copytree(self.folder,path)
            path.chmod(0o700)
            data=json.loads((path/'manifest.json').read_text())
            data['run_id']=run
            (path/'manifest.json').write_text(json.dumps(data))
            (path/'COMPLETE').write_text('APP_BACKUP_V2 '+backup_set.digest(path/'manifest.json')+'\n')
            (path/'UPLOADED').write_text('verified previously\n')
            subprocess.run(['rclone','--config',str(self.crypt),'copy',str(path),'crypt:sets/'+run],check=True,capture_output=True)
            old.append(path)
        with self.config.open('a') as f: f.write('RESTORE_TEST_MIN_FREE_GB=0.001\n')
        # Simulate pressure consistently for the Python supervisor and Bash df.
        # Once one old local set is removed, the real free-space figure returns.
        bindir=self.base/'bin'
        bindir.mkdir()
        script=self.base/'supervisor.py'
        script.write_text('import os,runpy,shutil,sys\nfrom pathlib import Path\noriginal=shutil.disk_usage\npressure=Path('+repr(str(old[0]))+')\ndef usage(p):\n value=original(p)\n return value._replace(free=0) if pressure.exists() else value\nshutil.disk_usage=usage\npath=sys.argv.pop(1)\nsys.path.insert(0,str(Path(path).parent))\nrunpy.run_path(path,run_name="__main__")\n')
        python=bindir/'python3'
        python.write_text('#!/bin/sh\ncase "$1" in */restore_test.py) exec /usr/bin/python3 "'+str(script)+'" "$@";; esac\nexec /usr/bin/python3 "$@"\n')
        python.chmod(0o700)
        df=bindir/'df'
        df.write_text('#!/bin/sh\nif test -d "'+str(old[0])+'"; then printf "Filesystem 1B-blocks Used Available Use%% Mounted\\nroot 100 100 0 100%% /\\n"; else exec /usr/bin/df "$@"; fi\n')
        df.chmod(0o700)
        result=subprocess.run(['bash',str(REPO/'usr-local-sbin/app-restore-test.sh'),'--config',str(self.config)],env=dict(os.environ,PATH=str(bindir)+':'+os.environ['PATH']),capture_output=True,text=True,timeout=60)
        self.assertEqual(result.returncode,0,result.stdout+result.stderr)
        self.assertFalse(old[0].exists())
        self.assertTrue(all(p.exists() for p in old[1:]))
        self.assertEqual(self.state()['status'],'SUCCESS')
        self.clean()

    def test_adaptive_budget_accepts_zero(self):
        with self.config.open('a') as f: f.write('RESTORE_TEST_MAX_WORK_GB=0\n')
        self.execute()
        self.clean()
        self.assertEqual(self.state()['status'],'SUCCESS')

    def no_container(self):
        self.clean()
        names=subprocess.check_output(restore_test.DOCKER+['ps','-a','--filter',f'label={restore_test.OWNER_LABEL}', '--format','{{.Names}}'],text=True)
        self.assertNotIn('app-backup-restore-test-'+hashlib.sha256(str(self.root).encode()).hexdigest()[:16],names)

    @unittest.skipUnless(os.environ.get('RESTORE_TEST_REAL_DOCKER_IMAGE'), 'opt-in local Docker integration')
    def test_real_isolated_db_import_and_container_cleanup(self):
        self.prepare_database(b'CREATE TABLE example (id INT PRIMARY KEY); INSERT INTO example VALUES (1);')
        self.execute()
        state=self.state()
        db=next(c for c in state['components'] if c['label']=='db-wordpress')
        self.assertEqual(db['tables'],1)
        self.no_container()

    @unittest.skipUnless(os.environ.get('RESTORE_TEST_REAL_DOCKER_IMAGE'), 'opt-in local Docker integration')
    def test_real_db_sql_failure_cleans_container_and_data(self):
        self.prepare_database(b'CREATE TABLE example (id INT); invalid SQL contains-secret-value;')
        self.execute(ok=False)
        self.no_container()
        self.assertEqual(self.state()['cleanup'],'OK')
        self.assertNotIn('contains-secret-value',(self.root/'report.eml').read_text())
        self.assertNotIn('contains-secret-value',(self.root/'restore-test.log').read_text())

    @unittest.skipUnless(os.environ.get('RESTORE_TEST_REAL_DOCKER_IMAGE'), 'opt-in local Docker integration')
    def test_real_db_sigkill_recovery_removes_container_and_data(self):
        self.prepare_database(b'CREATE TABLE example (id INT); SELECT SLEEP(40);')
        p=self.launch()
        container='app-backup-restore-test-'+hashlib.sha256(str(self.root).encode()).hexdigest()[:16]
        end=time.monotonic()+20
        while time.monotonic()<end:
            result=subprocess.run(restore_test.DOCKER+['exec',container,'mariadb','--no-defaults','--socket=/var/lib/mysql/mysql.sock','-uroot','--execute=SELECT COUNT(*) FROM restore_test.example'],capture_output=True)
            if result.returncode==0:
                break
            time.sleep(.2)
        else:
            p.terminate()
            self.fail(p.communicate(timeout=20)[0])
        info=json.loads(subprocess.check_output(restore_test.DOCKER+['inspect',container]))[0]
        self.assertEqual(info['HostConfig']['NetworkMode'],'none')
        self.assertTrue(info['HostConfig']['ReadonlyRootfs'])
        self.assertEqual(info['HostConfig']['CapDrop'],['ALL'])
        self.assertEqual(info['HostConfig']['LogConfig']['Type'],'none')
        self.assertEqual(len(info['Mounts']),1)
        self.assertEqual(info['Mounts'][0]['Source'],str(self.root/'scratch/database'))
        p.kill()
        p.communicate(timeout=20)
        self.execute('--cleanup',ok=False)
        self.no_container()
        self.assertIn('interrupted',self.state()['error'])


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