项目文件夹

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

183 行
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