diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index c18d6a5..0eb11da 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -20,6 +20,7 @@ import supervise as _sv # noqa: E402 # type: ignore from bot_bottle import supervise_server # noqa: E402 from bot_bottle.supervise_server import ( + ERR_INTERNAL, ERR_INVALID_PARAMS, ERR_INVALID_REQUEST, ERR_METHOD_NOT_FOUND, @@ -29,7 +30,9 @@ from bot_bottle.supervise_server import ( PROPOSED_FILE_FIELD, ServerConfig, TOOL_DEFINITIONS, + _RpcClientError, _RpcError, + _RpcInternalError, _response_timeout_from_env, format_response_text, handle_initialize, @@ -47,15 +50,15 @@ from bot_bottle.supervise_server import ( class TestValidation(unittest.TestCase): - def test_capability_block_accepts_anything_nonempty(self): - validate_proposed_file( - _sv.TOOL_CAPABILITY_BLOCK, - "FROM python:3.13\nRUN apk add git\n", - ) - def test_empty_proposed_file_rejected_for_tools_with_file_field(self): with self.assertRaises(_RpcError): - validate_proposed_file(_sv.TOOL_CAPABILITY_BLOCK, " \n\t") + validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, " \n\t") + + def test_capability_block_rejected_as_unknown_tool(self): + with self.assertRaises(_RpcError) as cm: + validate_proposed_file("capability-block", "FROM python:3.13\n") + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + self.assertIn("unknown tool", cm.exception.message) def test_egress_routes_yaml_is_validated(self): validate_proposed_file( @@ -77,6 +80,65 @@ class TestValidation(unittest.TestCase): self.assertIn("must not change egress logging", cm.exception.message) +# --- Error taxonomy -------------------------------------------------------- + + +class TestRpcErrorTaxonomy(unittest.TestCase): + def test_rpc_client_error_is_rpc_error(self): + e = _RpcClientError(ERR_INVALID_PARAMS, "bad param") + self.assertIsInstance(e, _RpcError) + self.assertEqual(ERR_INVALID_PARAMS, e.code) + self.assertEqual("bad param", e.message) + + def test_rpc_internal_error_is_rpc_error(self): + e = _RpcInternalError("disk full") + self.assertIsInstance(e, _RpcError) + self.assertEqual(ERR_INTERNAL, e.code) + self.assertEqual("disk full", e.message) + + def test_rpc_internal_error_preserves_cause(self): + cause = OSError("no space left on device") + try: + raise _RpcInternalError("failed to write") from cause + except _RpcInternalError as e: + self.assertIs(cause, e.__cause__) + + def test_parse_error_is_client_error(self): + with self.assertRaises(_RpcClientError): + parse_jsonrpc(b"{bad json") + + def test_validation_error_is_client_error(self): + with self.assertRaises(_RpcClientError): + validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n") + + def test_unknown_tool_in_tools_call_is_client_error(self): + config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")) + with self.assertRaises(_RpcClientError) as cm: + handle_tools_call({"name": "no-such-tool", "arguments": {}}, config) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + + +class TestRpcInternalErrorOnIoFailure(unittest.TestCase): + def test_write_proposal_os_error_raises_internal(self): + config = ServerConfig( + bottle_slug="dev", + queue_dir=Path("/dev/null/cannot-exist"), + ) + with self.assertRaises(_RpcInternalError) as cm: + handle_tools_call( + { + "name": _sv.TOOL_EGRESS_ALLOW, + "arguments": { + "routes_yaml": "routes:\n - host: example.com\n", + "justification": "x", + }, + }, + config, + ) + self.assertEqual(ERR_INTERNAL, cm.exception.code) + self.assertIsNotNone(cm.exception.__cause__) + + # --- JSON-RPC parsing ------------------------------------------------------ @@ -157,7 +219,6 @@ class TestHandleToolsList(unittest.TestCase): self.assertEqual( sorted([ _sv.TOOL_EGRESS_ALLOW, - _sv.TOOL_CAPABILITY_BLOCK, _sv.TOOL_EGRESS_BLOCK, _sv.TOOL_LIST_EGRESS_ROUTES, ]), @@ -233,10 +294,10 @@ class TestHandleToolsCall(unittest.TestCase): try: result = handle_tools_call( { - "name": _sv.TOOL_CAPABILITY_BLOCK, + "name": _sv.TOOL_EGRESS_BLOCK, "arguments": { - "dockerfile": "FROM python:3.13\n", - "justification": "need git", + "routes_yaml": "routes:\n - host: example.com\n", + "justification": "need example.com", }, }, self.config, @@ -273,9 +334,9 @@ class TestHandleToolsCall(unittest.TestCase): try: result = handle_tools_call( { - "name": _sv.TOOL_CAPABILITY_BLOCK, + "name": _sv.TOOL_EGRESS_ALLOW, "arguments": { - "dockerfile": "FROM python:3.13\n", + "routes_yaml": "routes:\n - host: example.com\n", "justification": "needed for tests", }, }, @@ -297,20 +358,52 @@ class TestHandleToolsCall(unittest.TestCase): with self.assertRaises(_RpcError): handle_tools_call( { - "name": _sv.TOOL_CAPABILITY_BLOCK, - "arguments": {"dockerfile": "FROM python:3.13\n"}, + "name": _sv.TOOL_EGRESS_ALLOW, + "arguments": {"routes_yaml": "routes:\n - host: example.com\n"}, }, self.config, ) + def test_missing_name_raises(self): + with self.assertRaises(_RpcError) as cm: + handle_tools_call({"arguments": {}}, self.config) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + + def test_arguments_must_be_object(self): + with self.assertRaises(_RpcError) as cm: + handle_tools_call( + { + "name": _sv.TOOL_EGRESS_ALLOW, + "arguments": [], + }, + self.config, + ) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + self.assertIn("must be an object", cm.exception.message) + + def test_capability_block_call_raises_unknown_tool(self): + with self.assertRaises(_RpcError) as cm: + handle_tools_call( + { + "name": "capability-block", + "arguments": { + "dockerfile": "FROM python:3.13\n", + "justification": "need git", + }, + }, + self.config, + ) + self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) + self.assertIn("unknown tool", cm.exception.message) + def test_archives_proposal_after_response(self): responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED) try: handle_tools_call( { - "name": _sv.TOOL_CAPABILITY_BLOCK, + "name": _sv.TOOL_EGRESS_ALLOW, "arguments": { - "dockerfile": "FROM python:3.13\n", + "routes_yaml": "routes:\n - host: example.com\n", "justification": "x", }, }, @@ -332,10 +425,10 @@ class TestHandleToolsCall(unittest.TestCase): ) result = handle_tools_call( { - "name": _sv.TOOL_CAPABILITY_BLOCK, + "name": _sv.TOOL_EGRESS_ALLOW, "arguments": { - "dockerfile": "FROM python:3.13\n", - "justification": "need a capability", + "routes_yaml": "routes:\n - host: example.com\n", + "justification": "need egress", }, }, config, @@ -350,6 +443,31 @@ class TestHandleToolsCall(unittest.TestCase): class TestHandleListEgressRoutes(unittest.TestCase): + def test_success_returns_body_text(self): + class _Resp: + def __enter__(self): + return self + + def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object) -> bool: + return False + + def read(self): + return b"[{\"host\": \"example.com\"}]" + + class _Opener: + def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore + return _Resp() + + with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()): + result = handle_list_egress_routes( + {}, + ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")), + ) + + self.assertFalse(result["isError"]) # type: ignore[index] + text = result["content"][0]["text"] # type: ignore[index] + self.assertIn("example.com", text) + def test_url_error_returns_tool_error(self): class _Opener: def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore @@ -409,6 +527,13 @@ class TestFormatResponseText(unittest.TestCase): self.assertIn("the operator modified", text.lower()) +class TestFormatPendingResponseText(unittest.TestCase): + def test_formats_timeout_message(self): + text = supervise_server.format_pending_response_text(12.5) + self.assertIn("status: pending", text) + self.assertIn("12.5s", text) + + # --- End-to-end HTTP sanity ------------------------------------------------ @@ -459,7 +584,7 @@ class TestHttpEndToEnd(unittest.TestCase): self.assertEqual("2.0", result["jsonrpc"]) self.assertEqual(1, result["id"]) names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index] - self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names) + self.assertNotIn("capability-block", names) self.assertIn(_sv.TOOL_EGRESS_ALLOW, names) self.assertIn(_sv.TOOL_EGRESS_BLOCK, names) @@ -469,6 +594,26 @@ class TestHttpEndToEnd(unittest.TestCase): ) self.assertEqual(ERR_METHOD_NOT_FOUND, result["error"]["code"]) # type: ignore[index] + def test_internal_error_returns_err_internal_over_http(self): + with patch.object( + supervise_server._sv, "write_proposal", + side_effect=OSError("disk full"), + ): + result = self._post_jsonrpc({ + "jsonrpc": "2.0", + "id": 99, + "method": "tools/call", + "params": { + "name": _sv.TOOL_EGRESS_ALLOW, + "arguments": { + "routes_yaml": "routes:\n - host: example.com\n", + "justification": "x", + }, + }, + }) + self.assertIn("error", result) + self.assertEqual(ERR_INTERNAL, result["error"]["code"]) # type: ignore[index] + def test_health_endpoint(self): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) try: