50c7a7f7f5
Some checks failed
CD Pipeline / select-latest-carrier (push) Successful in 1m9s
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 10m26s
CD Pipeline / revalidate-deploy-carrier (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / revalidate-post-deploy-carrier (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
MCP External Protocol Replay / replay-and-verify (push) Successful in 44s
MCP External Artifact Mirror / mirror-and-verify (push) Successful in 33s
E2E Health Check / e2e-health (push) Failing after 48s
AI 技術雷達監控 / ai-technology-watch (push) Successful in 34s
Agent Version Lifecycle Watch / version-lifecycle-watch (push) Successful in 7s
Agent Market Watch / market-watch (push) Failing after 26s
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 23s
616 lines
23 KiB
Python
616 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from copy import deepcopy
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
PREFLIGHT = ROOT / "scripts/reboot-recovery/agent99-live-preflight.ps1"
|
|
|
|
|
|
def _function_source() -> str:
|
|
source = PREFLIGHT.read_text(encoding="utf-8")
|
|
start = source.index("function Get-AgentLivePreflightDecision")
|
|
end = source.index("function Get-TaskReadback", start)
|
|
return source[start:end]
|
|
|
|
|
|
def _evidence_function_source() -> str:
|
|
source = PREFLIGHT.read_text(encoding="utf-8")
|
|
start = source.index("function Get-AgentBoundedNewestFiles")
|
|
end = source.index("$requiredTasks = @(", start)
|
|
return source[start:end]
|
|
|
|
|
|
def _base_case() -> dict[str, Any]:
|
|
return {
|
|
"agentRootPresent": True,
|
|
"manifest": {
|
|
"exists": True,
|
|
"sourceRevision": "a" * 40,
|
|
"runtimeMatched": True,
|
|
"fileCount": 18,
|
|
"mismatchCount": 0,
|
|
"parseError": "",
|
|
},
|
|
"manifestCheckRequired": True,
|
|
"taskFailureCount": 0,
|
|
"relayListenerCount": 1,
|
|
"requiredEvidenceStaleCount": 0,
|
|
"requiredEvidenceReserveFailureCount": 0,
|
|
"requiredEvidenceParseFailureCount": 0,
|
|
"requiredEvidenceContentFailureCount": 0,
|
|
"selfHealthSeverity": "ok",
|
|
"selfHealthPerformanceIssueCount": 0,
|
|
"staleAlertExecutionCount": 0,
|
|
"alertIncomingCount": 0,
|
|
"staleAlertIncomingCount": 0,
|
|
"pendingCommandBlockingCount": 0,
|
|
"promotionDeferredCommandCount": 0,
|
|
}
|
|
|
|
|
|
def _with_overrides(**overrides: Any) -> dict[str, Any]:
|
|
case = deepcopy(_base_case())
|
|
manifest_overrides = overrides.pop("manifest", None)
|
|
case.update(overrides)
|
|
if manifest_overrides:
|
|
case["manifest"].update(manifest_overrides)
|
|
return case
|
|
|
|
|
|
def _as_list(value: Any) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, list):
|
|
return [str(item) for item in value]
|
|
return [str(value)]
|
|
|
|
|
|
def _run_decision(powershell: str, case: dict[str, Any]) -> dict[str, Any]:
|
|
fixture = json.dumps(case, separators=(",", ":"))
|
|
command = f"""
|
|
{_function_source()}
|
|
$case = @'
|
|
{fixture}
|
|
'@ | ConvertFrom-Json
|
|
$parameters = @{{
|
|
AgentRootPresent = [bool]$case.agentRootPresent
|
|
Manifest = $case.manifest
|
|
ManifestCheckRequired = [bool]$case.manifestCheckRequired
|
|
ExpectedRuntimeFileCount = 18
|
|
TaskFailureCount = [int]$case.taskFailureCount
|
|
RelayListenerCount = [int]$case.relayListenerCount
|
|
RequiredEvidenceStaleCount = [int]$case.requiredEvidenceStaleCount
|
|
RequiredEvidenceReserveFailureCount = [int]$case.requiredEvidenceReserveFailureCount
|
|
RequiredEvidenceParseFailureCount = [int]$case.requiredEvidenceParseFailureCount
|
|
RequiredEvidenceContentFailureCount = [int]$case.requiredEvidenceContentFailureCount
|
|
SelfHealthSeverity = [string]$case.selfHealthSeverity
|
|
SelfHealthPerformanceIssueCount = [int]$case.selfHealthPerformanceIssueCount
|
|
StaleAlertExecutionCount = [int]$case.staleAlertExecutionCount
|
|
AlertIncomingCount = [int]$case.alertIncomingCount
|
|
StaleAlertIncomingCount = [int]$case.staleAlertIncomingCount
|
|
PendingCommandBlockingCount = [int]$case.pendingCommandBlockingCount
|
|
PromotionDeferredCommandCount = [int]$case.promotionDeferredCommandCount
|
|
}}
|
|
Get-AgentLivePreflightDecision @parameters |
|
|
ConvertTo-Json -Depth 6 -Compress
|
|
"""
|
|
result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
payload = next(
|
|
line for line in reversed(result.stdout.splitlines()) if line.startswith("{")
|
|
)
|
|
return json.loads(payload)
|
|
|
|
|
|
def test_warning_receipt_is_visible_without_weakening_integrity_blockers() -> None:
|
|
source = PREFLIGHT.read_text(encoding="utf-8")
|
|
decision = _function_source()
|
|
|
|
assert '$warningReasons += "self_health_warning"' in decision
|
|
assert '$warningReasons += "performance_warning"' in decision
|
|
assert '$warningReasons += "fresh_alert_pending_next_inbox_tick"' in decision
|
|
assert (
|
|
'$warningReasons += "maintenance_pending_commands_deferred_until_post_promotion"'
|
|
in decision
|
|
)
|
|
assert '"critical" { $blockingReasons += "self_health_not_ok" }' in decision
|
|
assert 'default { $blockingReasons += "self_health_unclassified" }' in decision
|
|
assert '$blockingReasons += "runtime_manifest_identity_invalid"' in decision
|
|
assert '$blockingReasons += "runtime_manifest_mismatch"' in decision
|
|
assert '$blockingReasons += "scheduled_task_unhealthy"' in decision
|
|
assert '$blockingReasons += "sre_alert_relay_not_listening"' in decision
|
|
assert (
|
|
'$blockingReasons += "required_evidence_freshness_reserve_insufficient"'
|
|
in decision
|
|
)
|
|
assert "deploymentEligible = [bool]($blockingReasons.Count -eq 0)" in decision
|
|
assert "preflightGreen = [bool]$decision.deploymentEligible" in source
|
|
assert "-ExpectedRuntimeFileCount 18" in source
|
|
assert "warningReasons = $warningReasons" in source
|
|
assert "selfHealthPerformanceIssues = @(" in source
|
|
assert 'if ($safeHost -notmatch "^[A-Za-z0-9.:-]{1,128}$")' in source
|
|
assert 'if ($reason -match "^[a-z0-9_]{1,80}$")' in source
|
|
assert (
|
|
"[int]$selfCheckEvidence.safeSummary.selfHealthPerformanceIssueCount"
|
|
in source
|
|
)
|
|
assert 'if ($result.summary.deploymentEligible)' in source
|
|
assert 'if ($selfHealthSeverity -ne "ok")' not in source
|
|
assert '"agent99_background_mode_deferred_v1"' in source
|
|
assert '"agent99_recovery_transport_deferred_v1"' in source
|
|
assert "deferredCandidateCount = [int]$deferredCandidates.Count" in source
|
|
assert "[IO.Directory]::EnumerateFiles(" in source
|
|
assert "[Collections.Generic.SortedSet[string]]::new(" in source
|
|
assert "$null = $selected.Remove($selected.Min)" in source
|
|
assert "candidateWindowExhausted = $candidateWindowExhausted" in source
|
|
assert "function Get-AgentControlLoopMaintenanceReadback" in source
|
|
assert '"agent99-control-loop-maintenance-2*.json"' in source
|
|
assert '$ageMinutes -le $(if ($legacyReceipt) { 360 } else { 720 })' in source
|
|
assert '$legacyPropertySetMatches' in source
|
|
assert '$legacyIdentityFieldsValid' in source
|
|
assert '$receipt.taskEnabled -is [bool] -and $receipt.taskEnabled' in source
|
|
assert '"legacy_v2_exact_upgrade_bridge"' in source
|
|
assert '$receipt.precheckPassed -is [bool]' in source
|
|
assert '$receipt.taskEnabledAfter -is [bool]' in source
|
|
assert '$receipt.runtimeWritePerformed -is [bool]' in source
|
|
assert '$intent.secretValueRead -is [bool]' in source
|
|
assert '$activeControlProcesses.Count -eq 0' in source
|
|
assert '-not $leasePresent' in source
|
|
assert '-not $recoveryMutexActive' in source
|
|
assert '$promotionEvidenceRequired = [bool]($ReadinessScope -eq "FullRuntime")' in source
|
|
assert '$task | Add-Member -NotePropertyName runtimeHealthy' in source
|
|
assert '$task | Add-Member -NotePropertyName promotionReady' in source
|
|
assert '$controlLoopMaintenance.valid' in source
|
|
assert "function Get-AgentPendingCommandPromotionReadback" in source
|
|
assert '"bounded_maintenance_auto_recover_deferred_until_post_promotion"' in source
|
|
assert '"full_runtime_requires_maintenance_auto_recover_to_drain"' in source
|
|
assert '$payload.controlledApply -is [bool] -and $payload.controlledApply' in source
|
|
assert '$payload.observation -is [pscustomobject]' in source
|
|
assert 'instructionContentReadback = $false' in source
|
|
assert '$rootTaskPath = "\\"' in source
|
|
assert "Get-ScheduledTask -TaskPath $TaskPath -TaskName $TaskName" in source
|
|
assert "Get-ScheduledTaskInfo -TaskPath $TaskPath -TaskName $TaskName" in source
|
|
assert "[string[]]$RequiredArgumentMarkers" in source
|
|
assert "$actions.Count -eq 1" in source
|
|
assert "actionDefinitionReady = $actionDefinitionReady" in source
|
|
assert "principalReady = $principalReady" in source
|
|
assert "definitionReady = $definitionReady" in source
|
|
assert '$task.Principal.LogonType -eq "S4U"' in source
|
|
assert '$task.Principal.RunLevel -eq "Highest"' in source
|
|
assert "$task.definitionReady" in source
|
|
|
|
|
|
def test_evidence_selection_skips_deferred_but_not_malformed_candidates(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
powershell = shutil.which("pwsh") or shutil.which("powershell")
|
|
if powershell is None:
|
|
pytest.skip("PowerShell runtime is not installed on this macOS verifier")
|
|
|
|
evidence = tmp_path / "evidence"
|
|
evidence.mkdir()
|
|
canonical = evidence / "agent99-SelfCheck-20260715-200000.json"
|
|
deferred = evidence / "agent99-SelfCheck-20260715-200500.json"
|
|
canonical.write_text(
|
|
json.dumps(
|
|
{
|
|
"mode": "SelfCheck",
|
|
"selfHealth": {
|
|
"severity": "warning",
|
|
"latestPerformance": {"severity": "ok", "hosts": []},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
deferred.write_text(
|
|
json.dumps(
|
|
{
|
|
"schemaVersion": "agent99_background_mode_deferred_v1",
|
|
"mode": "SelfCheck",
|
|
"deferred": True,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
now = canonical.stat().st_mtime
|
|
os.utime(canonical, (now - 10, now - 10))
|
|
os.utime(deferred, (now, now))
|
|
|
|
agent_root = str(tmp_path).replace("'", "''")
|
|
command = f"""
|
|
{_evidence_function_source()}
|
|
$AgentRoot = '{agent_root}'
|
|
Get-EvidenceReadback 'self_check' 'agent99-SelfCheck-*.json' 20 $true |
|
|
ConvertTo-Json -Depth 8 -Compress
|
|
"""
|
|
result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
payload = json.loads(
|
|
next(line for line in reversed(result.stdout.splitlines()) if line.startswith("{"))
|
|
)
|
|
assert payload["file"] == canonical.name
|
|
assert payload["deferredCandidateCount"] == 1
|
|
assert payload["parsed"] is True
|
|
assert payload["safeSummary"]["selfHealthSeverity"] == "warning"
|
|
|
|
malformed = evidence / "agent99-SelfCheck-20260715-201000.json"
|
|
malformed.write_text("{not-json", encoding="utf-8")
|
|
os.utime(malformed, (now + 10, now + 10))
|
|
malformed_result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert malformed_result.returncode == 0, (
|
|
malformed_result.stdout + malformed_result.stderr
|
|
)
|
|
malformed_payload = json.loads(
|
|
next(
|
|
line
|
|
for line in reversed(malformed_result.stdout.splitlines())
|
|
if line.startswith("{")
|
|
)
|
|
)
|
|
assert malformed_payload["file"] == malformed.name
|
|
assert malformed_payload["parsed"] is False
|
|
assert malformed_payload["healthy"] is False
|
|
|
|
|
|
def test_evidence_reserve_rejects_fresh_but_expiring_evidence_without_extending_slo(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
powershell = shutil.which("pwsh") or shutil.which("powershell")
|
|
if powershell is None:
|
|
pytest.skip("PowerShell runtime is not installed on this macOS verifier")
|
|
|
|
evidence = tmp_path / "evidence"
|
|
evidence.mkdir()
|
|
candidate = evidence / "agent99-Status-20260716-230000.json"
|
|
candidate.write_text(json.dumps({"mode": "Status"}), encoding="utf-8")
|
|
six_minutes_ago = candidate.stat().st_mtime - (6 * 60)
|
|
os.utime(candidate, (six_minutes_ago, six_minutes_ago))
|
|
|
|
agent_root = str(tmp_path).replace("'", "''")
|
|
command = f"""
|
|
{_evidence_function_source()}
|
|
$AgentRoot = '{agent_root}'
|
|
Get-EvidenceReadback 'status' 'agent99-Status-*.json' 20 $true 15 |
|
|
ConvertTo-Json -Depth 8 -Compress
|
|
"""
|
|
result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
payload = json.loads(
|
|
next(line for line in reversed(result.stdout.splitlines()) if line.startswith("{"))
|
|
)
|
|
assert payload["fresh"] is True
|
|
assert payload["maxAgeMinutes"] == 20
|
|
assert payload["minimumFreshnessReserveMinutes"] == 15
|
|
assert payload["freshnessReserveReady"] is False
|
|
assert payload["usable"] is False
|
|
assert payload["healthy"] is False
|
|
|
|
|
|
def test_evidence_candidate_window_fails_closed_when_recent_files_are_all_deferred(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
powershell = shutil.which("pwsh") or shutil.which("powershell")
|
|
if powershell is None:
|
|
pytest.skip("PowerShell runtime is not installed on this macOS verifier")
|
|
|
|
evidence = tmp_path / "evidence"
|
|
evidence.mkdir()
|
|
canonical = evidence / "agent99-SelfCheck-20260715-190000.json"
|
|
canonical.write_text(
|
|
json.dumps(
|
|
{
|
|
"mode": "SelfCheck",
|
|
"selfHealth": {"severity": "ok"},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
base_time = canonical.stat().st_mtime
|
|
os.utime(canonical, (base_time - 1000, base_time - 1000))
|
|
for index in range(40):
|
|
deferred = evidence / f"agent99-SelfCheck-20260715-20{index:04d}.json"
|
|
deferred.write_text(
|
|
json.dumps(
|
|
{
|
|
"schemaVersion": "agent99_background_mode_deferred_v1",
|
|
"deferred": True,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
candidate_time = base_time + index
|
|
os.utime(deferred, (candidate_time, candidate_time))
|
|
|
|
agent_root = str(tmp_path).replace("'", "''")
|
|
command = f"""
|
|
{_evidence_function_source()}
|
|
$AgentRoot = '{agent_root}'
|
|
Get-EvidenceReadback 'self_check' 'agent99-SelfCheck-*.json' 20 $true |
|
|
ConvertTo-Json -Depth 8 -Compress
|
|
"""
|
|
result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
payload = json.loads(
|
|
next(line for line in reversed(result.stdout.splitlines()) if line.startswith("{"))
|
|
)
|
|
assert payload["enumeratedCandidateCount"] == 41
|
|
assert payload["candidateWindowLimit"] == 32
|
|
assert payload["candidateWindowTruncated"] is True
|
|
assert payload["candidateWindowExhausted"] is True
|
|
assert payload["deferredCandidateCount"] == 32
|
|
assert payload["parseError"] == "EvidenceCandidateWindowExhausted"
|
|
assert payload["healthy"] is False
|
|
|
|
|
|
def test_promotion_reserve_defers_only_exact_maintenance_auto_recover(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
powershell = shutil.which("pwsh") or shutil.which("powershell")
|
|
if powershell is None:
|
|
pytest.skip("PowerShell runtime is not installed on this macOS verifier")
|
|
|
|
queue = tmp_path / "queue"
|
|
queue.mkdir()
|
|
now = datetime.now(timezone.utc)
|
|
incident_id = "agent99-auto-" + ("a" * 20)
|
|
command_id = "auto-recover-20260719-152746-39ca4a72"
|
|
command_path = queue / f"{command_id}.json"
|
|
payload = {
|
|
"approvalId": "",
|
|
"automationRunId": "096c8173-96b9-5659-94d3-5679c3fb770f",
|
|
"canonicalDigest": "b" * 64,
|
|
"controlledApply": True,
|
|
"correlationKey": "agent99-controlled:" + ("c" * 64),
|
|
"createdAt": now.isoformat(),
|
|
"dispatchIdentitySchemaVersion": "agent99_controlled_dispatch_identity_v1",
|
|
"dispatchRouteId": "agent99_local_observer",
|
|
"executionGeneration": "1",
|
|
"externalId": "host_unreachable_detected",
|
|
"id": command_id,
|
|
"idempotencyKey": "agent99-controlled:" + ("c" * 64),
|
|
"incidentId": incident_id,
|
|
"instruction": "recover exact observed unreachable host through controlled playbook",
|
|
"lockOwner": "096c8173-96b9-5659-94d3-5679c3fb770f",
|
|
"mode": "Recover",
|
|
"observation": {},
|
|
"projectId": "awoooi",
|
|
"reason": "host_unreachable_detected",
|
|
"requestedBy": "Agent99",
|
|
"singleFlightKey": "agent99-controlled-flight:" + ("d" * 64),
|
|
"source": "agent99-recovery-observer",
|
|
"sourceFingerprint": "e" * 64,
|
|
"traceId": "00-" + ("f" * 32) + "-" + ("1" * 16) + "-01",
|
|
"workItemId": f"agent99-auto:{incident_id}:host_unreachable_detected",
|
|
}
|
|
|
|
function_source = _evidence_function_source()
|
|
agent_root = str(tmp_path).replace("'", "''")
|
|
maintenance_timestamp = (now - timedelta(minutes=10)).isoformat()
|
|
|
|
def run(scope: str) -> dict[str, Any]:
|
|
command = f"""
|
|
{function_source}
|
|
$AgentRoot = '{agent_root}'
|
|
$maintenance = [pscustomobject]@{{
|
|
valid = $true
|
|
receiptFieldsValid = $true
|
|
intentFieldsValid = $true
|
|
receiptTimestamp = '{maintenance_timestamp}'
|
|
}}
|
|
Get-AgentPendingCommandPromotionReadback -MaintenanceReadback $maintenance -Scope {scope} |
|
|
ConvertTo-Json -Depth 8 -Compress
|
|
"""
|
|
result = subprocess.run(
|
|
[powershell, "-NoProfile", "-NonInteractive", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
return json.loads(
|
|
next(
|
|
line
|
|
for line in reversed(result.stdout.splitlines())
|
|
if line.startswith("{")
|
|
)
|
|
)
|
|
|
|
command_path.write_bytes(b"\xef\xbb\xbf" + json.dumps(payload).encode("utf-8"))
|
|
reserve = run("PromotionReserve")
|
|
assert reserve["observedCount"] == 1
|
|
assert reserve["deferredCount"] == 1
|
|
assert reserve["blockingCount"] == 0
|
|
assert reserve["contractValid"] is True
|
|
assert reserve["items"][0]["instructionContentReadback"] is False
|
|
assert reserve["items"][0]["identity"].startswith(f"{command_id}|")
|
|
|
|
full_runtime = run("FullRuntime")
|
|
assert full_runtime["deferredCount"] == 0
|
|
assert full_runtime["blockingCount"] == 1
|
|
assert full_runtime["eligibleMaintenanceAutoRecoverCount"] == 1
|
|
assert full_runtime["reason"] == (
|
|
"full_runtime_requires_maintenance_auto_recover_to_drain"
|
|
)
|
|
|
|
payload["mode"] = "Status"
|
|
command_path.write_text(json.dumps(payload), encoding="utf-8")
|
|
invalid = run("PromotionReserve")
|
|
assert invalid["deferredCount"] == 0
|
|
assert invalid["blockingCount"] == 1
|
|
assert invalid["contractValid"] is False
|
|
|
|
|
|
def test_decision_behavior_when_powershell_is_available() -> None:
|
|
powershell = shutil.which("pwsh") or shutil.which("powershell")
|
|
if powershell is None:
|
|
pytest.skip("PowerShell runtime is not installed on this macOS verifier")
|
|
|
|
cases = (
|
|
(_base_case(), True, (), ()),
|
|
(
|
|
_with_overrides(
|
|
selfHealthSeverity="warning",
|
|
selfHealthPerformanceIssueCount=1,
|
|
),
|
|
True,
|
|
(),
|
|
("self_health_warning", "performance_warning"),
|
|
),
|
|
(
|
|
_with_overrides(selfHealthSeverity="critical"),
|
|
False,
|
|
("self_health_not_ok",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(taskFailureCount=1),
|
|
False,
|
|
("scheduled_task_unhealthy",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(requiredEvidenceReserveFailureCount=1),
|
|
False,
|
|
("required_evidence_freshness_reserve_insufficient",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(manifest={"runtimeMatched": False, "mismatchCount": 1}),
|
|
False,
|
|
("runtime_manifest_mismatch",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(manifest={"sourceRevision": "not-a-full-sha"}),
|
|
False,
|
|
("runtime_manifest_identity_invalid",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(manifest={"fileCount": 15}),
|
|
False,
|
|
("runtime_manifest_identity_invalid",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(relayListenerCount=0),
|
|
False,
|
|
("sre_alert_relay_not_listening",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(alertIncomingCount=1),
|
|
True,
|
|
(),
|
|
("fresh_alert_pending_next_inbox_tick",),
|
|
),
|
|
(
|
|
_with_overrides(
|
|
alertIncomingCount=1,
|
|
staleAlertIncomingCount=1,
|
|
),
|
|
False,
|
|
("pending_work_before_reboot",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(pendingCommandBlockingCount=1),
|
|
False,
|
|
("pending_work_before_reboot",),
|
|
(),
|
|
),
|
|
(
|
|
_with_overrides(promotionDeferredCommandCount=1),
|
|
True,
|
|
(),
|
|
("maintenance_pending_commands_deferred_until_post_promotion",),
|
|
),
|
|
)
|
|
|
|
for case, expected_eligible, expected_blockers, expected_warnings in cases:
|
|
result = _run_decision(powershell, case)
|
|
assert result["deploymentEligible"] is expected_eligible
|
|
assert set(_as_list(result.get("blockingReasons"))) == set(expected_blockers)
|
|
assert set(_as_list(result.get("warningReasons"))) == set(expected_warnings)
|
|
|
|
promotion_reserve = _run_decision(
|
|
powershell,
|
|
_with_overrides(
|
|
manifestCheckRequired=False,
|
|
manifest={"sourceRevision": "old", "fileCount": 15, "runtimeMatched": False},
|
|
),
|
|
)
|
|
assert promotion_reserve["deploymentEligible"] is True
|
|
assert promotion_reserve["manifestCheckRequired"] is False
|
|
|
|
promotion_reserve_task_failure = _run_decision(
|
|
powershell,
|
|
_with_overrides(
|
|
manifestCheckRequired=False,
|
|
manifest={"sourceRevision": "old", "fileCount": 15, "runtimeMatched": False},
|
|
taskFailureCount=1,
|
|
),
|
|
)
|
|
assert promotion_reserve_task_failure["deploymentEligible"] is False
|
|
assert promotion_reserve_task_failure["blockingReasons"] == [
|
|
"scheduled_task_unhealthy"
|
|
]
|
|
|
|
preflight_path = str(PREFLIGHT).replace("'", "''")
|
|
parser_command = (
|
|
"$tokens=$null; $errors=$null; "
|
|
"[System.Management.Automation.Language.Parser]::ParseFile("
|
|
f"'{preflight_path}', [ref]$tokens, [ref]$errors) | Out-Null; "
|
|
"if ($errors.Count -gt 0) { "
|
|
"$errors | ForEach-Object { $_.Message }; exit 1 }"
|
|
)
|
|
parsed = subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-Command",
|
|
parser_command,
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert parsed.returncode == 0, parsed.stdout + parsed.stderr
|