diff --git a/bin/base-demo-services b/bin/base-demo-services index 69e12c1..5deec43 100755 --- a/bin/base-demo-services +++ b/bin/base-demo-services @@ -16,7 +16,13 @@ import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, NamedTuple + + +class CheckResult(NamedTuple): + ok: bool + label: str + detail: str def project_root() -> Path: @@ -246,79 +252,67 @@ def stop_process(service: dict[str, Any], root: Path) -> int: return 0 -def check_service(service: dict[str, Any], root: Path) -> tuple[bool, str]: +def check_service(service: dict[str, Any], root: Path) -> CheckResult: check = service.get("check") or {"type": "none"} check_type = check.get("type") if check_type == "none": - return True, "none" + return CheckResult(True, "none", "none") if check_type == "file": path_value = check.get("path") if not isinstance(path_value, str) or not path_value: - return False, "file:" + return CheckResult(False, "file:", "file:") path = Path(path_value) if not path.is_absolute(): path = root / path - return path.exists(), f"file:{path_value}" + label = f"file:{path_value}" + return CheckResult(path.exists(), label, label) if check_type == "http": - # HTTP checks use the service-level health_url so the display URL and probe target stay aligned. url = service.get("health_url") if not isinstance(url, str) or not url: - return False, "http:" + return CheckResult(False, "http:", "http:") + label = url try: with urllib.request.urlopen(url, timeout=1.0) as response: - return 200 <= response.status < 400, f"http:{url}" + return CheckResult(200 <= response.status < 400, label, f"http:{url}") except (OSError, urllib.error.URLError) as exc: - return False, f"http:{url} ({exc})" + return CheckResult(False, label, f"http:{url} ({exc})") if check_type == "compose": service_name = check.get("service") or service.get("compose_service") or service.get("name") if not isinstance(service_name, str) or not service_name: - return False, "compose:" + return CheckResult(False, "compose:", "compose:") + label = f"compose:{service_name}" if shutil.which("docker") is None: - return False, f"compose:{service_name} unavailable" + return CheckResult(False, label, f"{label} unavailable") command = compose_command(root, ["ps", "--services", "--status", "running"]) result = subprocess.run(command, check=False, capture_output=True, text=True) running = set(result.stdout.splitlines()) - return service_name in running, f"compose:{service_name}" + return CheckResult(service_name in running, label, label) if check_type == "command": command = check.get("command") if not isinstance(command, list) or not all(isinstance(item, str) for item in command): - return False, "command:" + return CheckResult(False, "command:", "command:") + label = f"command:{shlex.join(command)}" try: result = subprocess.run(command, cwd=root, check=False, capture_output=True, text=True, timeout=3.0) except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired) as exc: - return False, f"command:{shlex.join(command)} ({exc})" - return result.returncode == 0, f"command:{shlex.join(command)}" - - return False, f"unsupported:{check_type}" + return CheckResult(False, label, f"{label} ({exc})") + return CheckResult(result.returncode == 0, label, label) - -def health_label(service: dict[str, Any]) -> str: - check = service.get("check") or {"type": "none"} - if service.get("health_url"): - return str(service["health_url"]) - if check.get("type") == "compose": - return f"compose:{check.get('service', service.get('compose_service', '-'))}" - if check.get("type") == "file": - return f"file:{check.get('path', '-')}" - if check.get("type") == "command": - command = check.get("command") - if isinstance(command, list) and all(isinstance(item, str) for item in command): - return f"command:{shlex.join(command)}" - return "command:" - return str(check.get("type") or "-") + detail = f"unsupported:{check_type}" + return CheckResult(False, detail, detail) def service_rows(services: list[dict[str, Any]], root: Path) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] for service in services: process_state = read_process_state(root, service) - ok, detail = check_service(service, root) - if ok: + check_result = check_service(service, root) + if check_result.ok: state = "healthy" elif service.get("required", False): state = "unhealthy" @@ -337,11 +331,11 @@ def service_rows(services: list[dict[str, Any]], root: Path) -> list[dict[str, s "kind": str(service.get("kind", "-")), "runtime": str(service.get("runtime", "-")), "port": str(service.get("port") or "-"), - "health": health_label(service), + "health": check_result.label, "state": state, "since": since, "logs": str(service.get("logs") or "-"), - "detail": detail, + "detail": check_result.detail, "required": "true" if service.get("required", False) else "false", } ) diff --git a/tests/services_test.bats b/tests/services_test.bats index 905fe01..d07ab4f 100644 --- a/tests/services_test.bats +++ b/tests/services_test.bats @@ -66,6 +66,29 @@ write_optional_compose_catalog() { EOF } +write_missing_http_catalog() { + local catalog="$1" + + cat > "$catalog" <"* ]] + + run env BASE_DEMO_SERVICES_STATE_DIR="$TEST_TMPDIR/state" "$TEST_ROOT/bin/base-demo-services" --catalog "$catalog" check + + [ "$status" -eq 0 ] + [[ "$output" == *"missing-http skip optional http:"* ]] +}