from __future__ import annotations import contextlib import hashlib import importlib.util import os import re import shutil import subprocess from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] INSTALLER_PATH = ( ROOT / "ops" / "gitea-deploy-broker" / "tune_host_guard_memory_thresholds.py" ) SPEC = importlib.util.spec_from_file_location( "host_guard_memory_threshold_installer_test", INSTALLER_PATH ) assert SPEC and SPEC.loader installer = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(installer) @pytest.fixture def guard_script() -> bytes: return b"""#!/usr/bin/env bash set -euo pipefail readonly MEM_PAUSE_KB=4194304 readonly MEM_RESUME_KB=6291456 readonly SWAP_CRITICAL_MEMORY_KB=8388608 readonly SWAP_PAUSE_KBPS=1024 readonly SWAP_RESUME_KBPS=256 readonly SWAP_CRITICAL_SAMPLES=3 if [ \"$mem_kb\" -lt \"$MEM_PAUSE_KB\" ]; then runner_critical=1 fi """ def _baseline(monkeypatch, payload: bytes) -> None: monkeypatch.setattr(installer, "BASELINE_SHA256", hashlib.sha256(payload).hexdigest()) def _assignment_values(payload: bytes) -> dict[str, int]: rows = re.findall(rb"(?m)^readonly ([A-Z0-9_]+)=([0-9]+)$", payload) return {name.decode("ascii"): int(value) for name, value in rows} def _fixture_install(monkeypatch, tmp_path, guard_script, *, fsync_fails=False): target = tmp_path / "script" backup = tmp_path / ".script.backup" lock = tmp_path / ".lock" target.write_bytes(guard_script) _baseline(monkeypatch, guard_script) monkeypatch.setattr(installer, "TARGET_PATH", target) monkeypatch.setattr(installer, "BACKUP_PATH", backup) monkeypatch.setattr(installer, "LOCK_PATH", lock) monkeypatch.setattr(installer, "_assert_fixed_environment", lambda: target.stat()) monkeypatch.setattr(installer, "_exclusive_lock", contextlib.nullcontext) monkeypatch.setattr( installer, "_read_private_backup", lambda: backup.read_bytes() if backup.exists() else None, ) events: list[str] = [] def assert_private_file(path, *, allow_missing=False): if path == backup: events.append("backup_identity") elif path == target: events.append("target_identity") return path.lstat() if path.exists() else None def write_backup(path, payload): events.append("backup_create") with path.open("xb") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) def write_stage(payload): stage = tmp_path / f".stage-{len(events)}" stage.write_bytes(payload) return stage def fsync_directory(_path): events.append("directory_fsync") if fsync_fails: raise OSError("fixture directory fsync failure") def atomic_replace(stage, destination): events.append("replace") os.replace(stage, destination) events.append("post_replace_fsync") original_read_bytes = Path.read_bytes def traced_read_bytes(path): payload = original_read_bytes(path) if path == backup: events.append("backup_readback") return payload monkeypatch.setattr(installer, "_assert_private_regular_file", assert_private_file) monkeypatch.setattr(installer, "_write_private_file_exclusive", write_backup) monkeypatch.setattr(installer, "_write_stage", write_stage) monkeypatch.setattr(installer, "_bash_syntax_check", lambda _path: None) monkeypatch.setattr(installer, "_fsync_directory", fsync_directory) monkeypatch.setattr(installer, "_atomic_replace", atomic_replace) monkeypatch.setattr(Path, "read_bytes", traced_read_bytes) return target, backup, events def test_fixture_patch_changes_only_reviewed_memory_literals(guard_script, monkeypatch): _baseline(monkeypatch, guard_script) candidate = installer._render_candidate(guard_script) before = _assignment_values(guard_script) after = _assignment_values(candidate) assert after == { **before, "MEM_PAUSE_KB": 2_097_152, "MEM_RESUME_KB": 4_194_304, "SWAP_CRITICAL_MEMORY_KB": 4_194_304, } changed = [ (old.decode("ascii"), new.decode("ascii")) for old, new in zip(guard_script.splitlines(), candidate.splitlines()) if old != new ] assert changed == [ ("readonly MEM_PAUSE_KB=4194304", "readonly MEM_PAUSE_KB=2097152"), ("readonly MEM_RESUME_KB=6291456", "readonly MEM_RESUME_KB=4194304"), ( "readonly SWAP_CRITICAL_MEMORY_KB=8388608", "readonly SWAP_CRITICAL_MEMORY_KB=4194304", ), ] def test_candidate_passes_real_bash_syntax(guard_script, monkeypatch, tmp_path): bash = shutil.which("bash") if not bash: pytest.skip("bash is required for this fixture syntax check") _baseline(monkeypatch, guard_script) candidate = installer._render_candidate(guard_script) fixture = tmp_path / "Host_Resource_Guard.script" fixture.write_bytes(candidate) fixture_argument = str(fixture) if os.name == "nt": # Windows bash.exe here is the WSL launcher; translate C:\... to /mnt/c/... drive = fixture.drive.rstrip(":").lower() fixture_argument = f"/mnt/{drive}/{fixture.as_posix()[3:]}" completed = subprocess.run( [bash, "-n", fixture_argument], check=False, capture_output=True, timeout=15 ) assert completed.returncode == 0, completed.stderr.decode("utf-8", errors="replace") def test_baseline_fingerprint_mismatch_refuses_candidate(guard_script, monkeypatch): _baseline(monkeypatch, guard_script) changed = guard_script + b"# unreviewed source change\n" with pytest.raises(installer.InstallError, match="baseline_fingerprint_mismatch"): installer._render_candidate(changed) def test_duplicate_or_changed_memory_literal_refuses_candidate(guard_script, monkeypatch): duplicate = guard_script + b"readonly MEM_PAUSE_KB=4194304\n" _baseline(monkeypatch, duplicate) with pytest.raises(installer.InstallError, match="threshold_shape_invalid:MEM_PAUSE_KB"): installer._render_candidate(duplicate) wrong_rate = guard_script.replace( b"readonly SWAP_RESUME_KBPS=256", b"readonly SWAP_RESUME_KBPS=512" ) _baseline(monkeypatch, wrong_rate) with pytest.raises(installer.InstallError, match="threshold_shape_invalid:SWAP_RESUME_KBPS"): installer._render_candidate(wrong_rate) def test_backup_is_readback_verified_and_directory_synced_before_replace( guard_script, monkeypatch, tmp_path ): target, backup, events = _fixture_install( monkeypatch, tmp_path, guard_script ) status, after_sha = installer.install() assert status == "applied" assert hashlib.sha256(target.read_bytes()).hexdigest() == after_sha assert backup.read_bytes() == guard_script replace_at = events.index("replace") assert events.index("backup_create") < events.index("backup_readback") assert events.index("backup_readback") < events.index("directory_fsync") assert events.index("directory_fsync") < max( index for index, event in enumerate(events[:replace_at]) if event == "target_identity" ) < replace_at assert events.index("post_replace_fsync") > replace_at def test_backup_directory_sync_failure_leaves_target_unmodified( guard_script, monkeypatch, tmp_path ): target, backup, events = _fixture_install( monkeypatch, tmp_path, guard_script, fsync_fails=True ) with pytest.raises(installer.InstallError, match="backup_directory_sync_failed"): installer.install() assert target.read_bytes() == guard_script assert backup.read_bytes() == guard_script assert "replace" not in events