asynctest: Easier unit testing for asyncio code

`asynctest`
Easier unit testing for asyncio code

In a nutshell:

We will discover unit testing and asynctest with an example.

Why write unit tests,
look at a test case from an almost real-world example,
show you how asynctest adds asyncio support to unittest.

Why and when you want unit tests

you want reliability

reliability = 1 / anxiety (said someone at Google)
gives you confidence in your work, using facts
detect obvious mistakes like typos
prevent regressions during refactoring
you're less affraid of changing code (especially old and complex code)

you want consistency

spot inconcistencies, coupling and side effects
exercise your API by using it in your tests
assess the useability and readability of your code
ensure backward compatibility
you'll write a better API

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()

Instance of a test case: situation we want to test and verify the outcome
We use a test runner which creates and runs an instance for each test_* method

`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)
		
			Now we can invoke the test runner. PYTHONPATH=.
			tells the python interpreter where to look for our package,
			-m unittest invokes the standard unittest
			package (which has a "main" which is the test runner). The last
			argument is the package in which our test cases are located.
			The result here shows that our second test failed, and gives us
			some details.
			The first dot correspond to the first test, which ran
			successfully.

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")

what if piedpiper.com is slow or unavailable?

Introducing mocks

replacement of functions, classes or objects
mock ("act like") the object they replace
their behavior can be configured
provide assertion methods to check how the mock was used

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

each method and attribute is a mock,
a function call on a mock returns a mock, unless configurated,
asynctest mocks coroutines automatically.

`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):
    ...

can decorate a TestCase, or any function or coroutine
patch.multiple, patch.dict, patch.object
advanced features: control the scope of the patch in a coroutine

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

Called for each test case,
only runs if the test did not fail,
cleanup functions are always called,
can be functions or coroutines.

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`

Future:
- support of asynchronous iterators and context managers,
- mocking of IO (sockets, files, pipes, ...),
- support of Windows' proactor (maybe?).
pytest-asyncio: less features, but you can use asynctest with it
Users: alwaysdata, folks at Mozilla and Cisco, you?

Your turn!

Write how to run tests in your README or CONTRIBUTING file
Make sure tests are easy to run:
$ git clone
$ python -m unittest tests (or nose, pytest, tox, ...)
You can now enable travis-ci!

Thanks

asynctest.readthedocs.org and github.com/Martiusweb/asynctest
I'm Martin, come talk to me!
martius@martiusweb.net
https://marti.us
Twitter, github, etc : Martiusweb
slides are at: https://marti.us/t/pyconfr-2016/
Full example: https://marti.us/t/pyconfr-2016/asynctest_example.py

asynctestEasier unit testing for asyncio code

In a nutshell:

Why and when you want unit tests

you want reliability

you want consistency

Example (code!)

Where are the tests located?

network.py

test_network.py

TestCase

test_network.py

test_network.py

Run it!

More tests

Example with asyncio

ResourceDownloader.download

A first test of asynchronous code

Introducing mocks

Use asynctest.mock.Mock

Mocks

asynctest.mock.patch

as a decorator

Last example:Let's control time!

setUp, tearDown, addCleanup

Deal with scheduled callbacks

More about asynctest

Your turn!

Thanks

`asynctest`
Easier unit testing for asyncio code

`network.py`

`test_network.py`

`TestCase`

`test_network.py`

`test_network.py`

`ResourceDownloader.download`

Use `asynctest.mock.Mock`

`asynctest.mock.patch`

Last example:
Let's control time!

`setUp`, `tearDown`, `addCleanup`

More about `asynctest`