mirror of
https://github.com/wso2/fhir-mcp-server.git
synced 2025-11-14 22:18:14 +03:00
Refactor: fix failing test cases
This commit is contained in:
@@ -46,7 +46,7 @@ class ServerConfigs(BaseSettings):
|
||||
server_client_id: str = ""
|
||||
server_client_secret: str = ""
|
||||
server_scopes: str = ""
|
||||
server_base_url: str
|
||||
server_base_url: str = ""
|
||||
server_access_token: str | None = None
|
||||
|
||||
def callback_url(
|
||||
|
||||
@@ -290,7 +290,7 @@ def register_mcp_tools(mcp: FastMCP) -> None:
|
||||
async_resources: list[Any] = (
|
||||
await client.resources(type).search(Raw(**searchParam)).fetch_raw()
|
||||
)
|
||||
logger.debug("Async resources fetched:", async_resources) # Debugging line
|
||||
logger.debug("Async resources fetched:", async_resources)
|
||||
return async_resources
|
||||
except ValueError as ex:
|
||||
logger.exception(
|
||||
|
||||
@@ -135,8 +135,8 @@ async def test_tool_search(mcp_server, patient_id):
|
||||
response: Dict = await extract_resource(tool_result)
|
||||
assert (
|
||||
response is not None
|
||||
and response.get("resourceType") == "Patient"
|
||||
and response.get("id") == patient_id
|
||||
and response.get("entry")[0].get("resource").get("resourceType") == "Patient"
|
||||
and response.get("entry")[0].get("resource").get("id") == patient_id
|
||||
), f"No Patient resource in read result: {response}"
|
||||
except asyncio.TimeoutError as ex:
|
||||
logger.error(
|
||||
@@ -146,6 +146,38 @@ async def test_tool_search(mcp_server, patient_id):
|
||||
raise
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_search_condition_count(mcp_server):
|
||||
request_payload = {
|
||||
"type": "Condition",
|
||||
"searchParam": {
|
||||
"code": "http://snomed.info/sct|204256004",
|
||||
"_summary": "count",
|
||||
"_total": "estimate"
|
||||
}
|
||||
}
|
||||
logger.info("[TEST REQUEST] search Condition count:", request_payload)
|
||||
try:
|
||||
async with create_mcp_session() as mcp_session:
|
||||
tool_result: types.CallToolResult = await mcp_session.call_tool(
|
||||
name="search", arguments=request_payload
|
||||
)
|
||||
response: Dict = await extract_resource(tool_result)
|
||||
assert response.get("resourceType") == "Bundle", f"Not a Bundle: {response}"
|
||||
assert response.get("type") == "searchset", f"Not a searchset: {response}"
|
||||
assert "total" in response, f"No total count in response: {response}"
|
||||
assert isinstance(response["total"], int), f"Total is not int: {response}"
|
||||
# Optionally check for SUBSETTED tag
|
||||
tags = response.get("meta", {}).get("tag", [])
|
||||
assert any(tag.get("code") == "SUBSETTED" for tag in tags), f"Missing SUBSETTED tag: {tags}"
|
||||
except asyncio.TimeoutError as ex:
|
||||
logger.error(
|
||||
"[TOOL RESPONSE] Timeout waiting for Condition count search response from MCP server",
|
||||
exc_info=ex,
|
||||
)
|
||||
raise
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_update(mcp_server, patient_id):
|
||||
request_payload = {
|
||||
|
||||
@@ -24,7 +24,6 @@ from fhir_mcp_server.oauth.common import (
|
||||
discover_oauth_metadata,
|
||||
is_token_expired,
|
||||
get_endpoint,
|
||||
handle_successful_authentication,
|
||||
handle_failed_authentication,
|
||||
generate_code_verifier,
|
||||
generate_code_challenge,
|
||||
@@ -230,14 +229,6 @@ class TestGetEndpoint:
|
||||
class TestHandleAuthentication:
|
||||
"""Test the authentication handling functions."""
|
||||
|
||||
def test_handle_successful_authentication(self):
|
||||
"""Test successful authentication response."""
|
||||
response: HTMLResponse = handle_successful_authentication()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Authentication Successful!" in bytes(response.body).decode()
|
||||
assert "text/html" in response.media_type
|
||||
|
||||
def test_handle_failed_authentication_default(self):
|
||||
"""Test failed authentication response with default message."""
|
||||
response = handle_failed_authentication()
|
||||
|
||||
@@ -105,11 +105,11 @@ class TestOAuthServerProvider:
|
||||
|
||||
# Retrieve client
|
||||
retrieved_client: OAuthClientInformationFull | None = await provider.get_client(
|
||||
"test_client_id"
|
||||
"fhir_mcp_test_client_id"
|
||||
)
|
||||
assert retrieved_client is not None
|
||||
assert retrieved_client == client_info
|
||||
assert retrieved_client.client_id == "test_client_id"
|
||||
assert retrieved_client.client_id == "fhir_mcp_test_client_id"
|
||||
|
||||
# Test non-existent client
|
||||
non_existent = await provider.get_client("non_existent")
|
||||
|
||||
Reference in New Issue
Block a user