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_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) if __name__ == "__main__": unittest.main()