test_services.py•7.19 kB
# tests/test_services.py
import pytest
import httpx # For creating mock response objects and error types
import json # For json.loads
from unittest.mock import patch, AsyncMock, MagicMock # MagicMock for synchronous methods
# Import the function to test and the custom exception
from app.services import fetch_weather_data_from_provider, WeatherServiceError
from app.core.config import get_settings # To ensure settings are loaded for API key check
# Ensure settings are loaded for tests that might check for API key presence
settings = get_settings()
@pytest.fixture
def mock_successful_openweathermap_response_data():
"""Provides a sample successful JSON response from OpenWeatherMap."""
return {
"coord": {"lon": -0.1276, "lat": 51.5074},
"weather": [{"id": 800, "main": "Clear", "description": "clear sky", "icon": "01d"}],
"base": "stations",
"main": {"temp": 20.0, "feels_like": 19.5, "temp_min": 18.0, "temp_max": 22.0, "pressure": 1012, "humidity": 50},
"visibility": 10000,
"wind": {"speed": 5.0, "deg": 180}, # speed in meter/sec
"clouds": {"all": 0},
"dt": 1678886400,
"sys": {"type": 1, "id": 1414, "country": "GB", "sunrise": 1678857600, "sunset": 1678899600},
"timezone": 0,
"id": 2643743,
"name": "London",
"cod": 200
}
@pytest.mark.asyncio
async def test_fetch_weather_success(mock_successful_openweathermap_response_data):
mock_response = AsyncMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = mock_successful_openweathermap_response_data
mock_response.raise_for_status = MagicMock() # Synchronous, does nothing for 200
with patch("app.services.httpx.AsyncClient") as mock_async_client_constructor:
mock_client_instance = AsyncMock()
mock_client_instance.get.return_value = mock_response
mock_async_client_constructor.return_value.__aenter__.return_value = mock_client_instance
location_to_test = "London"
processed_data = await fetch_weather_data_from_provider(location_to_test)
assert processed_data["location"] == "London"
assert processed_data["temperature_celsius"] == 20.0
assert processed_data["temperature_fahrenheit"] == (20.0 * 9/5) + 32
assert processed_data["condition"] == "Clear"
assert processed_data["description"] == "Clear sky"
assert processed_data["humidity_percent"] == 50.0
assert processed_data["wind_kph"] == (5.0 * 3.6)
assert processed_data["pressure_hpa"] == 1012.0
mock_client_instance.get.assert_called_once()
args, kwargs = mock_client_instance.get.call_args
assert "http://api.openweathermap.org/data/2.5/weather" in args[0]
assert kwargs["params"]["q"] == location_to_test
assert kwargs["params"]["appid"] == settings.OPENWEATHERMAP_API_KEY
@pytest.mark.asyncio
async def test_fetch_weather_api_key_not_configured():
original_api_key = settings.OPENWEATHERMAP_API_KEY
settings.OPENWEATHERMAP_API_KEY = ""
with pytest.raises(WeatherServiceError) as excinfo:
await fetch_weather_data_from_provider("SomeCity")
assert "Weather service is not configured on the server." in str(excinfo.value)
assert excinfo.value.status_code == 500
settings.OPENWEATHERMAP_API_KEY = original_api_key
@pytest.mark.asyncio
@pytest.mark.parametrize(
"api_status_code, api_response_text_str, expected_error_message, expected_status_code",
[
(401, '{"message": "Invalid API key"}', "Invalid API key or subscription issue with the weather service.", 401),
(404, '{"message": "city not found"}', "Weather data not found for location: TestCity.", 404),
(429, '{"message": "Too many requests"}', "Rate limit exceeded with the weather service. Please try again later.", 429),
(500, '{"message": "Internal server error"}', "Error fetching data from weather service: HTTP 500.", 500),
]
)
async def test_fetch_weather_http_status_errors(
api_status_code, api_response_text_str, expected_error_message, expected_status_code
):
mock_httpx_response_for_error = httpx.Response(
status_code=api_status_code,
request=httpx.Request("GET", "http://mockurl"), # A dummy request object
content=api_response_text_str.encode('utf-8') # httpx.Response needs content as bytes
)
# We want the actual raise_for_status() to be called on this mock response
# so that it raises the httpx.HTTPStatusError that our service code expects.
# The service code then catches this httpx.HTTPStatusError.
with patch("app.services.httpx.AsyncClient") as mock_async_client_constructor:
mock_client_instance = AsyncMock()
# Configure the 'get' method to return our specially crafted error response
mock_client_instance.get.return_value = mock_httpx_response_for_error
mock_async_client_constructor.return_value.__aenter__.return_value = mock_client_instance
with pytest.raises(WeatherServiceError) as excinfo:
await fetch_weather_data_from_provider("TestCity")
assert expected_error_message in str(excinfo.value)
assert excinfo.value.status_code == expected_status_code
@pytest.mark.asyncio
async def test_fetch_weather_network_error():
with patch("app.services.httpx.AsyncClient") as mock_async_client_constructor:
mock_client_instance = AsyncMock()
mock_client_instance.get.side_effect = httpx.ConnectError("Mocked connection error")
mock_async_client_constructor.return_value.__aenter__.return_value = mock_client_instance
with pytest.raises(WeatherServiceError) as excinfo:
await fetch_weather_data_from_provider("TestCity")
assert "Could not connect to the weather service." in str(excinfo.value)
assert excinfo.value.status_code == 503
@pytest.mark.asyncio
async def test_fetch_weather_incomplete_api_data():
incomplete_data = {"weather": [{"main": "Cloudy"}]}
mock_response = AsyncMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = incomplete_data
mock_response.raise_for_status = MagicMock() # Does nothing for 200
with patch("app.services.httpx.AsyncClient") as mock_async_client_constructor:
mock_client_instance = AsyncMock()
mock_client_instance.get.return_value = mock_response
mock_async_client_constructor.return_value.__aenter__.return_value = mock_client_instance
with pytest.raises(WeatherServiceError) as excinfo:
await fetch_weather_data_from_provider("TestCity")
# Check the message first
assert "Incomplete data received from weather API" in str(excinfo.value)
# Then check the status code if the message is as expected
# This means your app.services.py should raise WeatherServiceError with status_code=502
# specifically for this condition.
# If it's still 500, the generic handler in app/services.py is catching it.
assert excinfo.value.status_code == 502