fix(reboot): reject unknown uptime as fresh boot
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 2m5s
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
Your Name
2026-07-03 00:45:28 +08:00
parent b97bc6f35e
commit 85d8eeb0db
4 changed files with 100 additions and 8 deletions

View File

@@ -45,6 +45,13 @@ def int_value(value: Any, default: int = -1) -> int:
return default
def known_boot_id(value: Any) -> str:
boot_id = str(value or "")
if boot_id in {"", "unknown", "reachable_unknown_boot"}:
return ""
return boot_id
def parse_host_probe(text: str) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for raw_line in text.splitlines():
@@ -114,19 +121,22 @@ def build_payload(args: argparse.Namespace) -> dict[str, Any]:
if not current["reachable"]:
unreachable_hosts.append(alias)
previous_boot_id = (
str(previous_host.get("boot_id"))
if isinstance(previous_host, dict) and previous_host.get("boot_id")
known_boot_id(previous_host.get("boot_id"))
if isinstance(previous_host, dict)
else ""
)
current_boot_id = str(current.get("boot_id") or "")
current_boot_id = known_boot_id(current.get("boot_id"))
boot_id_changed = bool(
previous_boot_id
and previous_boot_id != "unknown"
and current_boot_id
and current_boot_id != "unknown"
and previous_boot_id != current_boot_id
)
fresh_boot = bool(current.get("reachable") and int_value(current.get("uptime_seconds")) <= target_seconds)
uptime_seconds = int_value(current.get("uptime_seconds"))
fresh_boot = bool(
current.get("reachable")
and uptime_seconds >= 0
and uptime_seconds <= target_seconds
)
if boot_id_changed:
changed_boot_id_hosts.append(alias)
if fresh_boot:
@@ -142,7 +152,7 @@ def build_payload(args: argparse.Namespace) -> dict[str, Any]:
"uptime_seconds": current.get("uptime_seconds"),
"deadline_at": (
observed_at
+ timedelta(seconds=max(0, target_seconds - int_value(current.get("uptime_seconds"), 0)))
+ timedelta(seconds=max(0, target_seconds - uptime_seconds))
).isoformat(timespec="seconds"),
}
)

View File

@@ -110,3 +110,63 @@ def test_reboot_detector_fails_visible_when_windows_or_vm_host_missing(tmp_path:
assert "99" in payload["missing_hosts"]
assert payload["all_required_hosts_observed"] is False
assert payload["all_required_hosts_in_reboot_window"] is False
def test_reboot_detector_does_not_treat_unknown_uptime_as_fresh_boot(
tmp_path: Path,
) -> None:
probe_path = tmp_path / "host-probe.txt"
state_path = tmp_path / "state.json"
output_path = tmp_path / "event.json"
probe_path.write_text(
"\n".join(
[
"AWOOOI_REBOOT_AUTO_RECOVERY_HOST_PROBE=1",
"TARGET_HOSTS=99",
(
"HOST_BOOT alias=99 target=192.168.0.99 "
"startup_unit=vmware-host-autostart reachable=1 "
"boot_id=reachable_unknown_boot uptime_seconds=unknown "
"systemd_state=ping_reachable startup_enabled=unknown "
"startup_active=unknown"
),
]
)
+ "\n",
encoding="utf-8",
)
state_path.write_text(
json.dumps({"hosts": {"99": {"boot_id": "win-boot-1"}}}),
encoding="utf-8",
)
subprocess.run(
[
sys.executable,
str(SCRIPT),
"--host-probe-file",
str(probe_path),
"--state-file",
str(state_path),
"--target-minutes",
"10",
"--generated-at",
"2026-06-30T18:00:00+08:00",
"--output",
str(output_path),
"--required-host",
"99",
"--no-write-state",
],
check=True,
)
payload = json.loads(output_path.read_text(encoding="utf-8"))
assert payload["observed_hosts"] == ["99"]
assert payload["reboot_detected"] is False
assert payload["fresh_boot_hosts"] == []
assert payload["rebooted_hosts"] == []
assert payload["all_required_hosts_observed"] is True
assert payload["all_required_hosts_in_reboot_window"] is False
assert payload["state_written"] is False