76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
import check_mcp_live_readiness as readiness
|
|
|
|
|
|
class McpLiveReadinessTests(unittest.TestCase):
|
|
def test_confirm_live_skips_probe_when_proxy_reports_no_polling(self) -> None:
|
|
calls: list[tuple[str, str]] = []
|
|
|
|
def fake_request_json(url: str, **kwargs: Any) -> tuple[dict[str, Any] | None, str | None, float]:
|
|
calls.append((url, str(kwargs.get("method") or "GET")))
|
|
if url.endswith("/api/health"):
|
|
return {"ok": True}, None, 0.01
|
|
if url.endswith("/health"):
|
|
return {"status": "healthy", "polling_channels_count": 0}, None, 0.01
|
|
return {"success": True}, None, 0.01
|
|
|
|
result = readiness.check_readiness(confirm_live=True, request_json_func=fake_request_json)
|
|
|
|
self.assertEqual(result["verdict"]["status"], "not_ready")
|
|
self.assertFalse(result["verdict"]["ready_for_live_replay"])
|
|
self.assertIn("/1c/poll", result["verdict"]["reason"])
|
|
self.assertNotIn(("http://127.0.0.1:6003/api/get_metadata?channel=default", "POST"), calls)
|
|
|
|
def test_confirm_live_allows_probe_when_polling_was_observed(self) -> None:
|
|
calls: list[tuple[str, str]] = []
|
|
|
|
def fake_request_json(url: str, **kwargs: Any) -> tuple[dict[str, Any] | None, str | None, float]:
|
|
calls.append((url, str(kwargs.get("method") or "GET")))
|
|
if url.endswith("/api/health"):
|
|
return {"ok": True}, None, 0.01
|
|
if url.endswith("/health"):
|
|
return {"status": "healthy", "polling_channels_count": 1}, None, 0.01
|
|
return {"success": True, "data": []}, None, 0.02
|
|
|
|
result = readiness.check_readiness(confirm_live=True, request_json_func=fake_request_json)
|
|
|
|
self.assertEqual(result["verdict"]["status"], "ready")
|
|
self.assertTrue(result["verdict"]["ready_for_live_replay"])
|
|
self.assertIn(("http://127.0.0.1:6003/api/get_metadata?channel=default", "POST"), calls)
|
|
|
|
def test_confirm_live_waits_for_polling_before_probe(self) -> None:
|
|
proxy_health_calls = 0
|
|
|
|
def fake_request_json(url: str, **kwargs: Any) -> tuple[dict[str, Any] | None, str | None, float]:
|
|
nonlocal proxy_health_calls
|
|
if url.endswith("/api/health"):
|
|
return {"ok": True}, None, 0.01
|
|
if url.endswith("/health"):
|
|
proxy_health_calls += 1
|
|
return {"status": "healthy", "polling_channels_count": 1 if proxy_health_calls >= 2 else 0}, None, 0.01
|
|
return {"success": True, "data": []}, None, 0.02
|
|
|
|
result = readiness.check_readiness(
|
|
confirm_live=True,
|
|
wait_for_polling_seconds=0.1,
|
|
poll_interval_seconds=0,
|
|
request_json_func=fake_request_json,
|
|
)
|
|
|
|
self.assertEqual(result["verdict"]["status"], "ready")
|
|
self.assertTrue(result["poll_wait"]["observed_polling"])
|
|
self.assertGreaterEqual(proxy_health_calls, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|