import copy import importlib.util import json from pathlib import Path import tempfile import unittest from unittest.mock import Mock spec = importlib.util.spec_from_file_location("storage_migration", Path(__file__).resolve().parents[2] / "selfhost-storage-migrate.py") m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) class HostSafetyTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) for name in ("minio", "postgres"): (self.root / name).mkdir() self.config = {"services": {n: {} for n in ("db", "minio", "createbuckets", "backend", "frontend", "traefik", "node-exporter")}} for name, path, target in (("minio", "minio", "/data"), ("db", "postgres", "/var/lib/postgresql/data")): self.config["services"][name]["volumes"] = [{"type": "bind", "source": str(self.root / path), "target": target}] self.config["services"]["minio"].update(networks={"internal": {}}, environment={"MINIO_ROOT_USER": "user", "MINIO_ROOT_PASSWORD": "secret"}) self.config["services"]["db"]["environment"] = {"POSTGRES_USER": "u", "POSTGRES_PASSWORD": "p", "POSTGRES_DB": "db"} self.config["services"]["backend"]["environment"] = {"DATABASE_URL": "postgres://u:p@db:5432/db", "S3_ENDPOINT": "http://minio:9000", "S3_REGION": "eu-central-1", "S3_BUCKET": "fedeo", "S3_ACCESS_KEY": "user", "S3_SECRET_KEY": "secret"} def test_rejects_external_or_unbacked_storage_and_secret_drift(self): m.validate_config(self.config, self.root) mutations = [lambda c: c["services"]["minio"].update(ports=["9000:9000"]), lambda c: c["services"]["backend"]["environment"].update(S3_ENDPOINT="https://external.example"), lambda c: c["services"]["backend"]["environment"].update(S3_SECRET_KEY="different"), lambda c: c["services"]["minio"]["volumes"][0].update(source="/elsewhere"), lambda c: c["services"].update(unknown={}), lambda c: c["services"]["backend"]["environment"].update(DATABASE_URL="postgres://u:p@db:5432/other")] for mutate in mutations: config = copy.deepcopy(self.config) mutate(config) with self.assertRaises(m.Error): m.validate_config(config, self.root) def test_rejects_symlink_source(self): (self.root / "minio").rmdir() (self.root / "minio").symlink_to(self.root / "postgres") with self.assertRaises(m.Error): m.validate_config(self.config, self.root) def test_retired_matrix_remains_stopped_and_outside_compose(self): container = {"Config": {"Labels": {"com.docker.compose.service": "matrix-redis"}}, "State": {"Running": False}, "HostConfig": {"RestartPolicy": {"Name": "no"}}, "Mounts": [{"Type": "volume", "Source": "/var/lib/docker/volumes/old/_data", "Destination": "/data"}]} indexed, retained = m.index_containers(self.config, [container]) self.assertEqual(set(indexed), {"matrix-redis"}) self.assertEqual(retained, {"matrix-redis": {"volumes": []}}) self.assertNotIn("matrix-redis", self.config["services"]) mutations = [lambda c: c["State"].update(Running=True), lambda c: c["State"].update(Restarting=True), lambda c: c["State"].update(Paused=True), lambda c: c["HostConfig"]["RestartPolicy"].update(Name="unless-stopped"), lambda c: c["HostConfig"].update(AutoRemove=True), lambda c: c["Config"]["Labels"].update({"com.docker.compose.service": "unknown"}), lambda c: c["Config"]["Labels"].update({"com.docker.compose.oneoff": "True"}), lambda c: c["Mounts"][0].update(Destination="/unexpected")] for mutate in mutations: changed = copy.deepcopy(container) mutate(changed) with self.assertRaises(m.Error): m.index_containers(self.config, [changed]) with self.assertRaises(m.Error): m.index_containers(self.config, [container, container]) def test_retired_matrix_bind_mounts_keep_strict_backup_validation(self): container = {"Config": {"Labels": {"com.docker.compose.service": "matrix-db"}}, "State": {"Running": False}, "HostConfig": {"RestartPolicy": {"Name": "no"}}, "Mounts": [{"Type": "bind", "Source": "/outside/postgres", "Destination": "/var/lib/postgresql/data"}]} migration = m.Migration(self.root) _, retained = m.index_containers(self.config, [container]) migration.engine.current = {"services": retained} migration.engine.retired_services = list(retained) with self.assertRaises(m.Error): migration.engine.validate_legacy_matrix() (self.root / "matrix/postgres").mkdir(parents=True) retained["matrix-db"]["volumes"][0]["source"] = str(self.root / "matrix/postgres") migration.engine.validate_legacy_matrix() self.assertEqual(migration.engine.matrix_backup_paths, ["matrix"]) def test_cold_index_detects_objects_hidden_by_minio_listing(self): source = self.root / "minio" (source / ".minio.sys").mkdir() (source / ".minio.sys/format.json").write_text(json.dumps({"format": "xl-single", "xl": {"version": "3"}})) (source / "fedeo/a").mkdir(parents=True) (source / "fedeo/a/xl.meta").write_bytes(b"opaque") m.validate_minio_listing(self.root, "fedeo", {"objects": [{"Key": "a"}]}, exact=True) (source / "fedeo/a/b").mkdir() (source / "fedeo/a/b/xl.meta").write_bytes(b"hidden-from-S3-LIST") with self.assertRaisesRegex(m.Error, "Konflikt|konflikt"): m.validate_minio_listing(self.root, "fedeo", {"objects": [{"Key": "a"}]}, exact=True) (source / "fedeo/a/b/xl.meta").unlink() with self.assertRaisesRegex(m.Error, "weichen ab"): m.validate_minio_listing(self.root, "fedeo", {"objects": [{"Key": "a"}, {"Key": "missing"}]}, exact=True) def test_update_and_migration_share_exclusive_lock(self): a, b = m.Migration(self.root), m.Migration(self.root) with a.engine.lock(): with self.assertRaises(m.Error): with b.engine.lock(): self.fail("lock must be exclusive") def test_failed_freeze_keeps_persistent_recovery_guard(self): migration = m.Migration(self.root) migration.state = {"startedAt": "test", "writesReleased": False, "containers": {"backend": {"id": "backend"}}} def fail(*args, **kwargs): state = m.u.read_json(migration.engine.home / "status.json") self.assertTrue(state["recoveryRequired"]) self.assertEqual(m.u.read_json(migration.home / "state.json")["phase"], "freezing") raise m.Error("power loss") migration.docker = fail with self.assertRaises(m.Error): migration.freeze() self.assertFalse(m.u.read_json(migration.home / "state.json")["writesReleased"]) def test_backup_corruption_prevents_progress(self): migration = m.Migration(self.root) backup = migration.home / "backup" backup.mkdir() (backup / "files.tar").write_bytes(b"valid") m.u.write_json(backup / "complete.json", {"sha256": {"files.tar": m.sha_file(backup / "files.tar")}}) migration.check_backup() (backup / "files.tar").write_bytes(b"corrupt") with self.assertRaises(m.Error): migration.check_backup() def test_start_failure_is_already_past_rollback_boundary(self): migration = m.Migration(self.root) migration.state = {"startedAt": "test", "phase": "verified", "network": "internal", "writesReleased": False, "containers": {"minio": {"id": "old"}}} for method in ("assert_frozen", "verify_original", "check_backup", "cleanup_helper", "helper", "docker", "compose"): setattr(migration, method, Mock()) migration.inspect = lambda _: {"NetworkSettings": {"Networks": {}}} migration.restore_test = Mock() migration.runner = Mock() (migration.home / "target.json").write_text('{}') def fail_start(): self.assertTrue(m.u.read_json(migration.home / "state.json")["writesReleased"]) raise m.Error("backend did not start") migration.start_application = fail_start from unittest.mock import patch (migration.home / "credentials.json").write_text('{"bucket":"fedeo"}') with patch.object(m, "validate_minio_listing"): (migration.home / "copy").mkdir() (migration.home / "copy/inventory.json").write_text('{}') with self.assertRaises(m.Error): migration.activate() with self.assertRaises(m.Error): migration.rollback() def test_compose_roundtrip_preserves_dollar_signs(self): value = {"environment": {"KEY": "literal-$-$$-${NOT_AN_ENV_VAR}"}, "command": ["echo $VALUE"]} self.assertEqual(m.resolved(m.literal(value)), value) def restart_fixture(self): migration = m.Migration(self.root) (self.root / "docker-compose.yml").write_text('{}') (migration.home / "target.json").write_text('{}') migration.state = {"writesReleased": True, "phase": "released", "containers": { "backend": {"running": True, "health": "unhealthy"}, "db": {"running": True, "health": "healthy"}, "matrix-synapse": {"running": True, "health": "healthy"}, "matrix-element": {"running": True, "health": "unhealthy"}, "matrix-redis": {"running": False, "health": None}, }} migration.compose = Mock(return_value="matrix-element-id") migration.inspect = Mock(return_value={"State": {"Running": True}}) return migration def test_restart_preserves_existing_matrix_health_failure_but_waits_for_core(self): migration = self.restart_fixture() migration.start_application() calls = [call.args for call in migration.compose.call_args_list] start = next(args for args in calls if "up" in args and "--no-deps" not in args) self.assertIn("matrix-element", start) self.assertNotIn("matrix-redis", start) wait = next(args for args in calls if "--wait" in args) for name in ("backend", "db", "matrix-synapse"): self.assertIn(name, wait) self.assertNotIn("matrix-element", wait) self.assertEqual(migration.state["preexistingUnhealthyServices"], ["matrix-element"]) self.assertEqual(migration.state["phase"], "complete") def test_preexisting_unhealthy_matrix_must_still_restart(self): migration = self.restart_fixture() migration.inspect.return_value = {"State": {"Running": False}} with self.assertRaisesRegex(m.Error, "nicht wieder gestartet"): migration.start_application() self.assertEqual(migration.state["phase"], "released") self.assertFalse(any("traefik" in call.args for call in migration.compose.call_args_list)) if __name__ == "__main__": unittest.main()