fix(recovery): orchestrate 110 harbor local repair

scripts/reboot-recovery/tests/test_recover_110_control_path_and_harbor_local.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import os
+import stat
+import subprocess
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+RECOVERY = ROOT / "scripts/reboot-recovery/recover-110-control-path-and-harbor-local.sh"
+
+
+def test_recover_110_orchestrator_contracts() -> None:
+    text = RECOVERY.read_text(encoding="utf-8")
+
+    assert "--check" in text
+    assert "--apply-ssh-metadata" in text
+    assert "--repair-harbor-once" in text
+    assert "--apply-all" in text
+    assert "operation_boundary_secret_value_read=false" in text
+    assert "operation_boundary_host_reboot_performed=false" in text
+    assert "operation_boundary_docker_daemon_restart_performed=false" in text
+    assert "repair-110-ssh-publickey-auth-local.sh" in text
+    assert "harbor-watchdog.sh" in text
+    assert "cat \"$home_dir/.ssh/authorized_keys\"" not in text
+
+    forbidden = [
+        "systemctl restart docker",
+        "service docker restart",
+        "\nreboot",
+        "\nsudo reboot",
+        "\nshutdown",
+        "\nsudo shutdown",
+        "docker system prune",
+        "docker volume rm",
+    ]
+    for pattern in forbidden:
+        assert pattern not in text
+
+
+def test_recover_110_check_uses_fake_helpers_without_writes(tmp_path: Path) -> None:
+    ssh_helper = tmp_path / "ssh-helper.sh"
+    harbor_helper = tmp_path / "harbor-helper.sh"
+    ssh_helper.write_text(
+        "#!/usr/bin/env bash\n"
+        "echo SSH_HELPER_MODE=$1\n"
+        "echo SSH_METADATA_WRITE=false\n",
+        encoding="utf-8",
+    )
+    harbor_helper.write_text(
+        "#!/usr/bin/env bash\n"
+        "echo HARBOR_HELPER_MODE=$1\n"
+        "echo HARBOR_RUNTIME_WRITE=false\n",
+        encoding="utf-8",
+    )
+    for helper in (ssh_helper, harbor_helper):
+        helper.chmod(helper.stat().st_mode | stat.S_IXUSR)
+
+    env = {
+        **os.environ,
+        "ALLOW_NON_110": "1",
+        "AWOOOI_110_SSH_REPAIR_SCRIPT": str(ssh_helper),
+        "AWOOOI_HARBOR_WATCHDOG_SCRIPT": str(harbor_helper),
+    }
+    result = subprocess.run(
+        ["bash", str(RECOVERY), "--check"],
+        check=False,
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert "AWOOOI_110_CONTROL_PATH_AND_HARBOR_LOCAL_RECOVERY mode=check" in result.stdout
+    assert "SSH_HELPER_MODE=--check" in result.stdout
+    assert "HARBOR_HELPER_MODE=--check" in result.stdout
+    assert "SSH_METADATA_WRITE=false" in result.stdout
+    assert "HARBOR_RUNTIME_WRITE=false" in result.stdout
+    assert "operation_boundary_secret_value_read=false" in result.stdout