didericis
72286f188d
The launch path's counterpart to the gateway-side PolicyResolver: a trusted host-side HTTP client for the orchestrator control plane, so the CLI can register/teardown/re-policy bottles. Stdlib-only. - OrchestratorClient.register_bottle(source_ip, *, image_ref, metadata, policy) -> RegisteredBottle(bottle_id, identity_token) (POST /bottles) - teardown_bottle(id) -> bool (DELETE; 404 -> False, idempotent for cleanup) - set_policy(id, policy) -> bool (PUT; live reload) - list_bottles() / health() - Unlike the fail-closed data-plane resolver, this is the trusted caller: a non-2xx (other than the meaningful 404s) raises OrchestratorClientError so the launch path surfaces failures rather than swallowing them. pyright 0 errors; pylint 9.82/10; unit suite green (1742 tests; the 13 test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Unit: host-side orchestrator control-plane client (PRD 0070). HTTP mocked."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
import urllib.error
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bot_bottle.orchestrator.client import (
|
|
OrchestratorClient,
|
|
OrchestratorClientError,
|
|
RegisteredBottle,
|
|
)
|
|
|
|
_URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
|
|
|
|
|
|
def _resp(status: int, payload: object) -> MagicMock:
|
|
m = MagicMock()
|
|
inner = m.__enter__.return_value
|
|
inner.status = status
|
|
inner.read.return_value = json.dumps(payload).encode()
|
|
return m
|
|
|
|
|
|
def _http_error(code: int, payload: object = None) -> urllib.error.HTTPError:
|
|
del payload # the client tolerates an empty error body; keep the signature
|
|
return urllib.error.HTTPError("http://x", code, "err", {}, None) # type: ignore[arg-type]
|
|
|
|
|
|
class TestRegister(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.c = OrchestratorClient("http://orch:8080")
|
|
|
|
def test_register_returns_id_and_token(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1", "identity_token": "tok"})):
|
|
got = self.c.register_bottle("10.0.0.2", policy="routes: []\n", metadata="{}")
|
|
self.assertEqual(RegisteredBottle("b1", "tok"), got)
|
|
|
|
def test_register_posts_source_ip_and_policy(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b", "identity_token": "t"})) as m:
|
|
self.c.register_bottle("10.0.0.9", policy="P", metadata="M", image_ref="img")
|
|
sent = json.loads(m.call_args.args[0].data)
|
|
self.assertEqual("10.0.0.9", sent["source_ip"])
|
|
self.assertEqual("P", sent["policy"])
|
|
self.assertEqual("img", sent["image_ref"])
|
|
|
|
def test_register_missing_fields_raises(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1"})):
|
|
with self.assertRaises(OrchestratorClientError):
|
|
self.c.register_bottle("10.0.0.2")
|
|
|
|
def test_register_non_2xx_raises(self) -> None:
|
|
with patch(_URLOPEN, side_effect=_http_error(400, {"error": "bad"})):
|
|
with self.assertRaises(OrchestratorClientError):
|
|
self.c.register_bottle("")
|
|
|
|
|
|
class TestTeardown(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.c = OrchestratorClient("http://orch:8080")
|
|
|
|
def test_teardown_true_on_success(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})):
|
|
self.assertTrue(self.c.teardown_bottle("b1"))
|
|
|
|
def test_teardown_false_on_404(self) -> None:
|
|
# Idempotent: an already-gone bottle is a clean no-op.
|
|
with patch(_URLOPEN, side_effect=_http_error(404)):
|
|
self.assertFalse(self.c.teardown_bottle("gone"))
|
|
|
|
def test_teardown_uses_delete(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})) as m:
|
|
self.c.teardown_bottle("b1")
|
|
self.assertEqual("DELETE", m.call_args.args[0].get_method())
|
|
|
|
|
|
class TestHealthAndPolicy(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.c = OrchestratorClient("http://orch:8080")
|
|
|
|
def test_health_true_on_200(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(200, {"status": "ok"})):
|
|
self.assertTrue(self.c.health())
|
|
|
|
def test_health_false_when_unreachable(self) -> None:
|
|
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
|
self.assertFalse(self.c.health())
|
|
|
|
def test_set_policy_false_on_404(self) -> None:
|
|
with patch(_URLOPEN, side_effect=_http_error(404)):
|
|
self.assertFalse(self.c.set_policy("gone", "P"))
|
|
|
|
def test_set_policy_true_on_success(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp(200, {"updated": True})):
|
|
self.assertTrue(self.c.set_policy("b1", "P"))
|
|
|
|
def test_unreachable_raises(self) -> None:
|
|
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
|
with self.assertRaises(OrchestratorClientError):
|
|
self.c.register_bottle("10.0.0.2")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|