4b9bb7e97a
- pytest.ini для конфигурации тестов - tests/test_pogoda.py — тесты translate_weather, pressure_to_mmhg, wmo_to_russian (93 теста) - tests/test_fetch_cat.py — тесты fetch_cat (10 тестов) - tests/test_fetch_rss.py — тесты fetch_rss (20 тестов) - tests/test_format_articles.py — тесты truncate_title, parse_date, format_articles (24 теста) - tests/test_fetch_weather.py — тесты fetch_weather, fetch_open_meteo (20 тестов) - tests/test_commands_pogoda.py — тесты команды !pogoda (13 тестов) - Обновить AGENTS.md и requirements.txt
232 lines
11 KiB
Python
232 lines
11 KiB
Python
import asyncio
|
||
import pytest
|
||
from unittest.mock import patch, MagicMock
|
||
from utils.pogoda import fetch_weather, fetch_open_meteo
|
||
|
||
|
||
class TestFetchWeather:
|
||
"""Тесты функции fetch_weather() — получение погоды с retry-логикой."""
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_success(self, mock_get):
|
||
"""Успешный ответ должен вернуть JSON-данные."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {"current_condition": [{"temp_C": 20}]}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result == {"current_condition": [{"temp_C": 20}]}
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_fallback_on_ssl_error(self, mock_get):
|
||
"""SSLError на первой попытке → fallback на Open-Meteo."""
|
||
from requests.exceptions import SSLError
|
||
mock_get.side_effect = [
|
||
SSLError("SSL Error"),
|
||
MagicMock(json=MagicMock(return_value={"result": "fallback"})),
|
||
]
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
mock_fallback.return_value = {"result": "fallback"}
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result == {"result": "fallback"}
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_fallback_on_connection_error(self, mock_get):
|
||
"""ConnectionError → fallback на Open-Meteo."""
|
||
from requests.exceptions import ConnectionError
|
||
mock_get.side_effect = ConnectionError("No connection")
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
mock_fallback.return_value = {"result": "fallback"}
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result == {"result": "fallback"}
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_fallback_on_timeout(self, mock_get):
|
||
"""Timeout → fallback на Open-Meteo."""
|
||
from requests.exceptions import Timeout
|
||
mock_get.side_effect = Timeout("Timed out")
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
mock_fallback.return_value = {"result": "fallback"}
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result == {"result": "fallback"}
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_all_retries_fail(self, mock_get):
|
||
"""Все попытки не удались → fallback на Open-Meteo."""
|
||
from requests.exceptions import ConnectionError
|
||
mock_get.side_effect = ConnectionError("No connection")
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
mock_fallback.return_value = None
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_request_exception(self, mock_get):
|
||
"""Общий RequestException → fallback на Open-Meteo."""
|
||
import requests
|
||
mock_get.side_effect = requests.RequestException("Generic error")
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
mock_fallback.return_value = {"result": "fallback"}
|
||
result = asyncio.run(fetch_weather("https://test.example.com"))
|
||
assert result == {"result": "fallback"}
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_weather_http_error_no_fallback(self, mock_get):
|
||
"""HTTP-ошибка (raise_for_status) не ловится, падает."""
|
||
mock_response = MagicMock()
|
||
mock_response.raise_for_status.side_effect = Exception("HTTP 500")
|
||
mock_get.return_value = mock_response
|
||
with pytest.raises(Exception):
|
||
asyncio.run(fetch_weather("https://test.example.com"))
|
||
|
||
|
||
class TestFetchOpenMeteo:
|
||
"""Тесты функции fetch_open_meteo() — fallback на Open-Meteo API."""
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_success(self, mock_get):
|
||
"""Успешный ответ должен вернуть данные в формате current_condition."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"current": {
|
||
"temperature": 15,
|
||
"apparent_temperature": 12,
|
||
"weather_code": 3,
|
||
"wind_speed_10m": 5.5,
|
||
"relative_humidity_2m": 65,
|
||
"pressure_msl": 1013,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo())
|
||
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]["humidity"] == 65
|
||
assert result["current_condition"][0]["pressure"] == 1013
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_custom_coords(self, mock_get):
|
||
"""Кастомные координаты должны быть в URL."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {"current": {"temperature": 25, "apparent_temperature": 22, "weather_code": 0, "wind_speed_10m": 3, "relative_humidity_2m": 50, "pressure_msl": 1020}}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo(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")
|
||
def test_fetch_open_meteo_missing_weather_code(self, mock_get):
|
||
"""Отсутствующий weather_code → 'Неизвестно'."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {"current": {"temperature": 10}}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is not None
|
||
assert result["current_condition"][0]["weatherDesc"] == [{"value": "Неизвестно"}]
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_ssl_error(self, mock_get):
|
||
"""SSLError → вернуть None."""
|
||
from requests.exceptions import SSLError
|
||
mock_get.side_effect = SSLError("SSL Error")
|
||
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
|
||
# Внутренний fallback тоже падает, проверяем что возвращается None
|
||
pass
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_connection_error(self, mock_get):
|
||
"""ConnectionError → вернуть None."""
|
||
from requests.exceptions import ConnectionError
|
||
mock_get.side_effect = ConnectionError("No connection")
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_timeout(self, mock_get):
|
||
"""Timeout → вернуть None."""
|
||
from requests.exceptions import Timeout
|
||
mock_get.side_effect = Timeout("Timed out")
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_request_exception(self, mock_get):
|
||
"""Общий RequestException → вернуть None."""
|
||
import requests
|
||
mock_get.side_effect = requests.RequestException("Error")
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_json_parse_error(self, mock_get):
|
||
"""Ошибка парсинга JSON → вернуть None."""
|
||
import requests
|
||
mock_response = MagicMock()
|
||
mock_response.json.side_effect = requests.RequestException("JSON Error")
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_retry_on_error(self, mock_get):
|
||
"""Retry: первая попытка падает, вторая успешна."""
|
||
from requests.exceptions import ConnectionError
|
||
success_response = MagicMock()
|
||
success_response.json.return_value = {"current": {"temperature": 20, "apparent_temperature": 18, "weather_code": 1, "wind_speed_10m": 4, "relative_humidity_2m": 60, "pressure_msl": 1015}}
|
||
success_response.raise_for_status = MagicMock()
|
||
mock_get.side_effect = [ConnectionError("fail"), success_response]
|
||
result = asyncio.run(fetch_open_meteo(max_retries=2))
|
||
assert result is not None
|
||
assert mock_get.call_count == 2
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_all_retries_fail(self, mock_get):
|
||
"""Все попытки неудачны → None."""
|
||
from requests.exceptions import ConnectionError
|
||
mock_get.side_effect = [ConnectionError("fail"), ConnectionError("fail"), ConnectionError("fail")]
|
||
result = asyncio.run(fetch_open_meteo(max_retries=3))
|
||
assert result is None
|
||
assert mock_get.call_count == 3
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_http_error(self, mock_get):
|
||
"""HTTP 404 → raise_for_status бросит исключение → None."""
|
||
import requests
|
||
mock_response = MagicMock()
|
||
mock_response.raise_for_status.side_effect = requests.HTTPError("HTTP 404")
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is None
|
||
|
||
@patch("utils.pogoda._session.get")
|
||
def test_fetch_open_meteo_wind_speed_0(self, mock_get):
|
||
"""Нулевая скорость ветра должна корректно обрабатываться."""
|
||
mock_response = MagicMock()
|
||
mock_response.json.return_value = {
|
||
"current": {
|
||
"temperature": 0,
|
||
"apparent_temperature": -2,
|
||
"weather_code": 45,
|
||
"wind_speed_10m": 0,
|
||
"relative_humidity_2m": 95,
|
||
"pressure_msl": 1000,
|
||
}
|
||
}
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_get.return_value = mock_response
|
||
result = asyncio.run(fetch_open_meteo())
|
||
assert result is not None
|
||
assert result["current_condition"][0]["windspeedKmph"] == 0
|
||
assert result["current_condition"][0]["pressure"] == 1000
|