import copy import hashlib import importlib.util import json import os import io from pathlib import Path import subprocess import tarfile import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[3] spec = importlib.util.spec_from_file_location('updater', ROOT / 'scripts/selfhost-update.py') u = importlib.util.module_from_spec(spec) spec.loader.exec_module(u) def data(value): return json.dumps(value).encode() class Http: def __init__(self): self.manifest = {'schemaVersion': 1, 'version': '3.1.0', 'tag': 'v3.1.0', 'commit': 'a' * 40, 'repository': u.REPOSITORY, 'migrations': [], 'images': [{'name': name, 'repository': u.REGISTRY + '/' + name, 'digest': 'sha256:' + char * 64} for name, char in [('backend', 'a'), ('frontend', 'b')]]} self.files = {name: (ROOT / {'docker-compose.selfhost.yml': 'docker-compose.selfhost.yml', '.env.example': '.env.example', 'selfhost-setup.sh': 'scripts/selfhost-setup.sh'}[name]).read_bytes() for name in ['docker-compose.selfhost.yml', '.env.example', 'selfhost-setup.sh']} self.files.update({'release-manifest.json': data(self.manifest), 'CHANGELOG.md': b'New release'}) self.refresh() self.requests = [] self.bad_health = False self.release = {'id': 1, 'tag_name': 'v3.1.0', 'draft': False, 'prerelease': False} self.fail_download = False self.reported_version = '3.1.0' def refresh(self): self.files['SHA256SUMS'] = ''.join(hashlib.sha256(blob).hexdigest() + ' ' + name + '\n' for name, blob in self.files.items() if name != 'SHA256SUMS').encode() def get(self, url, **kwargs): self.requests.append(url) if url.endswith('/'): return b'FEDEO' if self.fail_download: raise u.UpdateError('Download fehlgeschlagen') return self.files[url.rsplit('/', 1)[-1]] def json(self, url): self.requests.append(url) if url.endswith('/releases/tags/v3.1.0'): return self.release if '/releases?limit=' in url: return [self.release, {'tag_name': 'v9.0.0', 'draft': True}, {'tag_name': 'v4.0.0-rc.1', 'prerelease': True}] if url.endswith('/releases/1/assets'): return [{'name': name, 'browser_download_url': u.REPOSITORY + '/assets/' + name} for name in self.files] if url.endswith('/tags/v3.1.0'): return {'commit': {'sha': 'a' * 40}} if url.endswith('/backend/version'): return {'product': 'FEDEO', 'version': self.reported_version} if url.endswith('/health'): return {'status': 'ok', 'product': 'FEDEO', 'version': '3.0.0' if self.bad_health else '3.1.0', 'revision': 'a' * 40} raise AssertionError(url) class Runner: """Real Compose parsing with a simulated daemon; no containers are started.""" def __init__(self, log, http): self.log = log self.http = http self.calls = [] self.fail = None self.on_pull = None self.interrupt = None self.matrix_containers = {} self.on_stop = None self.real_tar = False def run(self, args, output=None, input_file=None, **kwargs): self.calls.append(args) if self.interrupt and self.interrupt(args): raise KeyboardInterrupt() if self.fail and self.fail(args): raise u.UpdateError('simulated command failure') if args[:3] == ['docker', 'compose', 'version']: return '2.39.1' if args[:2] == ['docker', 'info']: return json.dumps({'OSType': 'linux', 'Architecture': 'amd64', 'DockerRootDir': '/tmp'}) if args[:3] == ['docker', 'context', 'inspect']: return 'unix:///var/run/docker.sock' if args[:3] == ['docker', 'image', 'inspect']: return 'image-' + args[3].rsplit('/', 1)[-1].split('@')[0] if args[:2] == ['docker', 'inspect']: name = args[2] if name in self.matrix_containers: return json.dumps([self.matrix_containers[name]]) return json.dumps([{'Image': 'image-' + name, 'Config': {'Labels': {'com.docker.compose.service': name}}, 'State': {'Running': name != 'createbuckets', 'Health': {'Status': 'healthy'}}}]) if args[:3] == ['docker', 'update', '--restart=no']: self.matrix_containers[args[3]]['HostConfig']['RestartPolicy'] = {'Name': 'no', 'MaximumRetryCount': 0} return args[3] if 'config' in args: result = subprocess.run(args, env={'PATH': os.environ['PATH'], 'HOME': os.environ['HOME']}, capture_output=True, text=True) if result.returncode: raise u.UpdateError('Compose invalid: ' + result.stderr) return result.stdout if 'ps' in args: if args[:2] == ['docker', 'ps']: return '' if args[-1] in ('backend', 'frontend', 'db', 'minio') or args[-1] in self.matrix_containers: return args[-1] return 'backend\nfrontend\ndb\nminio\ntraefik\nnode-exporter\ncreatebuckets\n' + '\n'.join(self.matrix_containers) if 'stop' in args or 'start' in args: for name, container in self.matrix_containers.items(): if name in args: container['State']['Running'] = 'start' in args if 'stop' in args and self.on_stop: self.on_stop() if 'pull' in args and self.on_pull: self.on_pull() if any('pg_database_size' in part for part in args): return '1000' if args[0] == 'du': return '1000\tminio' if args[0] == 'tar' and self.real_tar: with open(output, 'wb') if output else open(os.devnull, 'wb') as destination: subprocess.run(args, stdout=destination, check=True) return '' if output: Path(output).write_bytes(b'PGDMP fake dump') if args[0] == 'tar' and '-cpf' in args: Path(args[2]).write_bytes(b'backup') return '' class UpdateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) (self.root / 'minio').mkdir() (self.root / 'postgres').mkdir() (self.root / '.env').write_bytes(u.set_version((ROOT / '.env.example').read_bytes(), '3.0.0')) (self.root / 'docker-compose.yml').write_bytes((ROOT / 'docker-compose.selfhost.yml').read_bytes()) self.original_env = (self.root / '.env').read_bytes() self.original_compose = (self.root / 'docker-compose.yml').read_bytes() self.http = Http() self.runner = Runner(self.root / 'log', self.http) self.updater = u.Updater(self.root, http=self.http, runner_factory=lambda log: self.runner) self.updater.check_space = lambda *_: None self.env_patch = patch.dict(os.environ, {'DOCKER_HOST': 'unix:///var/run/docker.sock'}) self.env_patch.start() self.sleep_patch = patch.object(u.time, 'sleep') self.sleep_patch.start() def tearDown(self): self.env_patch.stop() self.sleep_patch.stop() self.temp.cleanup() def run_update(self, **kwargs): return self.updater.execute('apply', '3.1.0', accept_compose=True, **kwargs) def called(self, value): return any(value in args for args in self.runner.calls) def assert_untouched(self): self.assertEqual((self.root / '.env').read_bytes(), self.original_env) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) self.assertFalse(self.called('stop')) def add_legacy_matrix(self): services = {} for name in sorted(u.LEGACY_MATRIX_SERVICES): services[name] = {'image': 'example.test/' + name + ':legacy', 'restart': 'unless-stopped', 'container_name': 'fedeo-' + name, 'networks': ['internal']} self.runner.matrix_containers[name] = { 'Id': name, 'Image': 'image-' + name, 'Config': {'Labels': {'com.docker.compose.service': name}}, 'Mounts': [], 'HostConfig': {'RestartPolicy': {'Name': 'unless-stopped', 'MaximumRetryCount': 0}}, 'State': {'Running': True, 'Health': {'Status': 'healthy'}}} for name, directory, target in [('matrix-db', 'postgres', '/var/lib/postgresql/data'), ('matrix-synapse', 'synapse', '/data')]: source = self.root / 'matrix' / directory source.mkdir(parents=True) (source / 'data').write_text('preserved ' + name) services[name]['volumes'] = [{'type': 'bind', 'source': str(source), 'target': target}] self.runner.matrix_containers[name]['Mounts'] = [{'Type': 'bind', 'Source': str(source), 'Destination': target}] # Use the real Compose parser to model an older base file, including the removed dependency. config = json.loads(subprocess.check_output(['docker', 'compose', '--project-directory', str(self.root), '--env-file', str(self.root / '.env'), '-f', str(self.root / 'docker-compose.yml'), 'config', '--format', 'json'], env={'PATH': os.environ['PATH'], 'HOME': os.environ['HOME']})) config['services'].update(services) config['services']['backend']['depends_on']['matrix-synapse'] = {'condition': 'service_healthy'} (self.root / 'docker-compose.yml').write_bytes(data(config)) self.original_compose = (self.root / 'docker-compose.yml').read_bytes() def test_check_plans_matrix_retirement_without_stopping_or_disabling_it(self): self.add_legacy_matrix() result = self.updater.execute('check', '3.1.0') self.assertEqual(set(result['plan']['retiredServices']), u.LEGACY_MATRIX_SERVICES) self.assertEqual(result['plan']['retainedDataPaths'], ['matrix']) self.assertEqual(result['plan']['estimatedBackupBytes'], 4000 + u.GIB) self.assertFalse(self.called('--restart=no')) self.assert_untouched() def test_matrix_is_backed_up_then_retired_without_removing_containers_or_data(self): self.add_legacy_matrix() # An image-created Redis volume must remain attached to the retained container. self.runner.matrix_containers['matrix-redis']['Mounts'] = [ {'Type': 'volume', 'Source': '/var/lib/docker/volumes/redis/_data', 'Destination': '/data', 'Name': 'redis'}] result = self.run_update() self.assertEqual(result['status'], 'succeeded') self.assertNotIn(b'matrix-', (self.root / 'docker-compose.yml').read_bytes()) self.assertTrue((self.root / 'matrix/postgres/data').exists()) self.assertTrue((self.root / 'matrix/synapse/data').exists()) backup = Path(result['backup']) self.assertEqual((backup / 'config/docker-compose.yml').read_bytes(), self.original_compose) metadata = u.read_json(backup / 'complete.json') self.assertEqual(metadata['matrixDataPaths'], ['matrix']) self.assertEqual(set(metadata['retainedContainerIds']), u.LEGACY_MATRIX_SERVICES) original = u.read_json(backup / 'containers.json') self.assertEqual(next(c for c in original if c.get('Id') == 'matrix-db')['HostConfig']['RestartPolicy']['Name'], 'unless-stopped') for container in self.runner.matrix_containers.values(): self.assertFalse(container['State']['Running']) self.assertEqual(container['HostConfig']['RestartPolicy']['Name'], 'no') calls = self.runner.calls archive = next(i for i, a in enumerate(calls) if a[0] == 'tar' and '-cpf' in a) retire = next(i for i, a in enumerate(calls) if '--restart=no' in a) self.assertIn('matrix', calls[archive]) self.assertLess(archive, retire) self.assertFalse(any('--remove-orphans' in a or a[:2] in (['docker', 'rm'], ['docker', 'volume']) for a in calls)) # Future updates see stopped Matrix orphans; they do not bring them back. (self.root / '.env').write_bytes(self.original_env) self.assertEqual(self.run_update()['status'], 'succeeded') self.assertFalse(any(c['State']['Running'] for c in self.runner.matrix_containers.values())) def test_matrix_archive_failure_preserves_base_and_restarts_original_stack(self): self.add_legacy_matrix() self.runner.fail = lambda a: a[0] == 'tar' and '-cpf' in a with self.assertRaises(u.UpdateError): self.run_update() self.assertFalse(self.called('--restart=no')) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) self.assertFalse(self.updater.status()['recoveryRequired']) self.assertTrue(all(c['State']['Running'] for c in self.runner.matrix_containers.values())) def test_real_matrix_archive_contains_database_media_and_configuration(self): self.add_legacy_matrix() self.runner.real_tar = True (self.root / 'matrix/livekit.yaml').write_text('preserved config') (self.root / 'matrix/postgres/pg_wal').mkdir() (self.root / 'matrix/postgres/pg_wal/segment').write_bytes(b'preserved WAL') expected = {str(path.relative_to(self.root)): path.read_bytes() for path in (self.root / 'matrix').rglob('*') if path.is_file()} result = self.run_update() backup = Path(result['backup']) with tarfile.open(backup / 'files.tar') as archive: for name, contents in expected.items(): self.assertEqual(archive.extractfile(name).read(), contents) self.assertEqual(u.sha((backup / 'files.tar').read_bytes()), u.read_json(backup / 'complete.json')['sha256']['files.tar']) def test_matrix_must_be_stopped_before_a_cold_backup(self): self.add_legacy_matrix() self.runner.on_stop = lambda: self.runner.matrix_containers['matrix-db']['State'].update(Running=True) with self.assertRaisesRegex(u.UpdateError, 'Matrix-Container läuft noch'): self.run_update() self.assertFalse(self.called('-cpf')) self.assertFalse(self.called('--restart=no')) def test_interrupted_retirement_requires_recovery_and_preserves_backup(self): self.add_legacy_matrix() self.runner.interrupt = lambda a: '--restart=no' in a and a[-1] == 'matrix-element' with self.assertRaises(KeyboardInterrupt): self.run_update() state = self.updater.status() self.assertEqual(state['phase'], 'retiring_matrix') self.assertTrue(state['recoveryRequired']) self.assertTrue((Path(state['backup']) / 'complete.json').exists()) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) self.assertFalse(any(c['State']['Running'] for c in self.runner.matrix_containers.values())) def test_matrix_symlinks_and_nonstandard_mounts_are_rejected_before_stopping(self): self.add_legacy_matrix() link = self.root / 'matrix/postgres/pg_wal' link.symlink_to(self.root / 'postgres', target_is_directory=True) with self.assertRaisesRegex(u.UpdateError, 'Symbolische Links'): self.run_update() self.assert_untouched() link.unlink() self.runner.matrix_containers['matrix-db']['Mounts'][0]['Source'] = '/different/postgres' with self.assertRaisesRegex(u.UpdateError, 'Tatsächliche Matrix-Datenpfade'): self.run_update() self.assert_untouched() def test_unknown_additional_services_are_still_rejected(self): self.add_legacy_matrix() config = json.loads(self.original_compose) config['services']['matrix-custom'] = {'image': 'example.test/custom:latest'} (self.root / 'docker-compose.yml').write_bytes(data(config)) with self.assertRaisesRegex(u.UpdateError, 'matrix-custom'): self.run_update() self.assertFalse(self.called('stop')) def test_matrix_cannot_be_reintroduced_by_an_override(self): self.add_legacy_matrix() (self.root / 'docker-compose.override.yml').write_text('services:\n matrix-element:\n image: example.test/element:legacy\n') with self.assertRaisesRegex(u.UpdateError, 'Zielkonfiguration.*matrix-element'): self.run_update() self.assert_untouched() def test_nonstandard_matrix_bind_and_named_volume_are_rejected(self): self.add_legacy_matrix() for volume in [{'type': 'bind', 'source': str(self.root / 'postgres'), 'target': '/data'}, {'type': 'volume', 'source': 'matrixdata', 'target': '/data'}]: with self.subTest(volume=volume['type']): config = json.loads(self.original_compose) config['services']['matrix-synapse']['volumes'] = [volume] config['volumes'] = {'matrixdata': {}} (self.root / 'docker-compose.yml').write_bytes(data(config)) with self.assertRaisesRegex(u.UpdateError, 'Matrix-Daten müssen'): self.run_update() self.assertFalse(self.called('stop')) def test_resolve_target_requires_matrix_to_remain_retired(self): self.add_legacy_matrix() self.runner.fail = lambda a: 'migrate' in a with self.assertRaises(u.UpdateError): self.run_update() self.runner.fail = None self.runner.matrix_containers['matrix-db']['HostConfig']['RestartPolicy']['Name'] = 'always' with self.assertRaisesRegex(u.UpdateError, 'automatisch neu starten'): self.updater.resolve('3.1.0') self.assertTrue(self.updater.status()['recoveryRequired']) self.runner.matrix_containers['matrix-db']['HostConfig']['RestartPolicy']['Name'] = 'no' self.assertEqual(self.updater.resolve('3.1.0')['status'], 'recovered') def test_resolve_rollback_requires_original_matrix_images_state_and_restart_policy(self): self.add_legacy_matrix() originals = copy.deepcopy(self.runner.matrix_containers) self.runner.fail = lambda a: 'migrate' in a with self.assertRaises(u.UpdateError): self.run_update() self.runner.fail = None self.http.reported_version = '3.0.0' (self.root / 'docker-compose.yml').write_bytes(self.original_compose) (self.root / '.env').write_bytes(self.original_env) with self.assertRaisesRegex(u.UpdateError, 'Matrix-Container und Neustartrichtlinien'): self.updater.resolve('3.0.0') self.assertTrue(self.updater.status()['recoveryRequired']) self.runner.matrix_containers = copy.deepcopy(originals) for key, value in [('Image', 'wrong-image'), ('HostConfig', {'RestartPolicy': {'Name': 'no'}}), ('State', {'Running': True, 'Health': {'Status': 'unhealthy'}})]: with self.subTest(key=key): self.runner.matrix_containers['matrix-db'][key] = value with self.assertRaisesRegex(u.UpdateError, 'Matrix-Container und Neustartrichtlinien'): self.updater.resolve('3.0.0') self.runner.matrix_containers = copy.deepcopy(originals) self.assertEqual(self.updater.resolve('3.0.0')['status'], 'recovered') self.assertFalse(self.updater.status()['recoveryRequired']) def test_success_pins_images_preserves_configuration_and_checks_both_public_services(self): result = self.run_update() self.assertEqual(result['status'], 'succeeded') self.assertFalse(result['recoveryRequired']) self.assertEqual((self.root / '.env').read_bytes(), u.set_version(self.original_env, '3.1.0')) self.assertIn(b'backend@sha256:' + b'a' * 64, (self.root / 'docker-compose.yml').read_bytes()) backup = Path(result['backup']) self.assertTrue((backup / 'complete.json').exists()) self.assertEqual((backup / 'config/.env').read_bytes(), self.original_env) self.assertTrue(any(url.endswith('/backend/health') for url in self.http.requests)) self.assertTrue(any(url.endswith('/api/health') for url in self.http.requests)) self.assertEqual((self.root / '.env').stat().st_mode & 0o777, 0o600) self.assertEqual((self.root / '.updates').stat().st_mode & 0o777, 0o700) self.assertFalse(self.called('down')) self.assertFalse(self.called('--remove-orphans')) calls = self.runner.calls self.assertLess(next(i for i, a in enumerate(calls) if 'pull' in a), next(i for i, a in enumerate(calls) if 'stop' in a)) self.assertLess(next(i for i, a in enumerate(calls) if 'pg_restore' in a), next(i for i, a in enumerate(calls) if 'migrate' in a)) def test_check_only_creates_a_plan_and_ignores_unpublished_releases(self): result = self.updater.execute('check') self.assertEqual(result['version'], '3.1.0') self.assertEqual(result['status'], 'checked') self.assertTrue(Path(result['plan']['composeDiff']).exists()) self.assert_untouched() self.assertFalse(self.called('pull')) def test_checksum_failure_does_not_stop_anything(self): self.http.files['docker-compose.selfhost.yml'] += b'\n# tampered' with self.assertRaisesRegex(u.UpdateError, 'Prüfsumme'): self.run_update() self.assert_untouched() def test_download_failure_does_not_stop_anything(self): self.http.fail_download = True with self.assertRaisesRegex(u.UpdateError, 'Download'): self.run_update() self.assert_untouched() def test_draft_is_rejected_even_when_explicitly_selected(self): self.http.release['draft'] = True with self.assertRaisesRegex(u.UpdateError, 'veröffentlichte'): self.run_update() self.assert_untouched() def test_manifest_commit_must_match_git_tag(self): self.http.manifest['commit'] = 'b' * 40 self.http.files['release-manifest.json'] = data(self.http.manifest) self.http.refresh() with self.assertRaisesRegex(u.UpdateError, 'Git-Tag'): self.run_update() self.assert_untouched() def test_invalid_compose_fails_before_stopping(self): self.http.files['docker-compose.selfhost.yml'] += b'\nbroken: [\n' self.http.refresh() with self.assertRaises(u.UpdateError): self.run_update() self.assert_untouched() def test_unknown_base_requires_review(self): with self.assertRaisesRegex(u.UpdateError, 'compose.diff'): self.updater.execute('apply', '3.1.0') self.assert_untouched() def test_override_is_preserved_but_cannot_replace_release_images(self): override = self.root / 'docker-compose.override.yml' contents = 'services:\n frontend:\n environment:\n MY_CUSTOM_SETTING: preserved\n' override.write_text(contents) self.run_update() self.assertEqual(override.read_text(), contents) self.assertEqual(json.loads((self.root / '.updates/installed.json').read_text())['manifest']['version'], '3.1.0') def test_override_cannot_disable_healthcheck(self): (self.root / 'docker-compose.override.yml').write_text('services:\n backend:\n healthcheck:\n disable: true\n') with self.assertRaisesRegex(u.UpdateError, 'Healthcheck'): self.run_update() self.assert_untouched() def test_changed_mount_is_rejected(self): self.http.files['docker-compose.selfhost.yml'] = self.http.files['docker-compose.selfhost.yml'].replace(b'./postgres:', b'./different:') self.http.refresh() with self.assertRaisesRegex(u.UpdateError, 'Datenpfade'): self.run_update() self.assert_untouched() def test_external_storage_fails_before_maintenance(self): (self.root / '.env').write_bytes(self.original_env.replace(b'http://minio:9000', b'https://storage.example.test')) with self.assertRaisesRegex(u.UpdateError, 'Backup-Adapter'): self.run_update() self.assertFalse(self.called('stop')) def test_database_url_must_match_the_database_being_backed_up(self): (self.root / '.env').write_bytes(self.original_env.replace(b'@db:5432/fedeo', b'@db:5432/other_database')) with self.assertRaisesRegex(u.UpdateError, 'gesicherte lokale'): self.run_update() self.assertFalse(self.called('stop')) def test_missing_required_secret_is_rejected(self): (self.root / '.env').write_bytes(self.original_env.replace(b'ENCRYPTION_KEY=', b'IGNORED_KEY=')) with self.assertRaisesRegex(u.UpdateError, 'ENCRYPTION_KEY'): self.run_update() self.assertFalse(self.called('stop')) def test_pull_failure_never_stops_old_stack(self): self.runner.fail = lambda args: 'pull' in args with self.assertRaises(u.UpdateError): self.run_update() self.assert_untouched() def test_configuration_changed_during_pull_is_detected(self): self.runner.on_pull = lambda: (self.root / '.env').write_bytes(self.original_env + b'\nNEW=value\n') with self.assertRaisesRegex(u.UpdateError, 'während'): self.run_update() self.assertFalse(self.called('stop')) def test_backup_failure_restarts_previous_services_and_preserves_configuration(self): self.runner.fail = lambda args: any('pg_dump ' in arg for arg in args) with self.assertRaises(u.UpdateError): self.run_update() self.assertTrue(self.called('start')) self.assertFalse(self.called('migrate')) self.assertFalse(self.updater.status()['recoveryRequired']) self.assertEqual((self.root / '.env').read_bytes(), self.original_env) self.assertEqual((self.root / 'docker-compose.yml').read_bytes(), self.original_compose) def test_migration_failure_keeps_recovery_state_and_never_starts_old_application(self): self.runner.fail = lambda args: 'migrate' in args with self.assertRaises(u.UpdateError): self.run_update() state = self.updater.status() self.assertEqual(state['phase'], 'migrating') self.assertTrue(state['recoveryRequired']) self.assertTrue(Path(state['backup']).exists()) self.assertFalse(self.called('start')) with self.assertRaisesRegex(u.UpdateError, 'Wiederherstellung'): self.run_update() def test_interrupt_during_migration_is_persisted(self): self.runner.interrupt = lambda args: 'migrate' in args with self.assertRaises(KeyboardInterrupt): self.run_update() self.assertEqual(self.updater.status()['status'], 'interrupted') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertFalse(self.called('start')) def test_process_death_is_detected_and_does_not_erase_recovery_state(self): job = self.root / '.updates/interrupted' job.mkdir() u.write_json(self.root / '.updates/status.json', {'status': 'running', 'job': str(job), 'recoveryRequired': True}) self.assertEqual(self.updater.status()['status'], 'interrupted') self.assertTrue(self.updater.status()['recoveryRequired']) with self.assertRaisesRegex(u.UpdateError, 'Wiederherstellung'): self.run_update() def test_lock_prevents_parallel_updates(self): with self.updater.lock(): other = u.Updater(self.root, http=self.http) with self.assertRaisesRegex(u.UpdateError, 'läuft bereits'): other.execute('check') def test_old_migrations_must_not_change(self): u.write_json(self.root / '.updates/installed.json', {'manifest': {'migrations': [{'path': 'old.sql', 'sha256': 'a' * 64}]}}) with self.assertRaisesRegex(u.UpdateError, 'Migrationen'): self.run_update() self.assert_untouched() def test_insufficient_space_does_not_stop_services(self): self.updater.check_space = lambda *_: (_ for _ in ()).throw(u.UpdateError('Zu wenig freier Speicher')) with self.assertRaisesRegex(u.UpdateError, 'Speicher'): self.run_update() self.assert_untouched() def test_failed_health_verification_does_not_mark_release_installed(self): self.http.bad_health = True with self.assertRaisesRegex(u.UpdateError, 'Funktionsprüfung'): self.run_update() self.assertEqual(self.updater.status()['phase'], 'verifying') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertFalse((self.root / '.updates/installed.json').exists()) self.assertEqual(self.runner.calls[-1][-3:], ['backend', 'frontend', 'traefik']) def test_resolve_requires_a_working_manually_recovered_installation(self): self.runner.fail = lambda args: 'migrate' in args with self.assertRaises(u.UpdateError): self.run_update() self.runner.fail = None with self.assertRaisesRegex(u.UpdateError, 'Version stimmt nicht'): self.updater.resolve('3.0.0') self.assertTrue(self.updater.status()['recoveryRequired']) self.assertEqual(self.updater.resolve('3.1.0')['status'], 'recovered') self.assertFalse(self.updater.status()['recoveryRequired']) self.assertEqual(u.read_json(self.root / '.updates/installed.json')['manifest']['version'], '3.1.0') def test_second_managed_update_does_not_need_adoption(self): u.write_json(self.root / '.updates/installed.json', {'composeSha256': u.sha(self.original_compose)}) self.assertEqual(self.updater.execute('apply', '3.1.0')['status'], 'succeeded') def test_downgrade_is_rejected(self): (self.root / '.env').write_bytes(u.set_version(self.original_env, '3.2.0')) with self.assertRaisesRegex(u.UpdateError, 'Downgrades'): self.run_update() self.assertFalse(self.called('stop')) def test_backup_integrity_manifest_is_complete(self): result = self.run_update() backup = Path(result['backup']) manifest = u.read_json(backup / 'complete.json') for name in ('database.dump', 'globals.sql', 'files.tar', 'config/.env', 'config/docker-compose.yml'): self.assertEqual(u.sha((backup / name).read_bytes()), manifest['sha256'][name]) class HttpsTests(unittest.TestCase): def test_token_is_removed_after_cross_origin_redirect(self): requests = [] class Opener: def open(self, request, timeout): requests.append(request) if len(requests) == 1: raise u.urllib.error.HTTPError(request.full_url, 302, 'redirect', {'Location': 'https://objects.example.test/file'}, None) return io.BytesIO(b'asset') with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): self.assertEqual(u.Https('private-token').get(u.REPOSITORY + '/asset'), b'asset') self.assertEqual(requests[0].get_header('Authorization'), 'token private-token') self.assertIsNone(requests[1].get_header('Authorization')) def test_insecure_redirect_is_rejected(self): class Opener: def open(self, request, timeout): raise u.urllib.error.HTTPError(request.full_url, 302, 'redirect', {'Location': 'http://objects.example.test/file'}, None) with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): with self.assertRaisesRegex(u.UpdateError, 'HTTPS'): u.Https().get(u.REPOSITORY + '/asset') def test_download_size_is_limited(self): class Opener: def open(self, request, timeout): return io.BytesIO(b'123456') with patch.object(u.urllib.request, 'build_opener', return_value=Opener()): with self.assertRaisesRegex(u.UpdateError, 'Größe'): u.Https().get(u.REPOSITORY + '/asset', limit=5) if __name__ == '__main__': unittest.main()