f7df0f4900
- fetch_weather кэширует результат на 3600 сек (WEATHER_CACHE_TTL) - Провал запроса тоже кэшируется — не долбит API повторно - bypass_cache=True для форсирования свежего запроса - clear_weather_cache() для сброса кэша - asyncio.Lock для thread-safety - 5 новых тестов (cache hit, bypass, failure cache, clear, TTL expiry)
447 lines
16 KiB
Python
447 lines
16 KiB
Python
import pytest
|
||
import requests
|
||
from requests.exceptions import ConnectionError, Timeout, SSLError
|
||
from unittest.mock import patch, MagicMock
|
||
from utils.pogoda import fetch_weather, clear_weather_cache
|
||
|
||
|
||
class TestFetchWeather:
|
||
"""Тесты функции fetch_weather() — Яндекс Погода API с retry-логикой."""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_cache(self) -> None:
|
||
"""Очистить кэш перед каждым тестом."""
|
||
clear_weather_cache()
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _mock_api_key(self) -> None:
|
||
"""Мокаем _get_api_key для всех тестов в классе."""
|
||
with patch("utils.pogoda._get_api_key", return_value="test-key"):
|
||
yield
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_success(self, mock_get) -> None:
|
||
"""Успешный ответ должен вернуть данные в унифицированном формате."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 15,
|
||
"feels_like": 12,
|
||
"condition": "cloudy",
|
||
"wind_speed": 5.5,
|
||
"wind_gust": 8.0,
|
||
"wind_dir": "n",
|
||
"humidity": 65,
|
||
"pressure_mm": 735.0,
|
||
"pressure_pa": 980,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather()
|
||
|
||
assert result is not None
|
||
assert "current_condition" in result
|
||
assert result["current_condition"][0]["temp_C"] == 15
|
||
assert result["current_condition"][0]["FeelsLikeC"] == 12
|
||
assert result["current_condition"][0]["weatherDesc"] == [{"value": "Облачно"}]
|
||
assert result["current_condition"][0]["humidity"] == 65
|
||
assert result["current_condition"][0]["pressure"] == 735.0
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_custom_coords(self, mock_get) -> None:
|
||
"""Кастомные координаты должны быть в URL."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 25,
|
||
"feels_like": 22,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather(lat=55.7558, lon=37.6173)
|
||
|
||
assert result is not None
|
||
mock_get.assert_called_once()
|
||
call_url = mock_get.call_args[0][0]
|
||
assert "55.7558" in call_url
|
||
assert "37.6173" in call_url
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_missing_condition(self, mock_get) -> None:
|
||
"""Отсутствующий condition → 'Неизвестно'."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {"fact": {"temp": 10}}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather()
|
||
|
||
assert result is not None
|
||
assert result["current_condition"][0]["weatherDesc"] == [{"value": "Неизвестно"}]
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_ssl_error_retry(self, mock_get) -> None:
|
||
"""SSLError на первой попытке → retry → успех."""
|
||
success_response = MagicMock()
|
||
success_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "partly_cloudy",
|
||
"wind_speed": 4,
|
||
"wind_gust": 6,
|
||
"wind_dir": "w",
|
||
"humidity": 60,
|
||
"pressure_mm": 740.0,
|
||
"pressure_pa": 987,
|
||
}
|
||
}
|
||
success_response.raise_for_status = MagicMock()
|
||
mock_get.side_effect = [SSLError("SSL Error"), success_response]
|
||
|
||
result = await fetch_weather(max_retries=2)
|
||
|
||
assert result is not None
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_connection_error_retry(self, mock_get) -> None:
|
||
"""ConnectionError → retry."""
|
||
success_response = MagicMock()
|
||
success_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "partly_cloudy",
|
||
"wind_speed": 4,
|
||
"wind_gust": 6,
|
||
"wind_dir": "w",
|
||
"humidity": 60,
|
||
"pressure_mm": 740.0,
|
||
"pressure_pa": 987,
|
||
}
|
||
}
|
||
success_response.raise_for_status = MagicMock()
|
||
mock_get.side_effect = [ConnectionError("No connection"), success_response]
|
||
|
||
result = await fetch_weather(max_retries=2)
|
||
|
||
assert result is not None
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_timeout_retry(self, mock_get) -> None:
|
||
"""Timeout → retry."""
|
||
success_response = MagicMock()
|
||
success_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "partly_cloudy",
|
||
"wind_speed": 4,
|
||
"wind_gust": 6,
|
||
"wind_dir": "w",
|
||
"humidity": 60,
|
||
"pressure_mm": 740.0,
|
||
"pressure_pa": 987,
|
||
}
|
||
}
|
||
success_response.raise_for_status = MagicMock()
|
||
mock_get.side_effect = [Timeout("Timed out"), success_response]
|
||
|
||
result = await fetch_weather(max_retries=2)
|
||
|
||
assert result is not None
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_all_retries_fail(self, mock_get) -> None:
|
||
"""Все попытки не удались → None."""
|
||
mock_get.side_effect = [
|
||
ConnectionError("fail"),
|
||
ConnectionError("fail"),
|
||
ConnectionError("fail"),
|
||
]
|
||
|
||
result = await fetch_weather(max_retries=3)
|
||
|
||
assert result is None
|
||
assert mock_get.call_count == 3
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_request_exception(self, mock_get) -> None:
|
||
"""Общий RequestException → None."""
|
||
mock_get.side_effect = requests.RequestException("Generic error")
|
||
|
||
result = await fetch_weather(max_retries=1)
|
||
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_http_error(self, mock_get) -> None:
|
||
"""HTTP 404 → raise_for_status бросит исключение → None."""
|
||
mock_response = MagicMock()
|
||
mock_response.raise_for_status.side_effect = requests.HTTPError("HTTP 404")
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather(max_retries=1)
|
||
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_json_parse_error(self, mock_get) -> None:
|
||
"""Ошибка парсинга JSON → None."""
|
||
mock_response = MagicMock()
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_response.json.side_effect = requests.RequestException("JSON Error")
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather(max_retries=1)
|
||
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_wind_speed_0(self, mock_get) -> None:
|
||
"""Нулевая скорость ветра должна корректно обрабатываться."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 0,
|
||
"feels_like": -2,
|
||
"condition": "fog",
|
||
"wind_speed": 0,
|
||
"wind_gust": 0,
|
||
"wind_dir": "n",
|
||
"humidity": 95,
|
||
"pressure_mm": 750.0,
|
||
"pressure_pa": 1000,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather()
|
||
|
||
assert result is not None
|
||
assert result["current_condition"][0]["wind_speed_mps"] == 0
|
||
assert result["current_condition"][0]["pressure"] == 750.0
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_negative_temp(self, mock_get) -> None:
|
||
"""Отрицательная температура."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": -15,
|
||
"feels_like": -22,
|
||
"condition": "heavy_snow",
|
||
"wind_speed": 8,
|
||
"wind_gust": 12,
|
||
"wind_dir": "n",
|
||
"humidity": 90,
|
||
"pressure_mm": 720.0,
|
||
"pressure_pa": 960,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather()
|
||
|
||
assert result is not None
|
||
assert result["current_condition"][0]["temp_C"] == -15
|
||
assert result["current_condition"][0]["weatherDesc"] == [{"value": "Сильный снег"}]
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_includes_headers(self, mock_get) -> None:
|
||
"""Запрос должен содержать X-Yandex-API-Key заголовок."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
with patch("utils.pogoda._get_api_key", return_value="test-key"):
|
||
await fetch_weather()
|
||
|
||
mock_get.assert_called_once()
|
||
call_kwargs = mock_get.call_args[1]
|
||
assert "X-Yandex-API-Key" in call_kwargs["headers"]
|
||
assert call_kwargs["headers"]["X-Yandex-API-Key"] == "test-key"
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_fetch_weather_json_decode_error(self, mock_get) -> None:
|
||
"""json.JSONDecodeError (невалидный JSON от API) → graceful None."""
|
||
import json as _json
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_response.json.side_effect = _json.JSONDecodeError("Expecting value", "", 0)
|
||
mock_get.return_value = mock_response
|
||
|
||
result = await fetch_weather(max_retries=1)
|
||
|
||
assert result is None
|
||
|
||
|
||
class TestWeatherCache:
|
||
"""Тесты кэширования погоды."""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_cache(self) -> None:
|
||
clear_weather_cache()
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _mock_api_key(self) -> None:
|
||
with patch("utils.pogoda._get_api_key", return_value="test-key"):
|
||
yield
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_cache_returns_cached_data(self, mock_get) -> None:
|
||
"""Второй запрос в течение TTL возвращает кэш без вызова API."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
result1 = await fetch_weather()
|
||
result2 = await fetch_weather()
|
||
|
||
assert result1 is not None
|
||
assert result2 is not None
|
||
assert result1 == result2
|
||
# API вызван только один раз
|
||
assert mock_get.call_count == 1
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_bypass_cache_forces_new_request(self, mock_get) -> None:
|
||
"""bypass_cache=True делает новый запрос к API."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
await fetch_weather()
|
||
await fetch_weather(bypass_cache=True)
|
||
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_cache_stores_failure(self, mock_get) -> None:
|
||
"""Провал запроса кэшируется — повторный вызов не долбит API."""
|
||
mock_get.side_effect = ConnectionError("fail")
|
||
|
||
result1 = await fetch_weather(max_retries=1)
|
||
result2 = await fetch_weather()
|
||
|
||
assert result1 is None
|
||
assert result2 is None # из кэша
|
||
# API вызван только один раз (первый раз)
|
||
assert mock_get.call_count == 1
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_clear_cache_invalidates(self, mock_get) -> None:
|
||
"""clear_weather_cache сбрасывает кэш."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
await fetch_weather()
|
||
clear_weather_cache()
|
||
await fetch_weather()
|
||
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
async def test_cache_ttl_expiry(self, mock_get) -> None:
|
||
"""После истечения TTL кэш инвалидируется."""
|
||
import utils.pogoda as pogoda_module
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"fact": {
|
||
"temp": 20,
|
||
"feels_like": 18,
|
||
"condition": "clear",
|
||
"wind_speed": 3,
|
||
"wind_gust": 5,
|
||
"wind_dir": "s",
|
||
"humidity": 50,
|
||
"pressure_mm": 760.0,
|
||
"pressure_pa": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
|
||
# Устанавливаем TTL = 0, чтобы кэш сразу истёк
|
||
original_ttl = pogoda_module._WEATHER_CACHE_TTL
|
||
pogoda_module._WEATHER_CACHE_TTL = 0.0
|
||
|
||
try:
|
||
await fetch_weather()
|
||
await fetch_weather()
|
||
|
||
assert mock_get.call_count == 2
|
||
finally:
|
||
pogoda_module._WEATHER_CACHE_TTL = original_ttl
|