Extract OCS error messages from Nextcloud response body

bigcat88 · bigcat88 · commit cf16bcdf2594 · 2026-03-27T10:47:52.000Z
OCS endpoints return specific error details in ocs.meta.message (e.g.
"User already exists", "Wrong share ID, share does not exist") but the
client was discarding these and mapping only HTTP status codes to generic
messages. Add _raise_for_ocs_status() that extracts the OCS message when
available, falling back to the HTTP status code mapping otherwise.
diff --git a/src/nextcloud_mcp/client.py b/src/nextcloud_mcp/client.py
@@ -18,20 +18,46 @@ def __init__(self, message: str, status_code: int) -> None:
         super().__init__(message)
 
 
+_STATUS_MESSAGES: dict[int, str] = {
+    401: "Authentication failed. Check NEXTCLOUD_USER and NEXTCLOUD_PASSWORD.",
+    403: "Forbidden. The user does not have permission for this operation.",
+    404: "Not found.",
+    409: "Conflict. The resource may already exist or the parent directory is missing.",
+    423: "Locked. The resource is currently locked by another process.",
+}
+
+
 def _raise_for_status(response: niquests.Response, context: str = "") -> None:
     """Raise NextcloudError with a helpful message instead of raw HTTPError."""
     if response.ok:
         return
     code = response.status_code or 0
     prefix = f"{context}: " if context else ""
-    messages: dict[int, str] = {
-        401: "Authentication failed. Check NEXTCLOUD_USER and NEXTCLOUD_PASSWORD.",
-        403: "Forbidden. The user does not have permission for this operation.",
-        404: "Not found.",
-        409: "Conflict. The resource may already exist or the parent directory is missing.",
-        423: "Locked. The resource is currently locked by another process.",
-    }
-    detail = messages.get(code, f"HTTP {code}")
+    detail = _STATUS_MESSAGES.get(code, f"HTTP {code}")
+    raise NextcloudError(f"{prefix}{detail}", code)
+
+
+def _raise_for_ocs_status(response: niquests.Response, context: str = "") -> None:
+    """Raise NextcloudError using the OCS error message from the response body when available.
+
+    Nextcloud OCS endpoints return error details in ocs.meta.message (e.g.
+    "User already exists", "Wrong share ID, share does not exist"). This
+    function extracts that message for a much better error experience than
+    the generic HTTP status code mapping.
+
+    Falls back to _raise_for_status() when the OCS body cannot be parsed.
+    """
+    if response.ok:
+        return
+    code = response.status_code or 0
+    prefix = f"{context}: " if context else ""
+    try:
+        ocs_message: str = response.json()["ocs"]["meta"]["message"]
+        if ocs_message:
+            raise NextcloudError(f"{prefix}{ocs_message}", code)
+    except (ValueError, KeyError, TypeError):
+        pass
+    detail = _STATUS_MESSAGES.get(code, f"HTTP {code}")
     raise NextcloudError(f"{prefix}{detail}", code)
 
 
@@ -108,7 +134,7 @@ async def ocs_get(self, path: str, params: dict[str, Any] | None = None) -> Any:
         session = await self._get_session()
         url = f"{self._base_url}/ocs/v2.php/{path}"
         response = await session.get(url, params=params or {})
-        _raise_for_status(response, f"OCS GET {path}")
+        _raise_for_ocs_status(response, f"OCS GET {path}")
         result: dict[str, Any] = response.json()  # type: ignore[assignment]
         return result["ocs"]["data"]
 
@@ -117,7 +143,7 @@ async def ocs_post(self, path: str, data: dict[str, Any] | None = None) -> Any:
         session = await self._get_session()
         url = f"{self._base_url}/ocs/v2.php/{path}"
         response = await session.post(url, data=data or {})
-        _raise_for_status(response, f"OCS POST {path}")
+        _raise_for_ocs_status(response, f"OCS POST {path}")
         result: dict[str, Any] = response.json()  # type: ignore[assignment]
         return result["ocs"]["data"]
 
@@ -126,7 +152,7 @@ async def ocs_put(self, path: str, data: dict[str, Any] | None = None) -> Any:
         session = await self._get_session()
         url = f"{self._base_url}/ocs/v2.php/{path}"
         response = await session.put(url, data=data or {})
-        _raise_for_status(response, f"OCS PUT {path}")
+        _raise_for_ocs_status(response, f"OCS PUT {path}")
         result: dict[str, Any] = response.json()  # type: ignore[assignment]
         return result["ocs"]["data"]
 
@@ -135,7 +161,7 @@ async def ocs_delete(self, path: str) -> Any:
         session = await self._get_session()
         url = f"{self._base_url}/ocs/v2.php/{path}"
         response = await session.delete(url)
-        _raise_for_status(response, f"OCS DELETE {path}")
+        _raise_for_ocs_status(response, f"OCS DELETE {path}")
         result: dict[str, Any] = response.json()  # type: ignore[assignment]
         return result["ocs"]["data"]
 
diff --git a/tests/integration/test_errors.py b/tests/integration/test_errors.py
@@ -59,6 +59,45 @@ async def test_error_has_status_code_attribute(self, nc_client: NextcloudClient)
         assert exc_info.value.status_code > 0
 
 
+class TestOcsErrorMessages:
+    """Verify that OCS error messages from Nextcloud are extracted, not discarded."""
+
+    @pytest.mark.asyncio
+    async def test_get_nonexistent_user_includes_ocs_message(self, nc_client: NextcloudClient) -> None:
+        with pytest.raises(NextcloudError, match="User does not exist") as exc_info:
+            await nc_client.ocs_get("cloud/users/nonexistent-user-xyz-12345")
+        assert exc_info.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_get_nonexistent_share_includes_ocs_message(self, nc_client: NextcloudClient) -> None:
+        with pytest.raises(NextcloudError, match="Wrong share ID, share does not exist") as exc_info:
+            await nc_client.ocs_get("apps/files_sharing/api/v1/shares/999999")
+        assert exc_info.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_share_nonexistent_path_includes_ocs_message(self, nc_client: NextcloudClient) -> None:
+        with pytest.raises(NextcloudError, match="Wrong path") as exc_info:
+            await nc_client.ocs_post(
+                "apps/files_sharing/api/v1/shares",
+                data={"path": "/nonexistent-xyz-12345.txt", "shareType": 3},
+            )
+        assert exc_info.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_create_duplicate_user_includes_ocs_message(self, nc_client: NextcloudClient) -> None:
+        with pytest.raises(NextcloudError, match="already exists") as exc_info:
+            await nc_client.ocs_post("cloud/users", data={"userid": "admin", "password": "test"})
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_ocs_error_preserves_context_prefix(self, nc_client: NextcloudClient) -> None:
+        with pytest.raises(NextcloudError) as exc_info:
+            await nc_client.ocs_get("cloud/users/nonexistent-user-xyz-12345")
+        msg = str(exc_info.value)
+        assert "OCS GET" in msg
+        assert "User does not exist" in msg
+
+
 class TestErrorsThroughMcpTools:
     """Verify that errors propagate correctly through the MCP tool layer."""
 
@@ -78,6 +117,11 @@ async def test_delete_not_found_via_tool(self, nc_mcp: McpTestHelper) -> None:
             await nc_mcp.call("delete_file", path="nonexistent-12345.txt")
 
     @pytest.mark.asyncio
-    async def test_get_nonexistent_user_via_tool(self, nc_mcp: McpTestHelper) -> None:
-        with pytest.raises(ToolError):
+    async def test_get_nonexistent_user_ocs_message_via_tool(self, nc_mcp: McpTestHelper) -> None:
+        with pytest.raises(ToolError, match="User does not exist"):
             await nc_mcp.call("get_user", user_id="nonexistent-user-xyz-12345")
+
+    @pytest.mark.asyncio
+    async def test_get_nonexistent_share_ocs_message_via_tool(self, nc_mcp: McpTestHelper) -> None:
+        with pytest.raises(ToolError, match="Wrong share ID"):
+            await nc_mcp.call("get_share", share_id=999999)
diff --git a/tests/test_client_errors.py b/tests/test_client_errors.py
@@ -0,0 +1,109 @@
+"""Tests for HTTP client error handling — _raise_for_status and _raise_for_ocs_status."""
+
+import json
+from typing import Any
+
+import niquests
+import pytest
+
+from nextcloud_mcp.client import NextcloudError, _raise_for_ocs_status, _raise_for_status
+
+
+def _fake_response(status_code: int, body: dict[str, Any] | str | None = None) -> niquests.Response:
+    """Build a minimal niquests.Response with the given status and optional JSON body."""
+    resp = niquests.Response()
+    resp.status_code = status_code
+    if body is not None:
+        text = json.dumps(body) if isinstance(body, dict) else body
+        resp._content = text.encode("utf-8")
+        resp.headers["Content-Type"] = "application/json"
+    return resp
+
+
+def _ocs_error_body(message: str, statuscode: int = 404) -> dict[str, Any]:
+    return {"ocs": {"meta": {"status": "failure", "statuscode": statuscode, "message": message}, "data": []}}
+
+
+class TestRaiseForStatus:
+    def test_ok_response_does_not_raise(self) -> None:
+        _raise_for_status(_fake_response(200))
+
+    def test_404_generic_message(self) -> None:
+        with pytest.raises(NextcloudError, match="Not found") as exc_info:
+            _raise_for_status(_fake_response(404))
+        assert exc_info.value.status_code == 404
+
+    def test_401_auth_message(self) -> None:
+        with pytest.raises(NextcloudError, match="Authentication failed"):
+            _raise_for_status(_fake_response(401))
+
+    def test_403_forbidden_message(self) -> None:
+        with pytest.raises(NextcloudError, match="Forbidden"):
+            _raise_for_status(_fake_response(403))
+
+    def test_409_conflict_message(self) -> None:
+        with pytest.raises(NextcloudError, match="Conflict"):
+            _raise_for_status(_fake_response(409))
+
+    def test_423_locked_message(self) -> None:
+        with pytest.raises(NextcloudError, match="Locked"):
+            _raise_for_status(_fake_response(423))
+
+    def test_unmapped_code_shows_http_number(self) -> None:
+        with pytest.raises(NextcloudError, match="HTTP 418"):
+            _raise_for_status(_fake_response(418))
+
+    def test_context_prefix(self) -> None:
+        with pytest.raises(NextcloudError, match=r"Delete 'file\.txt': Not found"):
+            _raise_for_status(_fake_response(404), context="Delete 'file.txt'")
+
+
+class TestRaiseForOcsStatus:
+    def test_ok_response_does_not_raise(self) -> None:
+        _raise_for_ocs_status(_fake_response(200))
+
+    def test_extracts_ocs_meta_message(self) -> None:
+        with pytest.raises(NextcloudError, match="User does not exist") as exc_info:
+            _raise_for_ocs_status(_fake_response(404, _ocs_error_body("User does not exist")))
+        assert exc_info.value.status_code == 404
+
+    def test_ocs_message_with_context_prefix(self) -> None:
+        body = _ocs_error_body("User already exists", 102)
+        with pytest.raises(NextcloudError, match=r"OCS POST cloud/users: User already exists") as exc_info:
+            _raise_for_ocs_status(_fake_response(400, body), "OCS POST cloud/users")
+        assert exc_info.value.status_code == 400
+
+    def test_share_not_found_message(self) -> None:
+        body = _ocs_error_body("Wrong share ID, share does not exist")
+        with pytest.raises(NextcloudError, match="Wrong share ID, share does not exist"):
+            _raise_for_ocs_status(_fake_response(404, body))
+
+    def test_share_wrong_path_message(self) -> None:
+        body = _ocs_error_body("Wrong path, file/folder does not exist")
+        with pytest.raises(NextcloudError, match="Wrong path"):
+            _raise_for_ocs_status(_fake_response(404, body))
+
+    def test_falls_back_when_ocs_message_empty(self) -> None:
+        with pytest.raises(NextcloudError, match="HTTP 400"):
+            _raise_for_ocs_status(_fake_response(400, _ocs_error_body("", 400)))
+
+    def test_falls_back_when_body_not_json(self) -> None:
+        with pytest.raises(NextcloudError, match="Not found"):
+            _raise_for_ocs_status(_fake_response(404, "<html>404</html>"))
+
+    def test_falls_back_when_body_missing_ocs_key(self) -> None:
+        with pytest.raises(NextcloudError, match="Not found"):
+            _raise_for_ocs_status(_fake_response(404, {"error": "something"}))
+
+    def test_falls_back_when_body_missing_meta(self) -> None:
+        with pytest.raises(NextcloudError, match="HTTP 400"):
+            _raise_for_ocs_status(_fake_response(400, {"ocs": {"data": []}}))
+
+    def test_falls_back_when_no_body(self) -> None:
+        with pytest.raises(NextcloudError, match="HTTP 500"):
+            _raise_for_ocs_status(_fake_response(500))
+
+    def test_preserves_status_code(self) -> None:
+        with pytest.raises(NextcloudError) as exc_info:
+            _raise_for_ocs_status(_fake_response(400, _ocs_error_body("User already exists", 102)))
+        assert exc_info.value.status_code == 400
