fix(reboot): align windows99 vmx package cli with scorecard
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m10s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-07-03 11:04:53 +08:00
parent 44a308d4e5
commit b565e0aa5f
3 changed files with 46 additions and 147 deletions

View File

@@ -12,75 +12,28 @@ import argparse
import json
import sys
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
_REPO_ROOT = Path(__file__).resolve().parents[2]
_API_ROOT = _REPO_ROOT / "apps" / "api"
if str(_API_ROOT) not in sys.path:
sys.path.insert(0, str(_API_ROOT))
from src.services.reboot_auto_recovery_slo_scorecard import ( # noqa: E402
_build_windows99_vmx_source_repair_package,
_taipei_now_iso,
)
DEFAULT_SCORECARD_URL = (
"https://awoooi.wooo.work/api/v1/agents/reboot-auto-recovery-slo-scorecard"
)
DEFAULT_REQUIRED_ALIASES = ["111", "188", "120", "121", "112"]
EXPECTED_VMX_CANDIDATES = {
"111": [
r"D:\Documents\Virtual Machines\192.168.0.111_Ubuntu_64-bit\192.168.0.111_Ubuntu_64-bit.vmx",
],
"188": [
r"D:\Documents\Virtual Machines\Ollama_Ubuntu_64-bit\Ollama_Ubuntu_64-bit.vmx",
],
"120": [
r"D:\Documents\Virtual Machines\192.168.0.120_Ubuntu_64-bit\192.168.0.120_Ubuntu_64-bit.vmx",
],
"121": [
r"D:\Documents\Virtual Machines\192.168.0.121_Ubuntu_64-bit\192.168.0.121_Ubuntu_64-bit.vmx",
],
"112": [
r"D:\Downloads\kali-linux-2025.4-vmware-amd64\kali-linux-2025.4-vmware-amd64.vmwarevm\kali-linux-2025.4-vmware-amd64.vmx",
],
}
MISSING_VMX_NEXT_STEP = (
"restore_windows99_missing_vmx_source_for_aliases_then_rerun_no_secret_"
"collector_and_scorecard_no_vm_power_change"
)
POWERED_OFF_NEXT_STEP = (
"build_windows99_vmware_autostart_check_mode_package_for_powered_off_aliases_"
"then_rerun_no_secret_collector_no_vm_power_change"
)
FORBIDDEN_ACTIONS = [
"read_windows_password_or_secret",
"vm_power_change",
"vmrun_start_or_stop",
"host_reboot",
"windows_service_restart",
"windows_registry_apply",
"scheduled_task_register_or_modify",
"docker_nginx_k3s_db_firewall_restart",
]
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _strings(value: Any) -> list[str]:
if isinstance(value, list):
return [str(item).strip() for item in value if str(item).strip()]
if isinstance(value, str):
return [part.strip() for part in value.split(",") if part.strip()]
return []
def _unique(values: list[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for value in values:
if value in seen:
continue
seen.add(value)
result.append(value)
return result
def _load_json_file(path: Path) -> dict[str, Any]:
with path.open(encoding="utf-8") as handle:
payload = json.load(handle)
@@ -97,98 +50,16 @@ def _fetch_json_url(url: str, timeout_seconds: float) -> dict[str, Any]:
return payload if isinstance(payload, dict) else {}
def _taipei_now() -> str:
return datetime.now(ZoneInfo("Asia/Taipei")).replace(microsecond=0).isoformat()
def _powershell_test_path(path: str) -> str:
escaped = path.replace("'", "''")
return f"Test-Path -LiteralPath '{escaped}'"
def build_package(scorecard: dict[str, Any], *, generated_at: str | None = None) -> dict[str, Any]:
windows99 = _dict(scorecard.get("windows99_vmware_autostart"))
readback = _dict(scorecard.get("readback"))
missing_vmx_aliases = _unique(
_strings(windows99.get("missing_vmx_aliases"))
or _strings(readback.get("windows99_missing_vmx_aliases"))
)
powered_off_aliases = _unique(
_strings(windows99.get("powered_off_aliases"))
or _strings(readback.get("windows99_powered_off_aliases"))
)
required_aliases = _unique(
_strings(windows99.get("required_vm_aliases")) or DEFAULT_REQUIRED_ALIASES
)
package_ready = bool(missing_vmx_aliases or powered_off_aliases)
next_step = (
MISSING_VMX_NEXT_STEP
if missing_vmx_aliases
else POWERED_OFF_NEXT_STEP
if powered_off_aliases
else "rerun_no_secret_collector_and_reboot_slo_scorecard_verify_only"
)
existing_package = _dict(scorecard.get("windows99_vmx_source_repair_package"))
if existing_package and generated_at is None:
return existing_package
rows = []
for alias in required_aliases:
candidates = EXPECTED_VMX_CANDIDATES.get(alias, [])
rows.append(
{
"alias": alias,
"missing_vmx_source": alias in missing_vmx_aliases,
"powered_off": alias in powered_off_aliases,
"expected_vmx_candidates": candidates,
"check_mode_probe_commands": [
_powershell_test_path(candidate) for candidate in candidates
],
"controlled_restore_required": alias in missing_vmx_aliases,
"controlled_restore_hint": (
"restore_or_relink_existing_vmx_source_to_expected_path_then_verify"
if alias in missing_vmx_aliases
else ""
),
}
)
return {
"schema_version": "windows99_vmx_source_repair_check_mode_package_v1",
"generated_at": generated_at or _taipei_now(),
"status": (
"check_mode_package_ready_windows99_vmx_source_repair_required"
if missing_vmx_aliases
else "check_mode_package_ready_windows99_power_check_required"
if powered_off_aliases
else "ready_no_windows99_vmx_source_repair_required"
),
"package_ready": package_ready,
"target_host_alias": "99",
"scorecard_status": str(scorecard.get("status") or "unknown"),
"scorecard_generated_at": str(scorecard.get("generated_at") or ""),
"collector_status": str(
readback.get("windows99_no_secret_collector_status") or "unknown"
),
"missing_vmx_aliases": missing_vmx_aliases,
"powered_off_aliases": powered_off_aliases,
"required_vm_aliases": required_aliases,
"vm_rows": rows,
"safe_next_step": next_step,
"controlled_apply_mode": "check_mode_package_only_no_windows_or_vm_write",
"apply_allowed_by_this_package": False,
"post_verifier": [
"bash scripts/reboot-recovery/collect-windows99-vmware-verify.sh --collect",
"bash scripts/reboot-recovery/install-reboot-auto-recovery-slo-110.sh --verify-only",
"GET /api/v1/agents/reboot-auto-recovery-slo-scorecard",
],
"forbidden_actions": FORBIDDEN_ACTIONS,
"operation_boundaries": {
"secret_value_read": False,
"remote_write_performed": False,
"vm_power_change_performed": False,
"windows_service_restart_performed": False,
"windows_registry_apply_performed": False,
"host_reboot_performed": False,
},
}
return _build_windows99_vmx_source_repair_package(
_dict(scorecard.get("windows99_vmware_autostart")),
generated_at=generated_at
or str(scorecard.get("generated_at") or _taipei_now_iso()),
)
def parse_args(argv: list[str]) -> argparse.Namespace:

View File

@@ -81,6 +81,19 @@ def test_package_routes_missing_vmx_to_source_repair_check_mode() -> None:
assert "read_windows_password_or_secret" in payload["forbidden_actions"]
def test_package_prefers_existing_scorecard_package_readback() -> None:
module = _load_module()
scorecard = _scorecard()
scorecard["windows99_vmx_source_repair_package"] = {
"schema_version": "windows99_vmx_source_repair_check_mode_package_v1",
"sentinel": "from_scorecard_api",
}
payload = module.build_package(scorecard)
assert payload["sentinel"] == "from_scorecard_api"
def test_package_routes_power_only_to_autostart_check_mode() -> None:
module = _load_module()
scorecard = _scorecard()