W tym artykule omawiamy podstawy testowania kodu asynchronicznego na przykładzie klasy Worker zaimplementowanej w kodzie crawlera.

Omówiony i zamieszczony poniżej kod dostępny jest w repozytorium git.

Klasa Worker

Klasa Worker dziedziczy z bazowej klasy Base. Klasę Base testowaliśmy w poprzedniej odsłonie, ale dla jasności przypominamy tutaj krótki kod obu tych klasy. W architekturze naszego crawlera klasa Worker również pełni rolę klasy bazowej dla innych typów. Jej charakterystyka w kilku punktach:

• jest klasą abstrakcyjną
• wymusza w klasach pochodnych implementację metody _run() (dekorator @abc.abstractmethod)
• implementuje metodę szablonową __call__() (operator wywołania). W innych językach metoda byłaby oznaczona jako final, interpreter Python jednak nie dostarcza takiego słowa kluczowego ani mechanizmu, choć istnieje dekorator dedykowany dla linterów/analizy statycznej¹.
• pozwala typom pochodnym na dodatkową asynchroniczną inicjalizację: metoda initialize()
• zapewnia i gwarantuje wykonanie asynchronicznego uprzątnięcia: metoda shutdown()

class Base(abc.ABC):
    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger(str(self))
        self.logger.debug("Initialized")

    def __str__(self):
        return self.__class__.__name__


class Worker(Base):
    async def initialize(self, *args, **kwargs): ...
    async def shutdown(self, *args, **kwargs): ...

    async def __call__(self, *args, **kwargs):
        self.logger.debug("Running")
        await self.initialize(*args, **kwargs)
        result = None
        try:
            result = await self._run(*args, **kwargs)
        finally:
            await self.shutdown(*args, **kwargs)
        self.logger.debug("Done")
        return result

    @abc.abstractmethod
    async def _run(self, *args, **kwargs): ...

Template method:
Metoda __call__() jest metodą szablonową. Tutaj "szablon" to uruchomienie dodatkowej inicjalizacji, wywołanie wymuszonej implementacji _run(), gwarantowane wywołanie shutdown(). Pochodne typy nie implementują już własnego __call__(). Każdy typ pochodny Worker działa zatem w ten sam sposób, a właściwe sobie operacje implementuje w metodach, które __call__() wywołuje. Taki "dekorator OOP".

Oczywiście - __call__() nie jest final, więc MOŻNA w razie potrzeby przesłonić Worker.__call__(), zrobić coś po swojemu, a nawet wywołać po tym super().__call__(). Wszystko jest możliwe.

Zakres testów

Będziemy chcieli przetestować:

• czy klasa Worker jest abstrakcyjna
• czy typy pochodne implementujące _run() mogą być tworzone
• czy logowanie komunikatów działa poprawnie
• czy _run() jest wywoływane
• czy _initialize() jest wywoływane
• czy _shutdown() jest wywoływane zawsze (nawet jeśli wystąpi wyjątek)
• czy __call__() zwraca wyniki wywołania metody _run() klas pochodnych

Szkielet testu: IsolatedAsyncioTestCase

Biblioteka unittest dostarcza nam bazową klasę dla asynchronicznych testów jednostkowych: IsolatedAsyncioTestCase. Klasa ta rozszerza znany nam już interfejs TestCase. Dodaje jednak m.in:

• dodanie asyncSetUp()
• dodanie asyncTearDown()
• uruchamianie asynchronicznych metod testów jednostkowych (async def test_anything())

Metoda asyncSetUp() pozwala nam na inicjalizację kodu wymagającego interfejsu asynchronicznego. Uruchamiana jest PO zwykłym setUp(). Metoda asyncTearDown() wykonywana jest PRZED tearDown().
W przeciwieństwie do API synchronicznego nie mamy odpowiedników metod klasy:

• setUpClass() NIE POSIADA odpowiednika asyncSetUpClass()
• tearDownClass() NIE POSIADA odpowiednika asyncTearDownClass()

Podstawowe zachowanie testu możemy zaprezentować w następujący sposób:

import unittest


class TestCase(unittest.IsolatedAsyncioTestCase):
    @classmethod
    def setUpClass(cls):
        print("setUpClass")

    @classmethod
    def tearDownClass(cls):
        print("tearDownClass")

    def setUp(self):
        print("setUp")

    async def asyncSetUp(self):
        print("asyncSetUp")

    async def asyncTearDown(self):
        print("asyncTearDown")

    def tearDown(self):
        print("tearDown")


class TestMe(TestCase):
    def test_sync_case(self):
        print("*" * 16, "SYNC CASE")

    async def test_async_case(self):
        print("*" * 16, "ASYNC CASE")

Ot, "Template Method". To samo robimy w Worker (initialize/shutdown).

Testy uruchamiamy w standardowy sposób:

$ python -m unittest test_async.py 
setUpClass
setUp
asyncSetUp
**************** ASYNC CASE
asyncTearDown
tearDown
.setUp
asyncSetUp
**************** SYNC CASE
asyncTearDown
tearDown
.tearDownClass

----------------------------------------------------------------------
Ran 2 tests in 0.011s

OK

Testujemy

Podstawowe: abc.ABC, typy pochodne, logowanie

Interfejs klasy abstrakcyjnej możemy przetestować w łatwy sposób - próba stworzenia obiektu takiej klasy zakończy się wyjątkiem TypeError. Takie testy analizowaliśmy poprzednio. Do innych przypadków potrzebujemy jednak móc stworzyć jakiś obiekt. Posłużymy się zdefiniowanym w teście pomocniczym typem pochodnym klasy Worker. Wystarczy taki stworzyć. W pierwszej fazie nie używamy jeszcze kodu asynchronicznego.

import unittest
from crawler import Worker, Base


class MyWorker(Worker):
    async def initialize(self, *args, **kwargs): ...
    async def shutdown(self, *args, **kwargs): ...
    async def _run(self, *args, **kwargs): ...


class TestWorker(unittest.TestCase):
    def test_abc(self): ...
    def test_init(self): ...
    def test_logger(self): ...
    def test_callable(self): ...

Uzupełniamy testy. Pierwszy przypadek: sprawdzamy czy próba bezpośredniego utworzenia obiektu typu Worker zakończy się wyjątkiem TypeError.

class TestWorker(unittest.TestCase):
    def test_abc(self):
        with self.assertRaises(TypeError):
            Worker()

Jeśli tworzymy obiekt dodanego do celów testowych typu MyWorker, który implementuje _run(), stworzenie obiektu powinno odbyć się poprawnie.

    ...
    def test_init(self):
        obj = MyWorker()
        self.assertIsInstance(obj, Worker, "Worker")
        self.assertIsInstance(obj, Base, "Base")

Worker dziedziczy z Base, w inicjalizatorze tworzymy obiekt loggera, zatem krótki test wystarczy. Sprawdzamy nazwę tego loggera.

    ...
    def test_logger(self):
        with self.assertLogs(level=logging.DEBUG) as logs:
            MyWorker()
        self.assertEqual(len(logs.records), 1)
        self.assertEqual(logs.records[0].name, "MyWorker", "Logger name")

Sprawdzamy też, czy obiekty typu pochodnego Worker będziemy mogli bezpośrednio wywoływać (implementacja __call__()):

    ...
    def test_callable(self):
        obj = MyWorker()
        self.assertTrue(callable(obj), "Implements __call__")

Uruchamiamy testy dla tych przypadków:

$ python -m unittest tests/test_worker.py -k TestWorker -v
test_abc (tests.test_worker.TestWorker) ... ok
test_callable (tests.test_worker.TestWorker) ... ok
test_init (tests.test_worker.TestWorker) ... ok
test_logger (tests.test_worker.TestWorker) ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.008s

OK

AsyncMock

AsyncMock z biblioteki unittest.mock jest zgodny ze zwykłym Mock. Możemy tworzyć makiety wszelkich obiektów, podmieniać funkcje, ich rezultaty, zliczać wywołania funkcji i sprawdzać argumenty poszczególnych wywołań. Możemy podstawiać również funkcje zawierające jakiś efekt uboczny (side effect), np. zgłaszanie wyjątku. Jeśli nie znamy samego Mock, czy MagicMock, to zaraz poznamy. Klasy mają identyczny interfejs i zachowują się identycznie, a AsyncMock po prostu może być używany z korutynami (funkcje async def ...()).

Testy operacji: IsolatedAsyncioTestCase

Szkielet testu:

class TestCallOperator(unittest.IsolatedAsyncioTestCase):
    def setUp(self): ...
    async def test_initialize(self): ...
    async def test_shutdown(self): ...
    async def test_shutdown_guarantee(self): ...
    async def test_run(self): ...
    async def test_run_retval(self): ...

setUp()

W metodzie setUp() przygotowujemy sobie argumenty pozycyjne i nazwane, których będziemy używać przy wywołaniach:

class TestCallOperator(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        super().setUp()
        self.postitional_args = (1, 2, "test", None, ["abc"])
        self.keyword_args = {"one": 1, "two": 2}

initialize()

Wykorzystamy tutaj przygotowaną na samym początku pomocniczą klasę MyWorker, a pod metodę initialize() podstawimy AsyncMock. Nie będziemy jej wywoływać bezpośrednio, a skorzystamy po prostu z faktu, że Worker implementuje operator wywołania __call__(), w którym initialize() ma zostać wywołane. Jeśli __call__() jest zaimplementowane poprawnie, to wywoła jeden raz podstawione przez nas initialize(), a my sprawdzimy sobie czy było faktycznie tylko jedno wywołanie i czy do initialize() przekazane były wszystkie argumenty (te pozycyjne i te nazwane).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

    ...
    async def test_initialize(self):
        obj = MyWorker()
        obj.initialize = AsyncMock()
        await obj(*self.postitional_args, **self.keyword_args)
        self.assertEqual(obj.initialize.call_count, 1, "Calls initialize()")
        obj.initialize.assert_called_once_with(
            *self.postitional_args,
            **self.keyword_args,
        )

• W linii 4 nadpisujemy initialize() obiektem AsyncMock(), tak po prostu.
• W linii 5 wywołujemy obiekt z argumentami (implementuje __call__() więc możemy go wywołać jak funkcję /jest callable, a to już wcześniej testowaliśmy/).
• W linii 6 sprawdzamy czy nasza makieta była wywołana tylko jeden raz.
• W linii 7 wykorzystujemy fakt, że AsyncMock ma kilka dodatków - pozwala wykonać asercję (test) sprawdzającą czy makieta została wywołana z zadanymi argumentami.

Jeśli Worker implementuje poprawnie __call__(), to initialize() jest wywoływane tylko jeden raz i otrzymuje wszystkie argumenty z wywołania samego __call__().

shutdown()

Tutaj jeszcze postępujemy identycznie jak przy initialize(). Mockujemy metodę shutdown(), wywołujemy obiekt workera, sprawdzamy czy shutdown() się odbył i czy otrzymał wszystkie argumenty wywołania obiektu. Różnicą jest tutaj tylko shutdown() zamiast initialize().

    async def test_shutdown(self):
        obj = MyWorker()
        obj.shutdown = AsyncMock()
        await obj(*self.postitional_args, **self.keyword_args)
        self.assertEqual(obj.shutdown.call_count, 1, "Calls shutdown()")
        obj.shutdown.assert_called_once_with(
            *self.postitional_args,
            **self.keyword_args,
        )

shutdown() + side_effect

W tym przypadku sprawdzamy czy metoda shutdown() zostanie wywołana jeśli _run() zgłosi wyjątek. Worker pozwala na "wywrotkę" przy initialize() (jeśli coś nie może działać, niech wyskoczy wyjątek i niech wszyscy go zobaczą!). Później jednak __call__() ma zadbać o to, aby shutdown() się odbył.
Jeśli coś zainicjalizowaliśmy, to chcemy to też zwolnić (pozamykać ewentualne pliki, połączenia, pozbierać ewentualne zadania (asyncio.Task), itd.

Worker uruchamia _run() w bloku try/except, a shutdown() wywoływany jest w bloku finally, zatem zawsze. Dlatego mockujemy zarówno shutdown(), jak i _run(). Oba będą obiektami AsyncMock. W _run() wstawimy side_effect w postaci wyjątku. Kiedy metoda _run() zostanie wywołana, nastąpi zgłoszenie wyjątku. Typ wyjątku nie ma znaczenia - worker ma wykonać blok finally, a samego wyjątku nie obsługiwać. Jeśli wykona blok finally, to nasza makieta podpięta pod shutdown() zostanie wywołana. Sprawdzimy wtedy call_count jak robiliśmy już wcześniej. Stosujemy również TestCase.assertRaises(), by upewnić się, że wyjątek nie został stłumiony w __call__().

    ...
    async def test_shutdown_guarantee(self):
        obj = MyWorker()
        obj._run = AsyncMock()
        obj._run.side_effect = SyntaxError
        obj.shutdown = AsyncMock()

        with self.assertRaises(SyntaxError):
            await obj(*self.postitional_args, **self.keyword_args)

        self.assertEqual(obj.shutdown.call_count, 1, "Calls shutdown()")

_run()

W tym przypadku mockujemy metodę _run() i sprawdzamy:

• czy metoda __call__() prawidłowo loguje "Running" oraz "Done" kiedy logger ma poziom DEBUG
• czy metoda _run() jest wywoływana tylko raz
• czy _run() otrzymuje wszystkie argumenty wywołania obiektu

    ...
    async def test_run(self):
        obj = MyWorker()
        obj._run = AsyncMock()

        # calling object (__call__) calls run()
        with self.assertLogs(level=logging.DEBUG) as logs:
            await obj(*self.postitional_args, **self.keyword_args)

        # check logs
        self.assertEqual(len(logs), 2)
        self.assertEqual(logs.records[0].message, "Running")
        self.assertEqual(logs.records[1].message, "Done")

        # check run()
        self.assertTrue(obj._run.called, "Calls run")
        self.assertEqual(obj._run.call_count, 1, "Calls run only once")

        # check all passed arguments
        obj._run.assert_called_once_with(
            *self.postitional_args,
            **self.keyword_args,
        )

_run() - return value

Na koniec sprawdzamy czy wywołanie obiektu (__call__()) zwraca to, co zwróciło _run(). Tworzymy makietę dla _run() i podstawiamy tej metodzie wartość jaka miałaby być zwrócona. Tutaj napis "Hello".

   async def test_run_retval(self):
        value = "Hello"
        obj = MyWorker()
        obj._run = AsyncMock()
        obj._run.return_value = value
        retval = await obj()
        self.assertEqual(retval, value, "Returned valued")

Uruchomienie testów

Testy gotowe, uruchamiamy:

$ python -m unittest tests/test_worker.py  -v
test_initialize (tests.test_worker.TestCallOperator) ... ok
test_run (tests.test_worker.TestCallOperator) ... ok
test_run_retval (tests.test_worker.TestCallOperator) ... ok
test_shutdown (tests.test_worker.TestCallOperator) ... ok
test_shutdown_guarantee (tests.test_worker.TestCallOperator) ... ok
test_abc (tests.test_worker.TestWorker) ... ok
test_callable (tests.test_worker.TestWorker) ... ok
test_init (tests.test_worker.TestWorker) ... ok
test_logger (tests.test_worker.TestWorker) ... ok

----------------------------------------------------------------------
Ran 9 tests in 0.033s

OK

Szybko, sprawnie, bezboleśnie.

Podsumowanie

Składnia testów jest bardzo prosta. Szczególnie łatwo komponuje się testy w sposób obiektowy. Ja osobiście dlatego unikam pytest. Myślę raczej obiektowo, wiem kiedy co się inicjalizuje, nie grzęznę w magii dekoratorów, plikach konfiguracyjnych, a unittest jest w bibliotece standardowej Python. Jedna zależność mniej: +1 do szczęścia. Znacznie większym problemem w dziedzinie testowania jest wymyślenie strategii jak dany test zrealizować. Musimy mieć jasne założenia i rozumieć co kod ma robić w danych sytuacjach. Biblioteka unittest dostarcza nam wszelkie mechanizmy do sprawnego i wygodnego pisania testów, wraz z podstawianiem makiet w miejscach "trudnych" i "ciekawych".

W kolejnej odsłonie zajmiemy się testowaniem Schedulera. Tam będzie nieco więcej zagadnień, ponieważ scheduler działa jako zadanie "w tle" (asyncio.Task). Dojdzie zatem konieczność uruchomienia zadania, które robi coś po swojemu w dowolnym czasie. Czas² będzie jedną z kluczowych spraw w naszych operacjach asynchronicznych.