76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""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")
|