Initial curated Unraid-AppOps-Gateway source from 3bd6f6fee310125bfbb4d4650389a817f498724d

backend/tests/test_repopulse_maintenance_environment.py

@@ -0,0 +1,75 @@
+"""Verify that the legacy maintenance stager preserves existing environment bytes."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+SCRIPT = (
+    Path(__file__).resolve().parents[2]
+    / "ops"
+    / "gitea-deploy-broker"
+    / "stage_repopulse_maintenance_environment.py"
+)
+
+
+def _load_stage():
+    sys.path.insert(0, str(SCRIPT.parent))
+    try:
+        spec = importlib.util.spec_from_file_location("repopulse_maintenance_stage_test", SCRIPT)
+        assert spec is not None and spec.loader is not None
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        sys.path.pop(0)
+
+
+@pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="root-owned Linux files required")
+def test_mixed_legacy_line_endings_are_backed_up_byte_for_byte_and_staged_idempotently(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    stage = _load_stage()
+    legacy_env = tmp_path / "deploy.env"
+    managed_env = tmp_path / "deploy.managed.env"
+    backup_root = tmp_path / "environment-backup"
+    original = (
+        b"REPOPULSE_GITEA_URL=http://198.18.10.150:3000\n"
+        b"REPOPULSE_GITEA_TOKEN=validation-only-token-1234567890\n"
+        b"REPOPULSE_MAX_CONCURRENT_SCANS=1\r\n"
+    )
+    legacy_env.write_bytes(original)
+    legacy_env.chmod(0o600)
+    monkeypatch.setattr(stage, "LEGACY_ENV", legacy_env)
+    monkeypatch.setattr(stage, "MANAGED_ENV", managed_env)
+    monkeypatch.setattr(stage, "BACKUP_ROOT", backup_root)
+    monkeypatch.setattr(stage, "BACKUP_FILE", backup_root / "legacy-deploy.env")
+    monkeypatch.setattr(stage, "BACKUP_SHA_FILE", backup_root / "legacy-deploy.env.sha256")
+
+    first = stage.prepare()
+    assert first == {"status": "prepared", "legacyBackupSha256": hashlib.sha256(original).hexdigest()}
+    assert stage.BACKUP_FILE.read_bytes() == original
+    assert stage.BACKUP_SHA_FILE.read_text() == hashlib.sha256(original).hexdigest() + "\n"
+    staged_legacy = legacy_env.read_bytes()
+    assert staged_legacy.startswith(original)
+    assert staged_legacy[len(original) :].endswith(b"\r\n")
+    legacy_values = stage._parse_environment(staged_legacy)
+    managed_values = stage._parse_environment(managed_env.read_bytes())
+    assert legacy_values["REPOPULSE_GITEA_URL"] == "http://198.18.10.150:3000"
+    assert managed_values["REPOPULSE_GITEA_URL"] == "https://gitea.example.com"
+    assert managed_values["REPOPULSE_MAINTENANCE_TOKEN"] == legacy_values["REPOPULSE_MAINTENANCE_TOKEN"]
+
+    assert stage.prepare() == first
+    assert legacy_env.read_bytes() == staged_legacy
+    assert stage.BACKUP_FILE.read_bytes() == original
+
+
+def test_bare_carriage_return_is_rejected() -> None:
+    stage = _load_stage()
+    with pytest.raises(ValueError):
+        stage._parse_environment(b"REPOPULSE_GITEA_URL=https://gitea.example.com\rREPOPULSE_GITEA_TOKEN=x\n")