merge: update tests/unit/test_supervise_server.py from issue-277-coverage-ci

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: