W ramach prac nad CX-Metrics, systemem automatycznego rozpoznawania cech demograficznych i zachowań klientów, poprzez połączenie algorytmów Sztucznej Inteligencji i istniejącej infrastruktury kamer (CCTV), musieliśmy dokonywać szeregu testów, w tym testów dotyczących kodu pisanego w stylu asynchronicznym.
Jednym z przykładów jest nasz autorski serwer, którego zadaniem jest streamowanie wideo bezpośrednio na frontend, tak by z jednej strony móc oglądać obraz z kamer w przeglądarce www, a z drugiej zachować jak najmniejszą latencję (rozumianą jako opóźnienie między wyświetlanymi klatkami filmowymi, a rzeczywistymi wydarzeniami mającymi miejsce w sklepie, czy galerii handlowej).
Do napisania serwera wybraliśmy język Python wraz z pakietem aiohttp, który oparty jest o paradygmat programowania asynchronicznego, natomiast testy jednostkowe tego komponentu oparte były o framework pytest. W tym artykule chcemy pokazać jak podeszliśmy do zagadnienia testowania funkcji asynchronicznych przy użyciu pakietu pytest, niemniej poprzedzimy to krótkim wstępem dotyczącym programowania asynchronicznego w Pythonie – jeśli chcesz poznać więcej szczegółów, polecamy ten artykuł, który wyczerpuje tę materię w znacznie większym stopniu.
Programowanie współbieżne w Pythonie
Programowanie współbieżne w Pythonie to skomplikowany temat ze względu na ograniczenia interpretera (Global Interpreter Lock, GIL). Pomimo tego, Python w swojej bibliotece standardowej zawiera kilka pakietów, które pozwalają na programowanie współbieżne: multiprocessing – kod uruchamiany w osobnych procesach, wykonywany prawdziwie równolegle threading – kod uruchamiany w wątkach, ze względu na wspomniany wcześniej GIL, wątki są jedynie współbieżne asyncio – realizuje paradygmat programowania asynchronicznego, które opiszemy poniżej.
To, którego rozwiązania użyjemy zależy w dużej mierze od naszego problemu, w przypadku wspomnianego wyżej serwera, pośredniczącego między kamerą przemysłową, a przeglądarką www, użycie asyncio okazało się optymalne.
Pakiet asyncio
To co może wydawać się sprzeczne z intuicją to fakt, że w asyncio wszystkie operacje są wykonywane na jednym wątku. Dzięki temu synchronizacja między zadaniami jest zdecydowanie łatwiejsza niż podczas korzystania z threading, natomiast zadania dzielą ze sobą pamięć (w przeciwieństwie do multiprocessing) co ułatwia między nimi komunikację. Pakiet asyncio wprowadza w Pythonie dwa nowe słowa kluczowe: async – poprzedza definicję funkcji asynchronicznej, zwanej również coroutine await – poprzedza wywołanie funkcji asynchronicznej, może być użyte jedynie w funkcji oznaczonej jako async Za zarządzanie funkcjami asynchronicznymi odpowiedzialny jest tzw. event loop. Po napotkaniu słowa kluczowego await, event loop może zdecydować o tym aby przerwać aktualnie wykonywane zadanie i przydzielić czas procesora do innego zadania. Takie podejście doskonale sprawdza się, gdy znaczną część czasu spędzamy na oczekiwaniu na operacje wejścia/wyjścia (I/O bound task), tj. połączenie z bazą danych, zapytania sieciowe, czy odbiór strumienia wideo z kamery przemysłowej … Rozwiązanie to bardzo szybko zyskuje na popularności i coraz więcej bibliotek jest kompatybilnych z tym paradygmatem programowania (np. aiohttp, websockets, aiofiles, databases, więcej o tym tutaj).
Poniżej przykład kodu napisanego przy pomocy asyncio: Możemy zobaczyć, że zdefiniowane funkcje są rzeczywiście wykonywane współbieżnie, uruchamiając poniższą komendę: Przykładowy wydruk programu:
Przy okazji warto zwrócić uwagę na fakt, że funkcje asynchroniczne mogą być wywołane w tradycyjnym kodzie Pythonowym odwołując się bezpośrednio do event loop.
Testowanie
Wraz ze wzrostem liczby użytkowników rośnie też zapotrzebowanie na testowanie tak napisanego kodu. Mimo różnic w składni funkcji asynchronicznych integracja asyncio z frameworkiem pytest wymaga jedynie minimalnych zmian. Najlepiej zobaczymy to na przykładzie testów do funkcji long_computation zdefiniowanej powyżej. Pierwszym naiwnym rozwiązaniem, które przyjdzie nam do głowy może być dodanie słowa kluczowego async do definicji testu tak jak na przykładzie poniżej.
Wykonując ten test okaże się jednak, że pytest go ominął, ponieważ ten framework nie obsługuje domyślnie funkcji asynchronicznych. Aby rozwiązać ten problem możemy odwołać się bezpośrednio do event loop.
Wykorzystując udogodnienia dostępne w pytest (@pytest.fixture), tworzymy event loop oraz po zakończeniu testu gwarantujemy poprawne zwolnienie zasobów. Naszą funkcję asynchroniczną uruchamiamy przy pomocy metody run_until_complete. Uruchamiając ten test przekonamy się, że rzeczywiście działa poprawnie. Takie rozwiązanie zapewnia większą możliwość dostosowywania (np. zwracanie własnej implementacji event_loop), ale możemy również znaleźć prostszy sposób.
Pakiet pytest-asyncio
Po zainstalowaniu pluginu pytest-asyncio mamy do dyspozycji dekorator @pytest.mark.asyncio, który sprawia, że test jest traktowany jako funkcja asynchroniczna. Możemy wtedy korzystać ze słowa kluczowego await, co zwiększa przy okazji czytelność kodu.
Mockowanie
Drugim elementem, który wymaga pewnych modyfikacji podczas testowania funkcji asynchronicznych jest mockowanie (nie będziemy tutaj zagłębiać się w terminologię i różnice między stubami oraz mockami, zainteresowanych odsyłam tutaj). Mockowanie umożliwia odizolowanie zachowania testowanego komponentu bez testowania poprawności działania jego zależności. Domyślnie klasa Mock dostępna w pakiecie unittest nie wspiera składni asyncio, jednak dziedzicząc po tej klasie możemy napisać własną implementację kompatybilną z async/await. Nasz przeciążony operator wywołania powinien zwracać funkcję asynchroniczną.
Context manager patch mockuje naszą funkcję long_computation, co sprawia, że nie będzie ona w rzeczywistości wywołana jako zależność funkcji launch_tasks. W naszym przykładzie, dodatkową zaletą tego rozwiązania jest szybsze wykonywanie testów.
Pakiet asynctest
Tak jak to miało miejsce w przypadku używanego przez nas dekoratora, nie musimy dziedziczyć po klasie Mock. Instalując pakiet asynctest mamy dostęp do klasy CoroutineMock, która spełnia dokładnie to samo zadanie.
Podsumowanie
W ramach tego artykułu omówiliśmy temat testowania kodu napisanego w Pythonie korzystającego z pakietu asyncio oraz pokazaliśmy, że dostosowanie frameworku pytest w tym kierunku nie wymaga od nas wiele pracy. Testując kod asynchroniczny musimy pamiętać, że może on być wykonany jedynie po przydzieleniu zasobów przez event loop. Pytest dzięki swojej architekturze pozwala nam w łatwy sposób dodać wsparcie dla asyncio. Dodatkowo pakiety pytest-asyncio oraz asynctest dostarczają nam niezbędnych narzędzi, co sprawia, że dostosowanie testów do funkcji asynchronicznych staje się prostsze.
