import contextlib import hashlib import io import json from pathlib import Path import runpy import subprocess import tarfile import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[2] helper = runpy.run_path(str(ROOT / 'scripts/deploy-update-remote.py')) remote = runpy.run_path(str(ROOT / 'scripts/backup-remote.py'), init_globals=helper) local = runpy.run_path(str(ROOT / 'scripts/backup.py')) backup = remote['backup'] class Controlled: service = 'werkjournal-backend.service' def __init__(self, home): self.home = home self.app = home / 'opt/werkjournal' self.site = home / 'site' self.site.mkdir() self.flag = self.site / 'maintenance.flag' self.pending = self.app / 'pending-deployment.json' release = self.app / 'releases' / ('a' * 40) release.mkdir(parents=True) self.jar = b'fixture application' digest = hashlib.sha256(self.jar).hexdigest() self.expected = {'schemaVersion': 1, 'runtime': 'java-25', 'commit': 'a' * 40, 'version': 'fixture', 'sha256': digest, 'files': {'werkjournal.jar': digest}} (release / 'werkjournal.jar').write_bytes(self.jar) (release / 'manifest.json').write_text(json.dumps(self.expected)) (self.app / 'current').symlink_to(release) (self.app / 'deployed.json').write_text(json.dumps(self.expected)) (self.app / 'data').mkdir() self.database = self.app / 'data/werkjournal.mv.db' self.database.write_bytes(b'before stop') self.running = True self.calls = [] self.fail_ready = False def preflight(self): pass def command(self, *args): if '--property=ActiveState' in args: return 'active' if self.running else 'inactive' return '0' def maintenance(self): self.flag.write_text('maintenance') def stop(self): self.calls.append('stop') self.running = False self.database.write_bytes(b'last transaction flushed during stop') def start(self): self.calls.append('start') self.running = True def ready(self, expected): self.calls.append('ready') if self.fail_ready: raise RuntimeError('not ready') class BackupTest(unittest.TestCase): def setUp(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.root = Path(directory.name) self.deployment = Controlled(self.root) def archive(self): output = io.BytesIO() backup(self.deployment, output) path = self.root / 'captured.tgz' path.write_bytes(output.getvalue()) return path def test_backup_captures_flushed_database_and_matching_jar_then_restores_running_state(self): path = self.archive() metadata = local['verify'](path) self.assertEqual(hashlib.sha256(b'last transaction flushed during stop').hexdigest(), metadata['files']['data/werkjournal.mv.db']) self.assertEqual(self.deployment.expected, metadata['release']) self.assertEqual(['stop', 'start', 'ready'], self.deployment.calls) self.assertTrue(self.deployment.running) self.assertFalse(self.deployment.flag.exists()) self.assertEqual([], list(self.deployment.app.glob('manual-backup-*'))) def test_stopped_service_and_existing_maintenance_are_preserved(self): self.deployment.running = False self.deployment.flag.write_text('Operator-owned maintenance') metadata = local['verify'](self.archive()) self.assertEqual(hashlib.sha256(b'before stop').hexdigest(), metadata['files']['data/werkjournal.mv.db']) self.assertFalse(self.deployment.running) self.assertEqual([], self.deployment.calls) self.assertEqual('Operator-owned maintenance', self.deployment.flag.read_text()) def test_copy_failure_restores_service_and_publishes_no_archive(self): output = io.BytesIO() with patch.object(remote['shutil'], 'copytree', side_effect=OSError('disk full')): with self.assertRaisesRegex(OSError, 'disk full'): backup(self.deployment, output) self.assertTrue(self.deployment.running) self.assertFalse(self.deployment.flag.exists()) self.assertEqual(b'', output.getvalue()) def test_readiness_failure_keeps_maintenance_and_publishes_no_archive(self): self.deployment.fail_ready = True output = io.BytesIO() with self.assertRaisesRegex(RuntimeError, 'not ready'): backup(self.deployment, output) self.assertTrue(self.deployment.flag.exists()) self.assertEqual(b'', output.getvalue()) def test_pending_deployment_refuses_backup_before_stopping(self): self.deployment.pending.write_text('{}') with self.assertRaisesRegex(RuntimeError, 'pending'): self.archive() self.assertEqual([], self.deployment.calls) def test_download_is_private_verified_and_never_overwrites(self): payload = self.archive().read_bytes() destination = self.root / 'backups/result.tgz' def ssh(*args, **kwargs): kwargs['stdout'].write(payload) return subprocess.CompletedProcess([], 0) with patch.object(subprocess, 'run', side_effect=ssh) as run: local['download']('fixture@host', destination) self.assertEqual(0o600, destination.stat().st_mode & 0o777) self.assertEqual(payload, destination.read_bytes()) self.assertEqual([], list(destination.parent.glob('.werkjournal-backup-*'))) with self.assertRaises(FileExistsError): local['download']('fixture@host', destination) run.assert_called_once() def test_truncated_transfer_or_remote_failure_never_publishes_backup(self): destination = self.root / 'result.tgz' for status in (0, 1): def ssh(*args, **kwargs): kwargs['stdout'].write(b'not an archive') return subprocess.CompletedProcess([], status) with patch.object(subprocess, 'run', side_effect=ssh): with self.assertRaises(Exception): local['download']('fixture@host', destination) self.assertFalse(destination.exists()) self.assertEqual([], list(self.root.glob('.werkjournal-backup-*'))) def test_valid_archive_with_changed_database_fails_checksum_verification(self): source = self.archive() altered = self.root / 'altered.tgz' with tarfile.open(source, 'r:gz') as original, tarfile.open(altered, 'w:gz') as output: for entry in original: data = original.extractfile(entry).read() if entry.name == 'data/werkjournal.mv.db': data += b'changed' entry.size = len(data) output.addfile(entry, io.BytesIO(data)) with self.assertRaisesRegex(ValueError, 'checksums'): local['verify'](altered)