345 lines
11 KiB
Python
345 lines
11 KiB
Python
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.dark_sky import DarkSky
|
|
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 TestDarksky(WeatherTestCase):
|
|
def test_it_calls_forecast_io_with_datetime_in_utc_when_naive_datetime_is_provided( # noqa
|
|
self,
|
|
) -> None:
|
|
naive_point_time = datetime(
|
|
year=2022, month=6, day=11, hour=10, minute=23, second=00
|
|
)
|
|
point = self.get_gpx_point(naive_point_time)
|
|
darksky = DarkSky(api_key=self.api_key)
|
|
with patch(
|
|
'fittrackee.workouts.utils.weather.dark_sky.forecastio'
|
|
) as forecast_io_mock:
|
|
|
|
darksky.get_weather(point)
|
|
|
|
forecast_io_mock.load_forecast.assert_called_with(
|
|
self.api_key,
|
|
point.latitude,
|
|
point.longitude,
|
|
time=datetime(
|
|
year=2022,
|
|
month=6,
|
|
day=11,
|
|
hour=10,
|
|
minute=23,
|
|
second=00,
|
|
tzinfo=pytz.utc,
|
|
),
|
|
units='si',
|
|
)
|
|
|
|
def test_it_calls_forecast_io_with_provided_datetime_when_with_timezone(
|
|
self,
|
|
) -> None:
|
|
paris_point_time = datetime(
|
|
year=2022,
|
|
month=6,
|
|
day=11,
|
|
hour=10,
|
|
minute=23,
|
|
second=00,
|
|
).astimezone(pytz.timezone('Europe/Paris'))
|
|
point = self.get_gpx_point(paris_point_time)
|
|
darksky = DarkSky(api_key=self.api_key)
|
|
with patch(
|
|
'fittrackee.workouts.utils.weather.dark_sky.forecastio'
|
|
) as forecast_io_mock:
|
|
|
|
darksky.get_weather(point)
|
|
|
|
forecast_io_mock.load_forecast.assert_called_with(
|
|
self.api_key,
|
|
point.latitude,
|
|
point.longitude,
|
|
time=paris_point_time,
|
|
units='si',
|
|
)
|
|
|
|
def test_it_returns_forecast_currently_data(self) -> None:
|
|
darksky = DarkSky(api_key=self.api_key)
|
|
with patch(
|
|
'fittrackee.workouts.utils.weather.dark_sky.forecastio'
|
|
) as forecast_io_mock:
|
|
forecast_io_mock.load_forecast().currently.return_value = sentinel
|
|
|
|
weather_data = darksky.get_weather(
|
|
self.get_gpx_point(datetime.utcnow())
|
|
)
|
|
|
|
assert weather_data == {
|
|
'icon': sentinel.icon,
|
|
'temperature': sentinel.temperature,
|
|
'humidity': sentinel.humidity,
|
|
'wind': sentinel.windSpeed,
|
|
'windBearing': sentinel.windBearing,
|
|
}
|
|
|
|
|
|
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',
|
|
[
|
|
('', 'darksky'),
|
|
('valid_api_key', ''),
|
|
('valid_api_key', 'invalid_provider'),
|
|
],
|
|
)
|
|
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',
|
|
['darksky', 'DARKSKY'],
|
|
)
|
|
def test_weather_api_is_darksky_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, DarkSky)
|
|
|
|
@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 = DarkSky('api_key')
|
|
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
|