test_main.py•6.79 kB
# tests/test_main.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch # For mocking our service function
# Import your FastAPI app instance and Pydantic models/custom exceptions
# Adjust the import path based on your project structure if needed.
# Assuming your tests directory is at the same level as your 'app' directory:
from app.main import app # Your FastAPI application instance
from app.models import MCPResponse, MCPWeatherData # For asserting response structure
from app.services import WeatherServiceError # For simulating service errors
# Create a TestClient instance using your FastAPI app
# This client allows you to send requests to your app without running a live Uvicorn server.
client = TestClient(app)
# --- Test for the Root Endpoint ---
def test_read_root():
"""Test the root GET / endpoint."""
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Welcome to the MCP Weather Server! See /docs for API details."}
# --- Tests for the /mcp/weather Endpoint ---
# Define some sample valid and invalid MCP request payloads for testing
VALID_MCP_REQUEST_PAYLOAD = {
"protocol_version": "1.0",
"tool_id": "weather_tool",
"method": "get_current_weather",
"parameters": {
"location": "TestCity,TC"
}
}
INVALID_TOOL_ID_PAYLOAD = {
"protocol_version": "1.0",
"tool_id": "wrong_tool",
"method": "get_current_weather",
"parameters": {
"location": "TestCity,TC"
}
}
# Test Case 1: Successful weather request (mocking the service layer)
@patch("app.main.fetch_weather_data_from_provider") # Path to the function in app.main
def test_get_weather_mcp_success(mock_fetch_weather):
"""
Test the /mcp/weather endpoint for a successful scenario.
The fetch_weather_data_from_provider service is mocked.
"""
# Configure the mock to return a successful weather data dictionary
mock_successful_service_response = {
"location": "TestCity",
"temperature_celsius": 20.0,
"temperature_fahrenheit": 68.0,
"condition": "Sunny",
"description": "Clear sky",
"humidity_percent": 50.0,
"wind_kph": 10.0,
"pressure_hpa": 1012.0
}
mock_fetch_weather.return_value = mock_successful_service_response
# Send a request to the endpoint
response = client.post("/mcp/weather", json=VALID_MCP_REQUEST_PAYLOAD)
# Assertions
assert response.status_code == 200 # MCP standard is to return 200 even for app errors
response_data = response.json()
assert response_data["protocol_version"] == VALID_MCP_REQUEST_PAYLOAD["protocol_version"]
assert response_data["tool_id"] == VALID_MCP_REQUEST_PAYLOAD["tool_id"]
assert response_data["status"] == "success"
assert response_data["error_message"] is None
# Assert the structure and content of the 'data' field
assert response_data["data"] is not None
mcp_weather_data = MCPWeatherData(**response_data["data"]) # Validate with Pydantic model
assert mcp_weather_data.location == "TestCity"
assert mcp_weather_data.temperature_celsius == 20.0
# Ensure the mock was called correctly (with a positional argument)
mock_fetch_weather.assert_called_once_with("TestCity,TC")
# Test Case 2: Service layer returns a WeatherServiceError (e.g., location not found)
@patch("app.main.fetch_weather_data_from_provider")
def test_get_weather_mcp_service_error_location_not_found(mock_fetch_weather):
"""
Test /mcp/weather when the service layer raises a WeatherServiceError (location not found).
"""
# Configure the mock to raise a WeatherServiceError
error_message = "Weather data not found for location: TestCity,TC."
mock_fetch_weather.side_effect = WeatherServiceError(error_message, status_code=404)
response = client.post("/mcp/weather", json=VALID_MCP_REQUEST_PAYLOAD)
assert response.status_code == 200
response_data = response.json()
assert response_data["status"] == "error"
assert response_data["data"] is None
assert response_data["error_message"] == error_message
# Ensure the mock was called correctly (with a positional argument)
mock_fetch_weather.assert_called_once_with("TestCity,TC")
# Test Case 3: Service layer returns an unexpected error
@patch("app.main.fetch_weather_data_from_provider")
def test_get_weather_mcp_service_unexpected_error(mock_fetch_weather):
"""
Test /mcp/weather when the service layer raises an unexpected generic Exception.
"""
mock_fetch_weather.side_effect = Exception("A wild generic error appeared!")
response = client.post("/mcp/weather", json=VALID_MCP_REQUEST_PAYLOAD)
assert response.status_code == 200
response_data = response.json()
assert response_data["status"] == "error"
assert response_data["data"] is None
assert response_data["error_message"] == "An unexpected internal server error occurred while processing your request."
# Ensure the mock was called correctly (with a positional argument)
mock_fetch_weather.assert_called_once_with("TestCity,TC")
# Test Case 4: Invalid tool_id in the request payload
def test_get_weather_mcp_invalid_tool_id():
"""Test /mcp/weather with an invalid tool_id in the request."""
response = client.post("/mcp/weather", json=INVALID_TOOL_ID_PAYLOAD)
assert response.status_code == 200
response_data = response.json()
assert response_data["status"] == "error"
assert response_data["error_message"] == "Invalid tool_id 'wrong_tool'. This endpoint is dedicated to 'weather_tool'."
# You can add more tests for:
# - Invalid method
# - Malformed request payloads (e.g., missing 'parameters' or 'location')
# FastAPI and Pydantic will handle some of these automatically, returning 422 Unprocessable Entity.
# You can write tests to confirm that behavior if desired.
# Example of testing for a 422 Unprocessable Entity error from Pydantic validation
def test_get_weather_mcp_missing_location_parameter():
"""Test /mcp/weather when 'location' is missing from parameters."""
payload_missing_location = {
"protocol_version": "1.0",
"tool_id": "weather_tool",
"method": "get_current_weather",
"parameters": {} # Missing 'location'
}
response = client.post("/mcp/weather", json=payload_missing_location)
assert response.status_code == 422 # FastAPI/Pydantic validation error
# The response body for 422 errors has a specific "detail" structure.
# You can assert its contents if needed, e.g., checking that it mentions 'location'.
assert "detail" in response.json()
# Example: assert any("location" in err["loc"] for err in response.json()["detail"])