bauer/cordra-mcp
1
0
Fork 0
mirror of https://github.com/dnlbauer/cordra-mcp.git synced 2026-09-10 21:55:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: linter and unit tests

Browse Source
This commit is contained in:
Daniel Bauer
2025-06-30 11:44:25 +02:00
parent f34d3e6e7c
commit 9901f51f06
4 changed files with 130 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
src/mcp_cordra/client.py
View File

@@ -42,14 +42,14 @@ class CordraClient:
def __init__(self, config: CordraConfig) -> None:
"""Initialize the Cordra client.
Args:
config: Configuration settings for the Cordra connection
"""
self.config = config
self.session = requests.Session()
self.session.verify = config.verify_ssl
# Set up authentication
if config.username and config.password:
self.session.auth = (config.username, config.password)
@@ -58,14 +58,14 @@ class CordraClient:
def _handle_http_error(self, response: requests.Response, context: str) -> None:
"""Handle HTTP errors and raise appropriate exceptions.
Args:
response: The HTTP response object
context: Context description for the error message
Raises:
CordraNotFoundError: For 404 errors
CordraAuthenticationError: For 401/403 errors
CordraAuthenticationError: For 401/403 errors
CordraClientError: For other HTTP errors
"""
status_code = response.status_code
@@ -80,13 +80,13 @@ class CordraClient:
async def get_object(self, object_id: str) -> DigitalObject:
"""Retrieve a digital object by its ID.
Args:
object_id: The unique identifier of the object to retrieve
Returns:
The full digital object
Raises:
ValueError: If object_id is empty
CordraNotFoundError: If the object is not found
@@ -95,15 +95,15 @@ class CordraClient:
"""
url = f"{self.config.cordra_url}/objects/{object_id}"
params = {"full": "true"}
try:
response = self.session.get(url, params=params, timeout=self.config.timeout)
if not response.ok:
self._handle_http_error(response, f"Failed to retrieve object {object_id}")
cordra_obj = response.json()
return DigitalObject(
id=object_id,
type=cordra_obj.get('type', ''),
@@ -112,19 +112,19 @@ class CordraClient:
acl=cordra_obj.get('acl'),
payloads=cordra_obj.get('payloads'),
)
except requests.RequestException as e:
raise CordraClientError(f"Failed to retrieve object {object_id}: {e}") from e
async def find(self, query: str) -> list[dict[str, Any]]:
"""Find objects using a Cordra query.
Args:
query: The query string to search for objects
Returns:
List of objects matching the query as dictionaries
Raises:
ValueError: If query is empty
CordraAuthenticationError: If authentication fails
@@ -132,33 +132,33 @@ class CordraClient:
"""
url = f"{self.config.cordra_url}/search"
params = {"query": query}
try:
response = self.session.get(url, params=params, timeout=self.config.timeout)
if not response.ok:
self._handle_http_error(response, f"Failed to search with query '{query}'")
search_result = response.json()
# Extract the results array from the response
if isinstance(search_result, dict) and 'results' in search_result:
return search_result['results']
else:
return []
except requests.RequestException as e:
raise CordraClientError(f"Failed to search with query '{query}': {e}") from e
async def get_schema(self, schema_name: str) -> DigitalObject:
"""Retrieve a schema definition by its name.
Args:
schema_name: The name of the schema to retrieve
Returns:
The schema object containing the type definition
Raises:
CordraNotFoundError: If the schema is not found
CordraAuthenticationError: If authentication fails
@@ -166,20 +166,20 @@ class CordraClient:
"""
# Search for the specific schema by name using correct query format
query = f"type:Schema AND /name:{schema_name}"
try:
schemas = await self.find(query)
if not schemas:
raise CordraNotFoundError(f"Schema '{schema_name}' not found")
# Get the first matching schema (should be unique by name)
schema_data = schemas[0]
# Get the full schema object using its ID
return await self.get_object(schema_data['id'])
except (CordraNotFoundError, CordraAuthenticationError):
raise
except Exception as e:
raise CordraClientError(f"Failed to retrieve schema '{schema_name}': {e}") from e
raise CordraClientError(f"Failed to retrieve schema '{schema_name}': {e}") from e

10
src/mcp_cordra/server.py
View File

@@ -8,10 +8,10 @@ from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.resources import FunctionResource
from .client import (
CordraAuthenticationError,
CordraClient,
CordraClientError,
CordraNotFoundError,
CordraAuthenticationError,
)
from .config import CordraConfig
@@ -52,8 +52,8 @@ async def get_cordra_object(prefix: str, suffix: str) -> str:
except ValueError as e:
raise RuntimeError(f"Invalid parameters: {e}") from e
except CordraNotFoundError:
raise RuntimeError(f"Object not found: {object_id}")
except CordraNotFoundError as e:
raise RuntimeError(f"Object not found: {object_id}") from e
except CordraAuthenticationError as e:
raise RuntimeError(f"Authentication failed: {e}") from e
except CordraClientError as e:
@@ -66,8 +66,8 @@ async def create_schema_resource(schema_name: str) -> str:
schema_object = await cordra_client.get_schema(schema_name)
schema_dict = schema_object.model_dump()
return json.dumps(schema_dict, indent=2)
except CordraNotFoundError:
raise RuntimeError(f"Schema not found: {schema_name}")
except CordraNotFoundError as e:
raise RuntimeError(f"Schema not found: {schema_name}") from e
except CordraAuthenticationError as e:
raise RuntimeError(f"Authentication failed: {e}") from e
except CordraClientError as e:

21
tests/test_client.py
View File

@@ -8,7 +8,6 @@ from mcp_cordra.client import (
CordraClient,
CordraClientError,
CordraNotFoundError,
CordraAuthenticationError,
DigitalObject,
)
from mcp_cordra.config import CordraConfig
@@ -47,7 +46,7 @@ def mock_cordra_object():
"mediaType": "text/plain"
},
{
"name": "file2.pdf",
"name": "file2.pdf",
"filename": "file2.pdf",
"size": 2048,
"mediaType": "application/pdf"
@@ -175,14 +174,14 @@ class TestCordraClient:
mock_response.status_code = 200
mock_response.json.return_value = mock_response_data
mock_response.raise_for_status.return_value = None
result = await client.find("type:Schema")
assert len(result) == 3
assert result[0]["name"] == "User"
assert result[1]["name"] == "Project"
assert result[2]["name"] == "Document"
mock_get.assert_called_once_with(
"https://test.example.com/search",
params={"query": "type:Schema"},
@@ -197,9 +196,9 @@ class TestCordraClient:
mock_response.status_code = 200
mock_response.json.return_value = mock_response_data
mock_response.raise_for_status.return_value = None
result = await client.find("type:NonExistent")
assert result == []
mock_get.assert_called_once_with(
"https://test.example.com/search",
@@ -215,9 +214,9 @@ class TestCordraClient:
mock_response.status_code = 200
mock_response.json.return_value = mock_response_data
mock_response.raise_for_status.return_value = None
result = await client.find("type:Schema")
assert result == []
@patch('mcp_cordra.client.requests.Session.get')
@@ -225,10 +224,10 @@ class TestCordraClient:
"""Test find error handling."""
from requests import RequestException
mock_get.side_effect = RequestException("Search failed")
with pytest.raises(CordraClientError) as exc_info:
await client.find("invalid:query")
assert "Failed to search with query 'invalid:query'" in str(exc_info.value)
assert "Search failed" in str(exc_info.value)

185
tests/test_server.py
View File

@@ -40,9 +40,9 @@ class TestGetCordraObject:
async def test_get_object_success(self, mock_client, sample_digital_object):
"""Test successful object retrieval."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
result = await get_cordra_object("people", "john-doe-123")
# Verify the result is valid JSON
parsed_result = json.loads(result)
assert parsed_result["id"] == "people/john-doe-123"
@@ -52,7 +52,7 @@ class TestGetCordraObject:
assert parsed_result["metadata"]["created"] == "2023-01-01"
assert len(parsed_result["payloads"]) == 1
assert parsed_result["payloads"][0]["name"] == "profile_photo"
# Verify the client was called with the correct object ID
mock_client.get_object.assert_called_once_with("people/john-doe-123")
@@ -62,10 +62,10 @@ class TestGetCordraObject:
mock_client.get_object = AsyncMock(
side_effect=CordraNotFoundError("Object not found: people/nonexistent")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_object("people", "nonexistent")
assert "Object not found: people/nonexistent" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("people/nonexistent")
@@ -75,10 +75,10 @@ class TestGetCordraObject:
mock_client.get_object = AsyncMock(
side_effect=CordraClientError("Connection failed")
)
with pytest.raises(RuntimeError) as exc_info:
await get_cordra_object("people", "john-doe-123")
assert "Failed to retrieve object people/john-doe-123" in str(exc_info.value)
assert "Connection failed" in str(exc_info.value)
mock_client.get_object.assert_called_once_with("people/john-doe-123")
@@ -87,14 +87,14 @@ class TestGetCordraObject:
async def test_object_id_construction(self, mock_client, sample_digital_object):
"""Test that object ID is correctly constructed from prefix and suffix."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
# Test various prefix/suffix combinations
test_cases = [
("people", "john-doe-123", "people/john-doe-123"),
("documents", "report-2023", "documents/report-2023"),
("items", "item_with_underscores", "items/item_with_underscores"),
]
for prefix, suffix, expected_id in test_cases:
await get_cordra_object(prefix, suffix)
mock_client.get_object.assert_called_with(expected_id)
@@ -103,16 +103,16 @@ class TestGetCordraObject:
async def test_json_formatting(self, mock_client, sample_digital_object):
"""Test that the returned JSON is properly formatted."""
mock_client.get_object = AsyncMock(return_value=sample_digital_object)
result = await get_cordra_object("people", "john-doe-123")
# Verify it's valid JSON with proper indentation
parsed_result = json.loads(result)
assert isinstance(parsed_result, dict)
# Check that the result contains indentation (pretty-printed)
assert " " in result # Should have 2-space indentation
# Verify all expected fields are present
assert "id" in parsed_result
assert "type" in parsed_result
@@ -133,10 +133,10 @@ class TestGetCordraObject:
payloads=None
)
mock_client.get_object = AsyncMock(return_value=minimal_object)
result = await get_cordra_object("test", "minimal")
parsed_result = json.loads(result)
assert parsed_result["id"] == "test/minimal"
assert parsed_result["type"] == ""
assert parsed_result["content"]["id"] == "test/minimal"
@@ -145,103 +145,88 @@ class TestGetCordraObject:
assert parsed_result["payloads"] is None
class TestListCordraSchemas:
"""Test the list_cordra_schemas resource handler."""
class TestSchemaResourceFunctions:
"""Test the schema resource functions."""
@patch('mcp_cordra.server.cordra_client')
async def test_list_schemas_success(self, mock_client):
"""Test successful schema listing."""
mock_schemas = [
{"content": {"name": "User"}, "identifier": "test/user-schema"},
{"content": {"name": "Project"}, "identifier": "test/project-schema"},
{"content": {"name": "Document"}, "identifier": "test/doc-schema"},
{"content": {"name": "CaptureEvent"}, "identifier": "test/capture-schema"}
]
mock_client.find = AsyncMock(return_value=mock_schemas)
from mcp_cordra.server import list_cordra_schemas
result = await list_cordra_schemas()
async def test_create_schema_resource_success(self, mock_client):
"""Test successful schema resource creation."""
mock_schema = DigitalObject(
id="test/user-schema",
type="Schema",
content={"name": "User", "type": "object", "properties": {}}
)
mock_client.get_schema = AsyncMock(return_value=mock_schema)
from mcp_cordra.server import create_schema_resource
result = await create_schema_resource("User")
# Verify the result is valid JSON
parsed_result = json.loads(result)
assert "schemas" in parsed_result
assert "count" in parsed_result
assert parsed_result["count"] == 4
assert "User" in parsed_result["schemas"]
assert "Project" in parsed_result["schemas"]
assert "Document" in parsed_result["schemas"]
assert "CaptureEvent" in parsed_result["schemas"]
assert parsed_result["id"] == "test/user-schema"
assert parsed_result["type"] == "Schema"
assert parsed_result["content"]["name"] == "User"
# Verify the client was called with correct schema name
mock_client.get_schema.assert_called_once_with("User")
@patch('mcp_cordra.server.cordra_client')
async def test_create_schema_resource_not_found(self, mock_client):
"""Test schema resource creation with schema not found."""
mock_client.get_schema = AsyncMock(side_effect=CordraNotFoundError("Schema not found"))
from mcp_cordra.server import create_schema_resource
with pytest.raises(RuntimeError) as exc_info:
await create_schema_resource("NonExistent")
assert "Schema not found: NonExistent" in str(exc_info.value)
mock_client.get_schema.assert_called_once_with("NonExistent")
@patch('mcp_cordra.server.cordra_client')
async def test_register_schema_resources_success(self, mock_client):
"""Test successful schema resource registration."""
mock_schemas = [
{"content": {"name": "User"}, "id": "test/user-schema"},
{"content": {"name": "Project"}, "id": "test/project-schema"},
{"content": {"name": "Document"}, "id": "test/doc-schema"}
]
mock_client.find = AsyncMock(return_value=mock_schemas)
# Mock the mcp.add_resource method
with patch('mcp_cordra.server.mcp') as mock_mcp:
from mcp_cordra.server import register_schema_resources
await register_schema_resources()
# Verify the client was called with correct query
mock_client.find.assert_called_once_with("type:Schema")
# Verify add_resource was called for each schema
assert mock_mcp.add_resource.call_count == 3
@patch('mcp_cordra.server.cordra_client')
async def test_list_schemas_empty(self, mock_client):
"""Test schema listing with no results."""
mock_client.find = AsyncMock(return_value=[])
from mcp_cordra.server import list_cordra_schemas
result = await list_cordra_schemas()
parsed_result = json.loads(result)
assert parsed_result["schemas"] == []
assert parsed_result["count"] == 0
mock_client.find.assert_called_once_with("type:Schema")
@patch('mcp_cordra.server.cordra_client')
async def test_list_schemas_missing_name_field(self, mock_client):
"""Test schema listing with objects missing name field."""
async def test_register_schema_resources_missing_name(self, mock_client):
"""Test schema resource registration with objects missing name field."""
mock_schemas = [
{"content": {"name": "User"}, "identifier": "test/user-schema"},
{"content": {}, "identifier": "test/no-name-schema"}, # Missing name field
{"content": {"name": "Project"}, "identifier": "test/project-schema"},
{"content": {"other": "field"}} # No name or identifier
{"content": {"name": "User"}, "id": "test/user-schema"},
{"content": {}, "id": "test/no-name-schema"}, # Missing name field
{"content": {"name": "Project"}, "id": "test/project-schema"}
]
mock_client.find = AsyncMock(return_value=mock_schemas)
from mcp_cordra.server import list_cordra_schemas
result = await list_cordra_schemas()
parsed_result = json.loads(result)
assert parsed_result["count"] == 2 # Only objects with name field
assert "User" in parsed_result["schemas"]
assert "Project" in parsed_result["schemas"]
assert len(parsed_result["schemas"]) == 2
with patch('mcp_cordra.server.mcp') as mock_mcp:
from mcp_cordra.server import register_schema_resources
await register_schema_resources()
# Only 2 schemas should be registered (those with name field)
assert mock_mcp.add_resource.call_count == 2
@patch('mcp_cordra.server.cordra_client')
async def test_list_schemas_client_error(self, mock_client):
"""Test schema listing with client error."""
from mcp_cordra.client import CordraClientError
async def test_register_schema_resources_client_error(self, mock_client):
"""Test schema resource registration with client error."""
mock_client.find = AsyncMock(side_effect=CordraClientError("Search failed"))
from mcp_cordra.server import list_cordra_schemas
with pytest.raises(RuntimeError) as exc_info:
await list_cordra_schemas()
assert "Failed to list schemas" in str(exc_info.value)
assert "Search failed" in str(exc_info.value)
@patch('mcp_cordra.server.cordra_client')
async def test_list_schemas_json_format(self, mock_client):
"""Test that the returned JSON is properly formatted."""
mock_schemas = [
{"content": {"name": "TestSchema"}, "identifier": "test/schema"}
]
mock_client.find = AsyncMock(return_value=mock_schemas)
from mcp_cordra.server import list_cordra_schemas
result = await list_cordra_schemas()
# Verify it's valid JSON with proper indentation
parsed_result = json.loads(result)
assert isinstance(parsed_result, dict)
# Check that the result contains indentation (pretty-printed)
assert " " in result # Should have 2-space indentation
# Verify expected structure
assert "schemas" in parsed_result
assert "count" in parsed_result
assert isinstance(parsed_result["schemas"], list)
assert isinstance(parsed_result["count"], int)
# Should not raise an exception, just log a warning
from mcp_cordra.server import register_schema_resources
await register_schema_resources() # Should complete without raising
mock_client.find.assert_called_once_with("type:Schema")