FitTrackee/fittrackee/tests/workouts/test_utils/test_weather_service.py

248 lines
8.1 KiB
Python
Raw Normal View History

2022-11-17 00:13:43 +01:00
from datetime import datetime
from typing import Dict, Optional
from unittest.mock import Mock, patch, sentinel
import pytest
import pytz
import requests
from gpxpy.gpx import GPXTrackPoint
from fittrackee.tests.mixins import CallArgsMixin
from fittrackee.tests.utils import random_string
from fittrackee.workouts.utils.weather.visual_crossing import VisualCrossing
from fittrackee.workouts.utils.weather.weather_service import WeatherService
VISUAL_CROSSING_RESPONSE = {
"queryCost": 1,
"latitude": 48.866667,
"longitude": 2.333333,
"resolvedAddress": "48.866667,2.333333",
"address": "48.866667,2.333333",
"timezone": "Europe/Paris",
"tzoffset": 1.0,
"days": [
{
"datetime": "2022-11-15",
"datetimeEpoch": 1668466800,
"temp": 10.4,
"humidity": 93.3,
"windspeed": 18.9,
"winddir": 179.4,
"conditions": "Rain, Partially cloudy",
"description": "Partly cloudy throughout the day with rain.",
"icon": "rain",
}
],
"currentConditions": {
"datetime": "13:00:00",
"datetimeEpoch": 1668513600,
"temp": 11.3,
"humidity": 93.1,
"windspeed": 14.0,
"winddir": 161.9,
"conditions": "Rain, Overcast",
"icon": "rain",
},
}
class WeatherTestCase:
api_key = random_string()
@staticmethod
def get_gpx_point(time: Optional[datetime] = None) -> GPXTrackPoint:
return GPXTrackPoint(latitude=48.866667, longitude=2.333333, time=time)
class TestVisualCrossingGetTimestamp(WeatherTestCase):
def test_it_returns_expected_timestamp_as_integer(self) -> None:
time = datetime.utcnow()
visual_crossing = VisualCrossing(api_key=self.api_key)
timestamp = visual_crossing._get_timestamp(time)
assert isinstance(timestamp, int)
@pytest.mark.parametrize(
'input_datetime,expected_datetime',
[
('2020-12-15T13:00:00', '2020-12-15T13:00:00'),
('2020-12-15T13:29:59', '2020-12-15T13:00:00'),
('2020-12-15T13:30:00', '2020-12-15T14:00:00'),
('2020-12-15T13:59:59', '2020-12-15T14:00:00'),
],
)
def test_it_returns_rounded_time(
self, input_datetime: str, expected_datetime: str
) -> None:
time = datetime.strptime(input_datetime, '%Y-%m-%dT%H:%M:%S')
visual_crossing = VisualCrossing(api_key=self.api_key)
timestamp = visual_crossing._get_timestamp(time)
assert (
timestamp
== datetime.strptime(
expected_datetime, '%Y-%m-%dT%H:%M:%S'
).timestamp()
)
class TestVisualCrossingGetWeather(WeatherTestCase, CallArgsMixin):
@staticmethod
def get_response() -> Mock:
response_mock = Mock()
response_mock.raise_for_status = Mock()
response_mock.json = Mock()
response_mock.json.return_value = VISUAL_CROSSING_RESPONSE
return response_mock
def test_it_calls_api_with_time_and_point_location(self) -> None:
time = datetime(
year=2022,
month=11,
day=15,
hour=12,
minute=00,
second=00,
tzinfo=pytz.utc,
)
point = self.get_gpx_point(time)
visual_crossing = VisualCrossing(api_key=self.api_key)
with patch.object(requests, 'get') as get_mock:
visual_crossing.get_weather(point)
args = self.get_args(get_mock.call_args)
assert args[0] == (
'https://weather.visualcrossing.com/VisualCrossingWebServices/'
f'rest/services/timeline/{point.latitude},{point.longitude}/'
f'{int(point.time.timestamp())}' # type: ignore
)
def test_it_calls_api_with_expected_params(self) -> None:
visual_crossing = VisualCrossing(api_key=self.api_key)
with patch.object(requests, 'get') as get_mock:
visual_crossing.get_weather(self.get_gpx_point(datetime.utcnow()))
kwargs = self.get_kwargs(get_mock.call_args)
assert kwargs.get('params') == {
'key': self.api_key,
'iconSet': 'icons1',
'unitGroup': 'metric',
'contentType': 'json',
'elements': (
'datetime,datetimeEpoch,temp,humidity,windspeed,'
'winddir,conditions,description,icon'
),
'include': 'current',
}
def test_it_returns_data_from_current_conditions(self) -> None:
point = self.get_gpx_point(
datetime(
year=2022,
month=11,
day=15,
hour=13,
minute=00,
second=00,
tzinfo=pytz.utc,
).astimezone(pytz.timezone('Europe/Paris'))
)
visual_crossing = VisualCrossing(api_key=self.api_key)
with patch.object(requests, 'get', return_value=self.get_response()):
weather_data = visual_crossing.get_weather(point)
current_conditions: Dict = VISUAL_CROSSING_RESPONSE[ # type: ignore
'currentConditions'
]
assert weather_data == {
'icon': current_conditions['icon'],
'temperature': current_conditions['temp'],
'humidity': current_conditions['humidity'] / 100,
'wind': (current_conditions['windspeed'] * 1000) / 3600,
'windBearing': current_conditions['winddir'],
}
class TestWeatherService(WeatherTestCase):
@pytest.mark.parametrize(
'input_api_key,input_provider',
[
('', 'visualcrossing'),
2022-11-17 00:13:43 +01:00
('valid_api_key', ''),
('valid_api_key', 'invalid_provider'),
('valid_api_key', 'darksky'), # removed provider
2022-11-17 00:13:43 +01:00
],
)
def test_weather_api_is_none_when_configuration_is_invalid(
self,
monkeypatch: pytest.MonkeyPatch,
input_api_key: str,
input_provider: str,
) -> None:
monkeypatch.setenv('WEATHER_API_KEY', input_api_key)
monkeypatch.setenv('WEATHER_API_PROVIDER', input_provider)
weather_service = WeatherService()
assert weather_service.weather_api is None
@pytest.mark.parametrize(
'input_provider',
['visualcrossing', 'VisualCrossing'],
)
def test_weather_api_is_visualcrossing_when_configured(
self,
monkeypatch: pytest.MonkeyPatch,
input_provider: str,
) -> None:
monkeypatch.setenv('WEATHER_API_KEY', 'valid_api_key')
monkeypatch.setenv('WEATHER_API_PROVIDER', input_provider)
weather_service = WeatherService()
assert isinstance(weather_service.weather_api, VisualCrossing)
def test_it_returns_none_when_no_weather_api(self) -> None:
weather_service = WeatherService()
weather_service.weather_api = None
point = self.get_gpx_point(datetime.utcnow())
weather_data = weather_service.get_weather(point)
assert weather_data is None
def test_it_returns_none_when_point_time_is_none(self) -> None:
weather_service = WeatherService()
weather_service.weather_api = VisualCrossing('api_key')
2022-11-17 00:13:43 +01:00
point = self.get_gpx_point(None)
weather_data = weather_service.get_weather(point)
assert weather_data is None
def test_it_returns_none_when_weather_api_raises_exception(self) -> None:
weather_api = Mock()
weather_api.get_weather = Mock()
weather_api.get_weather.side_effect = Exception()
weather_service = WeatherService()
weather_service.weather_api = weather_api
point = self.get_gpx_point(datetime.utcnow())
weather_data = weather_service.get_weather(point)
assert weather_data is None
def test_it_returns_weather_data(self) -> None:
weather_api = Mock()
weather_api.get_weather = Mock()
weather_api.get_weather.return_value = sentinel
weather_service = WeatherService()
weather_service.weather_api = weather_api
point = self.get_gpx_point(datetime.utcnow())
weather_data = weather_service.get_weather(point)
assert weather_data == sentinel