from __future__ import annotations import socket from contextlib import closing from typing import Any, Dict, Iterable, List, Tuple from lab.utils import command_available, run def _result(name: str, ok: bool, detail: str, **extra: Any) -> Dict[str, Any]: payload = {"name": name, "ok": ok, "detail": detail} payload.update(extra) return payload def _parse_host_ports(profiles: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: ports: List[Dict[str, Any]] = [] for profile in profiles: for service_name, service in (profile.get("services") or {}).items(): for binding in service.get("ports", []) or []: host_port = None value = str(binding) parts = value.split(":") if len(parts) == 3 and parts[1].isdigit(): host_port = int(parts[1]) elif len(parts) >= 2 and parts[0].isdigit(): host_port = int(parts[0]) elif value.isdigit(): host_port = int(value) if host_port is None: continue ports.append( { "profile_id": profile.get("profile_id"), "service": service_name, "binding": value, "port": host_port, } ) return ports def _port_available(port: int) -> bool: with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: sock.bind(("127.0.0.1", port)) except OSError: return False return True def _check_docker_cli() -> Dict[str, Any]: ok = command_available("docker") return _result("docker-cli", ok, "docker CLI available" if ok else "docker CLI is not installed") def _check_docker_daemon() -> Dict[str, Any]: if not command_available("docker"): return _result("docker-daemon", False, "docker CLI unavailable") context = run(["docker", "context", "show"], check=False) info = run(["docker", "info"], check=False) detail = f"context={context.stdout.strip() or 'unknown'}" if info.returncode != 0: detail = info.stderr.strip() or info.stdout.strip() or "docker daemon unavailable" return _result("docker-daemon", False, detail) return _result("docker-daemon", True, detail or "docker daemon reachable") def _check_playwright_import() -> Dict[str, Any]: try: from playwright.sync_api import sync_playwright # noqa: F401 except Exception as exc: return _result("playwright-import", False, f"playwright import failed: {exc}") return _result("playwright-import", True, "playwright Python package import passed") def _check_chromium_launch() -> Dict[str, Any]: try: from playwright.sync_api import sync_playwright except Exception as exc: return _result("playwright-browser", False, f"playwright import failed: {exc}") try: with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) page = browser.new_page() page.set_content("ok") browser.close() except Exception as exc: return _result("playwright-browser", False, f"chromium launch failed: {exc}") return _result("playwright-browser", True, "chromium runtime launch passed") def _check_ports(profiles: Iterable[Dict[str, Any]]) -> Dict[str, Any]: requested = _parse_host_ports(profiles) if not requested: return _result("ports", True, "no host ports declared") conflicts: List[Dict[str, Any]] = [] for item in requested: if not _port_available(item["port"]): conflicts.append(item) if conflicts: detail = ", ".join( f"{item['port']}({item['profile_id']}::{item['service']})" for item in conflicts ) return _result("ports", False, f"host ports already in use: {detail}", conflicts=conflicts) return _result("ports", True, f"checked {len(requested)} host port bindings", bindings=requested) def run_checks(profiles: Iterable[Dict[str, Any]] | None = None) -> Dict[str, Any]: selected = list(profiles or []) require_browser = not selected or any(bool(item.get("browser_assertions", {}).get("required")) for item in selected) checks = [ _check_docker_cli(), _check_docker_daemon(), _check_playwright_import() if require_browser else _result("playwright-import", True, "not required for selected profiles"), _check_chromium_launch() if require_browser else _result("playwright-browser", True, "not required for selected profiles"), _check_ports(selected), ] ok = all(item["ok"] for item in checks) failures = [item for item in checks if not item["ok"]] return { "status": "passed" if ok else "failed", "ok": ok, "checks": checks, "profile_ids": [item.get("profile_id") for item in selected if item.get("profile_id")], "failure_count": len(failures), "summary": "; ".join(item["detail"] for item in failures) if failures else "all checks passed", }