s
This commit is contained in:
@@ -1,8 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.control_plane import SERVICE_SPECS, stack_summary
|
||||
from src.control_plane import SERVICE_SPECS, deployment_report, stack_summary
|
||||
|
||||
|
||||
class ControlPlaneServiceTests(unittest.TestCase):
|
||||
@@ -30,6 +33,83 @@ class ControlPlaneServiceTests(unittest.TestCase):
|
||||
):
|
||||
self.assertIn(key, links)
|
||||
|
||||
def test_deployment_report_marks_live_local_ready_when_checks_pass(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
base_dir = Path(tmp)
|
||||
console_index = base_dir / "web-console" / "out" / "index.html"
|
||||
console_index.parent.mkdir(parents=True, exist_ok=True)
|
||||
console_index.write_text("ok", encoding="utf-8")
|
||||
stack_env = base_dir / "config" / "system-stack.env"
|
||||
stack_env.parent.mkdir(parents=True, exist_ok=True)
|
||||
stack_env.write_text("PORT=8001\nENABLE_LIVE_EDGE_PROXY=1\nENABLE_MDNS=1\n", encoding="utf-8")
|
||||
|
||||
def probe(url: str, timeout: float = 2.5) -> dict[str, object]:
|
||||
return {
|
||||
"reachable": True,
|
||||
"http_code": 200,
|
||||
"latency_ms": 3.1,
|
||||
"error": "",
|
||||
}
|
||||
|
||||
with patch("src.control_plane.BASE_DIR", base_dir):
|
||||
with patch("src.control_plane.STACK_ENV", stack_env):
|
||||
with patch("src.control_plane.load_stack_env", return_value={
|
||||
"PORT": "8001",
|
||||
"ENABLE_LIVE_EDGE_PROXY": "1",
|
||||
"ENABLE_MDNS": "1",
|
||||
}):
|
||||
with patch("src.control_plane.stack_summary", return_value={
|
||||
"hostname": "live",
|
||||
"mdns_host": "live.local",
|
||||
"ips": ["192.168.1.10"],
|
||||
}):
|
||||
with patch("src.control_plane._probe_http_get", side_effect=probe):
|
||||
with patch("src.control_plane._resolve_host_ipv4s", return_value=["192.168.1.10"]):
|
||||
report = deployment_report()
|
||||
|
||||
self.assertTrue(report["ready"])
|
||||
ids = {item["id"] for item in report["checks"]}
|
||||
self.assertIn("api_loopback", ids)
|
||||
self.assertIn("edge_loopback", ids)
|
||||
self.assertIn("mdns_edge", ids)
|
||||
self.assertEqual(report["urls"]["mdns_root"], "http://live.local/")
|
||||
|
||||
def test_deployment_report_explains_edge_and_mdns_failures(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
base_dir = Path(tmp)
|
||||
console_index = base_dir / "web-console" / "out" / "index.html"
|
||||
console_index.parent.mkdir(parents=True, exist_ok=True)
|
||||
console_index.write_text("ok", encoding="utf-8")
|
||||
stack_env = base_dir / "config" / "system-stack.env"
|
||||
stack_env.parent.mkdir(parents=True, exist_ok=True)
|
||||
stack_env.write_text("PORT=8001\nENABLE_LIVE_EDGE_PROXY=1\nENABLE_MDNS=1\n", encoding="utf-8")
|
||||
|
||||
def probe(url: str, timeout: float = 2.5) -> dict[str, object]:
|
||||
if "127.0.0.1:8001/health" in url:
|
||||
return {"reachable": True, "http_code": 200, "latency_ms": 2.0, "error": ""}
|
||||
return {"reachable": False, "http_code": 0, "latency_ms": 2.0, "error": "connection refused"}
|
||||
|
||||
with patch("src.control_plane.BASE_DIR", base_dir):
|
||||
with patch("src.control_plane.STACK_ENV", stack_env):
|
||||
with patch("src.control_plane.load_stack_env", return_value={
|
||||
"PORT": "8001",
|
||||
"ENABLE_LIVE_EDGE_PROXY": "1",
|
||||
"ENABLE_MDNS": "1",
|
||||
}):
|
||||
with patch("src.control_plane.stack_summary", return_value={
|
||||
"hostname": "live",
|
||||
"mdns_host": "live.local",
|
||||
"ips": ["192.168.1.10"],
|
||||
}):
|
||||
with patch("src.control_plane._probe_http_get", side_effect=probe):
|
||||
with patch("src.control_plane._resolve_host_ipv4s", return_value=[]):
|
||||
report = deployment_report()
|
||||
|
||||
self.assertFalse(report["ready"])
|
||||
joined = "\n".join(report["hints"])
|
||||
self.assertIn("live.local does not resolve locally", joined)
|
||||
self.assertIn("reverse proxy is unhealthy", joined)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user