W dzisiejszym artykule, opiszę jak łatwo można testować klasy, wykonujące zapytania do zewnętrznych api. Dobrą praktyką jest wykonywanie testów bez odpytywania prawdziwych serwerów.
Poniżej przedstawię prosty kod programu (jaki musiałem napisać w ramach pewnego zadania rekturacyjnego :) ), który dla zadanego miasta zwróci nam dzisiejszą datę oraz temperaturę.
#!/usr/bin/env python3
import argparse
import os
import logging
import requests
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv('API_KEY')
ACCUWEATHER_MIRROR_URL = os.getenv('ACCUWEATHER_MIRROR_URL')
class AccuWeather(object):
"""
The class for interacting with API AccuWeather.
:param city:
:type city: str
:param __api_key:
:type __api_key: str
:param __accuweather_url:
:type __accuweather_url: str
"""
__name__ = "forecast"
__author__ = "Kamil Mirończuk"
__license__ = "GNU General Public License v3.0"
__version__ = "1.1"
__status__ = "production"
def __init__(self, api_key: str = API_KEY, accuweather_url: str = ACCUWEATHER_MIRROR_URL):
self.__api_key = api_key
self.__accuweather_url = accuweather_url
self.__forecasts = None
self.city_token = None
def __str__(self):
return "Custom API for AccuWeather"
@property
def city_token(self) -> str:
"""Get actual city token.
:return: city token
:rtype: str
"""
return self.__city_id
@city_token.setter
def city_token(self, city):
"""Set city token.
:raises TypeError: City type is not valid
"""
if city is None:
self.__city_id: str = ''
elif type(city) == str:
self.__city_id: str = self.get_city(city)
else:
raise TypeError("City must be a string.")
def get_city(self, city: str) -> str:
"""
get the current city id based on the name taken from the api
:param city:
:type city: str
:returns: returns city id
:rtype: str
"""
try:
r = requests.get(
f"{self.__accuweather_url}/locations/v1/cities/search?q={city}",
params={
"apikey": self.__api_key,
"metric": "true"
})
r.raise_for_status()
except requests.exceptions.HTTPError as errh:
logging.error("Http Error: {}".format(errh))
except requests.exceptions.ConnectionError as errc:
logging.error("Error Connecting: {}".format(errc))
except requests.exceptions.Timeout as errt:
logging.error("Timeout Error: {}".format(errt))
except requests.exceptions.RequestException as err:
logging.error("Request Error: {}".format(err))
try:
response: dict = r.json()
cityId: str = response[0].get('Key')
except:
logging.error("An error occurred while adjusting city parameter")
raise SystemExit()
return cityId
def store_forecasts(self):
"""
save forecasts to cache
"""
self.__forecasts = self.get_forecasts()
def get_forecasts(self) -> dict:
"""
retrieves weather for a previously set city
:returns: returns forecasts
:rtype: dict
"""
try:
r = requests.get(
f"{self.__accuweather_url}/forecasts/v1/daily/1day/{self.__city_id}",
params={
"apikey": self.__api_key,
"metric": "true"
})
r.raise_for_status()
except requests.exceptions.HTTPError as errh:
logging.error("Http Error: {}".format(errh))
except requests.exceptions.ConnectionError as errc:
logging.error("Error Connecting: {}".format(errc))
except requests.exceptions.Timeout as errt:
logging.error("Timeout Error: {}".format(errt))
except requests.exceptions.RequestException as err:
logging.error("Request Error: {}".format(err))
try:
response: dict = r.json()
except:
logging.error("An error occurred while adjusting forecasts parameters")
raise SystemExit()
__forecasts = response
return response
def __get_average_temperature(self) -> float:
"""
returns the average temperature based on the currently setting city
:returns: returns the average temperature
:rtype: float
"""
try:
temperature = self.__forecasts['DailyForecasts'][0]['Temperature']
max_temperature: float = temperature['Minimum']['Value']
min_temperature: float = temperature['Maximum']['Value']
average_temperature: float = (min_temperature + max_temperature) / 2
except:
logging.error("Date cannot be set. Get the weather forecast first.")
raise SystemExit()
return average_temperature
def __get_date_from_response(self) -> str:
"""
returns the date based on received weather forecast
:returns: returns date
:rtype: str
"""
if self.__forecasts:
forecasts_date_iso: str = self.__forecasts['DailyForecasts'][0]['Date']
forecasts_date = datetime.fromisoformat(forecasts_date_iso).strftime("%d.%m.%Y")
else:
logging.warning('Get the weather forecast first.')
raise SystemExit()
return forecasts_date
def get_day_temperature(self):
"""prints the temperature of the day on the screen"""
actual_date: str = self.__get_date_from_response()
average_temperature: float = self.__get_average_temperature()
return f"Date: {actual_date}, temperature: {average_temperature:.1f}°C"
def parse_args():
parser = argparse.ArgumentParser(prog='forecasts')
parser.add_argument('city', help='City name')
parser.add_argument("-v", "--verbose", action="count", default=0, help="Increase verbosity")
args = parser.parse_args()
try:
loglevel = {
0: logging.ERROR,
1: logging.WARN,
2: logging.INFO}[args.verbose]
except KeyError:
loglevel = logging.DEBUG
logging.basicConfig(level=loglevel,
format='%(asctime)s %(message)s',
filename='logs.log',)
console = logging.StreamHandler()
logging.getLogger().setLevel(loglevel)
logging.getLogger('').addHandler(console)
return args
def main(args, env):
if args.city:
logging.info('Program started')
instance = AccuWeather()
instance.city_token = args.city
instance.store_forecasts()
print(instance.get_day_temperature())
if __name__ == '__main__':
main(parse_args(), os.environ)
Sam kod do działania wymaga dodatkowo stworzenia poniższego pliku .env:
API_KEY = 'test-key'
ACCUWEATHER_MIRROR_URL = 'http://dataservice.accuweather.com'
W głąb kodu...
Ze względu na dużą przejrzystość samej klasy, skupimy bardziej się bezpośrednio na stworzeniu testów, samego api.
Na samym początku, zablokujemy możliwość odpytywania api zewnętrznego tworząc własny fixture, celem doprowadzenia testów do stanu, gdzie nie bedą go wymagały.
@pytest.fixture(autouse=True)
def no_requests(monkeypatch):
monkeypatch.delattr('requests.sessions.Session.request')
Idąc dalej, tworzymy prosty test, który z użyciem monkey patchingu sprawi że dokonamy patcha całej metody get_city.
def test_weatcher_set_city(monkeypatch):
weatcher = AccuWeather()
def monkey_get_city(city: str):
if city == 'Gdansk':
return '275174'
return None
monkeypatch.setattr(weatcher, 'get_city', monkey_get_city)
weatcher.city_token = 'Gdansk'
assert (weatcher.city_token, '275174')
My jednak chcemy, by zamiast dokonywać nadpisywania całej metody, jedynie samo wywołanie get było zmienione. W tym celu stworzymy klasę mockującą obiekt klasy requests, który wywoływany jest w metodzie get_city i zwraca nam unikalny klucz (id) miasta .
class ResponseGetMock(object):
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def raise_for_status():
pass
def json():
return [{'Version': 1, 'Key': '2696858', 'Type': 'City', 'Rank': 85, 'LocalizedName': 'Warszawa'}]
Idąc dalej tworzymy fixture, które za pomocą monkey patchingu, podmieni nam metodę get z klasy requests na powyżej zdefiniowaną.
@pytest.fixture(params=['Warszawa', 'Gdansk'], name='weatcher')
def fixture_weatcher(request, monkeypatch):
my_weatcher = AccuWeather()
def monkey_return(url, params):
return ResponseGetMock
monkeypatch.setattr(requests, 'get', monkey_return)
my_weatcher.city_token = request.param
yield my_weatcher
Dobre praktyki
W dalszej części chcemy spatchować metodę get_forecasts.
Na początku zdefiniujemy w stałej co zwróci nam api.
WEATCHER_API = {
'DailyForecasts':
[{'Date': '2023-01-27T07:00:00+01:00', 'EpochDate': 1674799200,
'Temperature': {
'Minimum': {'Value': -2.6, 'Unit': 'C', 'UnitType': 17},
'Maximum': {'Value': 3.5, 'Unit': 'C', 'UnitType': 17}
}}]}
Następnie możemy stworzyć test który z użyciem unit test, dokona patchingu
@patch.object(AccuWeather, 'get_forecasts', return_value=WEATCHER_API)
def test_weatcher_get_forecasts(weatcher_mock, weatcher):
with patch('weatcher.AccuWeather.get_forecasts', return_value=WEATCHER_API):
assert (weatcher.get_forecasts(), dict)
Lub też dokonać patchowania zgodnie z pytests
def test_weatcher_get_day_temperature(weatcher, monkeypatch):
def get_forecasts():
return WEATCHER_API
monkeypatch.setattr(weatcher, 'get_forecasts', get_forecasts)
weatcher.store_forecasts()
assert (weatcher.get_day_temperature(), 'Date: 27.01.2023, temperature: 0.4°C')
Na dziś to już koniec, polecam zagłębić się w różne metody mockowania obiektów, gdyż samo zagadnienie, pozostawia jeszcze dużo kwestii, które mogą zaskoczyć.
Link do repozytorium https://github.com/Kamwebdev/Forecast
Komentarze