asynctest: Easier unit testing for asyncio code

Martin Richard, alwaysdata

asynctest
Easier unit testing for asyncio code

In a nutshell:

We will discover unit testing and asynctest with an example.

Why and when you want unit tests

you want reliability

you want consistency

Example (code!)

Where are the tests located?

.
├── piedpiper
│   ├── __init__.py
│   ├── network.py
│   └── ...
├── tests
│   ├── __init__.py
│   ├── test_network.py
│   └── ...
├── README
└── setup.py

network.py

class ResourceDownloader:
    def __init__(self, url):
	    ...

    def get_parsed_url(self):
	    ...

    async def download(self):
	    ...

    def refresh(self, period, loop=None):
	    ...
		

test_network.py

import asynctest, piedpier.network

@asynctest.lenient
class Test_ResourceDownloader_get_parsed_url(
        asynctest.TestCase):

    def test_get_parsed_url(self):
        ...
		

TestCase

case = Test_ResourceDownloader_get_parsed_url(
	"test_get_parsed_url")
case.run()

test_network.py

def test_get_parsed_url(self):
	downloader = ResourceDownloader("http://piedpiper.com/foo")
	result = downloader.get_parsed_url()

	self.assertEqual(result,
	                 ("piedpiper.com", 80, "/foo", False))
        

test_network.py

def test_get_parsed_url_errors(self):
	downloader = ResourceDownloader("invalid address")

	with self.assertRaises(ValueError):
	    result = downloader.get_parsed_url()
        

Run it!

$ PYTHONPATH=. python -m unittest tests

.F
======================================================================
FAIL: test_get_parsed_url_errors (test_network.Test_ResourceDownloader_refresh)
----------------------------------------------------------------------
Traceback (most recent call last):
  (...)
  File ".../tests/test_network.py", line 39, in test_get_parsed_url_errors
    downloader.get_parsed_url()
AssertionError: ValueError not raised

----------------------------------------------------------------------
Ran 2 tests in 0.014s

FAILED (failures=1)
		
	

More tests

def test_get_parsed_url(self):
	downloader = ResourceDownloader("http://piedpiper.com/foo")
	result = downloader.get_parsed_url()

	self.assertEqual(result,
	                 ("piedpiper.com", 80, "/foo", False))

	downloader = ResourceDownloader("https://piedpiper.com/")
	result = downloader.get_parsed_url()

	self.assertEqual(result,
	                 ("piedpiper.com", 443, "/", True))
        
import collections
Case = collections.namedtuple("Case", "url expected")

def test_get_parsed_url(self):
    cases = {
        "simple http URL": Case(
            url="http://piedpiper.com/",
            expected=('piedpiper.com', 80, '/', False)),
        "simple https URL": Case(
            url="https://piedpiper.com/",
            expected=('piedpiper.com', 443, '/', True)),
        "URL with port": Case(
            url="http://piedpiper.com:8080/",
            expected=('piedpiper.com', 8080, '/', False)),
        ...
    }
        
import collections
Case = collections.namedtuple("Case", "url expected")

def test_get_parsed_url(self):
    ...

    for name, case in cases.items():
        with self.subTest(name=name, url=case.url):
	        downloader = ResourceDownloader(case.url)
	        result = downloader.get_parsed_url()
	        self.assertEqual(result, case.expected)
        

Example with asyncio

ResourceDownloader.download

async def download(self):
        host, port, query, ssl = self.get_parsed_url()
        reader, writer = await asyncio.open_connection(host, port, ssl=ssl)

        try:
            writer.write(self._build_request(host, query))
            response_headers = await reader.readuntil(b"/r/n/r/n")
            code, payload_size = self._parse_response_headers(response_headers)

            if code != 200:
                raise RuntimeError(
                    "Server answered with unsupported code {}".format(code))

            self.data = await reader.read(payload_size)
        finally:
            writer.close()

        return self.data

A first test of asynchronous code

class Test_ResourceDownloader_download(asynctest.TestCase):

    async def test_download_resource(self):
        downloader = ResourceDownloader(
            "http://piedpiper.com/compression")
        payload = await downloader.download()

        self.assertEqual(payload, b"MiddleOut")
		

Introducing mocks

Use asynctest.mock.Mock

def create_mocks():
	reader = asynctest.mock.Mock(asyncio.StreamReader)
    writer = asynctest.mock.Mock(asyncio.StreamWriter)

    reader.read.return_value = b"MiddleOut"
    reader.readuntil.return_value = b"HTTP/1.1 200 OK\r\n..."

    return reader, writer

Mocks

asynctest.mock.patch

async def test_download_resource(self):
    downloader = ResourceDownloader(
        "http://piedpiper.com/compression")

    with asynctest.mock.patch(
            "asyncio.open_connection",
            side_effect=create_mocks):
        payload = await downloader.download()

    self.assertEqual(payload, b"MiddleOut")

as a decorator

@patch("asyncio.open_connection", side_effect=create_mocks)
async def test_download_resource(self, open_connection_mock):
    ...

Last example:
Let's control time!

setUp, tearDown, addCleanup

class Test_ResourceDownloader_refresh(asynctest.TestCase):
	async def setUp(self):
        self.downloader = ResourceDownloader("http://piedpiper.com/")
        self.call_count = 0

        def set_data_value(*args, **kwargs):
            self.downloader.data = "Payload {}".format(self.call_count)
            self.call_count += 1
            return self.downloader.data

        patch = asynctest.mock.patch.object(
            self.downloader, "download", side_effect=set_data_value)
        patch.start()
        self.addCleanup(patch.stop)

class Test_ResourceDownloader_refresh(asynctest.TestCase):
    async def tearDown(self):
        self.downloader.refresh(None)
        self.downloader = None

Deal with scheduled callbacks

@asynctest.fail_on(active_handles=True)
class Test_ResourceDownloader_refresh(asynctest.ClockedTestCase):
    async def test_refresh(self):
        self.assertEqual("Payload 0", self.downloader.data)
        self.downloader.refresh(5)

        await self.advance(5)
        # 5 seconds after refresh(5) was set, data is updated
        self.assertEqual("Payload 1", self.downloader.data)

        await self.advance(10)
        # Updated two more times.
        self.assertEqual("Payload 3", self.downloader.data)

More about asynctest

Your turn!

Thanks