Files
wehub-resource-sync 4b6817381b
Benchmark image — build + push to ECR (any adapter) / build + push (push) Waiting to run
CI / quality (ubuntu-latest) (push) Waiting to run
CI / test (tools-runtime) (push) Waiting to run
CI / test (e2e-general) (push) Waiting to run
CI / test (cli-runtime) (push) Waiting to run
CI / test (e2e-provider-and-openclaw) (push) Waiting to run
CI / test (integrations-and-misc) (push) Waiting to run
CI / coverage-report (push) Blocked by required conditions
CI / test-kubernetes (push) Waiting to run
CI / should-run-thorough (push) Waiting to run
CI / test-thorough (cloudwatch-demo) (push) Blocked by required conditions
CI / test-thorough (flink-ecs) (push) Blocked by required conditions
CI / test-thorough (upstream-lambda) (push) Blocked by required conditions
CI / test-thorough (prefect-ecs-fargate) (push) Blocked by required conditions
CodeQL / Analyze (python) (push) Waiting to run
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Blocked by required conditions
Release / publish-release (push) Blocked by required conditions
Release / publish-main-release (push) Blocked by required conditions
Release / prepare (push) Waiting to run
Release / verify (push) Blocked by required conditions
Release / build-python-dist (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Blocked by required conditions
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Waiting to run
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

183 lines
5.5 KiB
Python

import json
from io import StringIO
from pathlib import Path
from unittest.mock import MagicMock, patch
from rich.console import Console
from platform.terminal.theme import ERROR, HIGHLIGHT, WARNING
from surfaces.interactive_shell.ui.health import (
_summary_counts,
render_health_json,
render_health_report,
status_badge,
)
def test_status_badge() -> None:
# Test passed statuses
for s in ["passed", "pass", "ok", "healthy", " PASSED "]:
badge = status_badge(s)
assert badge.plain == "PASSED"
assert badge.style == f"bold {HIGHLIGHT}"
# Test warn statuses
for s in ["warn", "warning", "degraded", "outdated"]:
badge = status_badge(s)
assert badge.plain == "WARN"
assert badge.style == f"bold {WARNING}"
# Test missing status
badge = status_badge("missing")
assert badge.plain == "MISSING"
assert badge.style == f"bold {WARNING}"
# Test failed statuses
for s in ["failed", "fail", "error", "unhealthy"]:
badge = status_badge(s)
assert badge.plain == "FAILED"
assert badge.style == f"bold {ERROR}"
# Test unknown status
badge = status_badge("unknown_status")
assert badge.plain == "UNKNOWN_STATUS"
assert badge.style == "bold"
# Test empty status
badge = status_badge("")
assert badge.plain == "UNKNOWN"
assert badge.style == "bold"
def test_summary_counts() -> None:
results = [
{"status": "passed"},
{"status": "PASSED"},
{"status": "missing"},
{"status": "failed"},
{"status": "unknown"},
{"status": "error"},
]
counts = _summary_counts(results)
assert counts["passed"] == 2
assert counts["missing"] == 1
assert counts["failed"] == 2
assert counts["other"] == 1
def test_summary_counts_normalizes_failure_aliases() -> None:
results = [
{"status": "failed"},
{"status": "fail"},
{"status": "error"},
{"status": "unhealthy"},
{"status": "FAILED"},
]
counts = _summary_counts(results)
assert counts["failed"] == 5
assert counts["other"] == 0
def test_summary_counts_normalizes_passed_aliases() -> None:
results = [
{"status": "passed"},
{"status": "pass"},
{"status": "ok"},
{"status": "healthy"},
]
counts = _summary_counts(results)
assert counts["passed"] == 4
assert counts["other"] == 0
def test_summary_counts_unknown_stays_other() -> None:
results = [
{"status": "warn"},
{"status": "degraded"},
{"status": "weird"},
{"status": ""},
]
counts = _summary_counts(results)
assert counts["other"] == 4
assert counts["passed"] == 0
assert counts["failed"] == 0
assert counts["missing"] == 0
def test_render_health_json(capsys) -> None:
environment = "test-env"
store_path = Path("/tmp/store")
results = [
{"service": "aws", "source": "env", "status": "passed", "detail": "ok"},
{"service": "github", "source": "config", "status": "failed", "detail": "error"},
]
render_health_json(
environment=environment,
integration_store_path=store_path,
results=results,
)
captured = capsys.readouterr()
output = captured.out
data = json.loads(output)
assert data["environment"] == environment
assert data["integration_store"] == str(store_path)
assert data["summary"] == {"passed": 1, "missing": 0, "failed": 1, "other": 0}
assert len(data["results"]) == 2
assert data["results"][0]["service"] == "aws"
assert data["results"][1]["status"] == "failed"
@patch("platform.guardrails.rules.get_default_rules_path")
def test_render_health_report_action_messages(mock_rules_path: MagicMock) -> None:
mock_rules_path.return_value = Path("/nonexistent/rules")
# Case 1: All healthy
console = Console(file=StringIO(), force_terminal=False, width=100)
render_health_report(
console=console,
environment="test",
integration_store_path="/tmp",
results=[{"service": "s1", "status": "passed"}],
)
output = console.file.getvalue()
assert "All configured integrations look healthy." in output
# Case 2: Some missing
console = Console(file=StringIO(), force_terminal=False, width=100)
render_health_report(
console=console,
environment="test",
integration_store_path="/tmp",
results=[{"service": "s1", "status": "missing"}],
)
output = console.file.getvalue()
assert "Action: Configure missing integrations" in output
# Case 3: Some failed
console = Console(file=StringIO(), force_terminal=False, width=100)
render_health_report(
console=console,
environment="test",
integration_store_path="/tmp",
results=[{"service": "s1", "status": "failed"}],
)
output = console.file.getvalue()
assert "Action: Fix failed integrations" in output
# Case 4: Both missing and failed (failed should take precedence based on if/elif)
console = Console(file=StringIO(), force_terminal=False, width=100)
render_health_report(
console=console,
environment="test",
integration_store_path="/tmp",
results=[
{"service": "s1", "status": "failed"},
{"service": "s2", "status": "missing"},
],
)
output = console.file.getvalue()
assert "Action: Fix failed integrations" in output
assert "Action: Configure missing integrations" not in output