index • post archive • atom feed syndication feed icon

Integration Testing in Python

2023-03-18

A minimal setup for an endlessly configurable HTTP server for integration testing in Python. Also, opinions on integration testing.

Please Stop Doing Integration Testing

I get that for a lot of cases, usually involving legacy code, integration testing seems like the easier solution. However I just don't see the kind of value in most integration tests that would justify the amount of work spent developing and maintaining huge suites of heavy tests. Unit testing provides a lot of bang for your buck and can force better architectures through the need for something like dependency injection or smaller units/modules/functions. In all of my experience I have yet to come across an integration test which I can attach a debugger to (meaningfully, even in the below example asyncio imposes some annoying limitations on pdb that make it less useful than might be the case with a more focused unit test).

If you really have to integration test something I think it is key to distinguish an integration test from end-to-end tests. Too often I find integration tests that require the entire software stack of messaging servers, databases, mock APIs and then a litany of testing support tools to inject "testability".

Motivation

All that being said, in some recent work I have found aiohttp annoyingly difficult to use. The desire for a simple looking interface results in an awkward API that is difficult to reason about in practice. Add to this a lack of documentation for flows that deviate in the slightest and you're left with a confusing pile of opaque code which is difficult to verify works in all cases. Unit testing my own code could only assert behavior that I was guessing at being produced by this external library.

In this particular case I wanted to answer a question like "does application code using a single aiohttp.ClientSession need to worry about unhandled errors to maintain the internal connection pool?" but these sorts of issues are not unique to aiohttp and the proposed solutions are pretty typical in Python — install more libraries to make things superficially easier. In the case of testing there are additional solutions like running mock servers with varying levels of complexity, usually in something like Docker which means dragging in a whole host of inessential complexity around networking and deployment.

I have written before about my distaste for complexity and unnecessary dependencies. Rather than throw the kitchen sink a 3rd party library or service at the issue I thought I might take some time to try making a minimal set of supporting functionality to perform an integration test with aiohttp.

A Minimal HTTP Server

The Python standard library has a number of different servers available. The interface is a little weird but I think this reflects their vintage more than anything. For my case of creating a tiny HTTP server to test HTTP errors they are totally sufficient. One interesting thing to work out was the right approach to using the standard library servers (synchronous) in a test for aiohttp (asynchronous). I settled on launching the server in a thread and wrapped it in a context manager so the server is closed and the thread cleaned up automatically. 

import asyncio
import contextlib
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

@contextlib.contextmanager
def run_test_server(*, host, port, handler):
    httpd = ThreadingHTTPServer((host, port), handler)
    server_thread = threading.Thread(target=lambda: httpd.serve_forever(poll_interval=0.01))
    server_thread.start()
    try:
        yield httpd
    finally:
        httpd.shutdown()
        httpd.server_close()
        server_thread.join()

Pretty standard stuff! Of course there is no logic or handling of requests yet. In my mind all of the handling will be pretty specific to each test I'll be running and thus defined in the supplied handler. I began my testing of the above using just the standard library so that I can add complexity incrementally (ensure requests work, make the requests async, use aiohttp to perform the requests). Here first is a synchronous HTTP request to PUT some JSON (to be eventually replaced with the "real" code under test): 

import http.client
import json
from urllib.parse import urlparse

def put_json(url, data):
    l = urlparse(url)
    conn = http.client.HTTPConnection(l.netloc)
    conn.request("PUT", l.path, body=json.dumps(data).encode())
    return conn.getresponse()

Finally, here is a synchronous test case that uses an HTTP server which always returns an HTTP 418: 

import unittest

class MyTestcase(unittest.TestCase):
    def test_request_error(self):

        class MyHandler(BaseHTTPRequestHandler):
            def do_PUT(self):
                length = int(self.headers['content-length'])
                body = self.rfile.read(length)
                self.send_error(418, message='simulated HTTP 418')

        with run_test_server(host='127.0.0.1', port=9999, handler=MyHandler):
            resp = put_json('http://127.0.0.1:9999/some/path', {'fizz': 'buzz'})
            self.assertEqual(resp.status, 418)

Hopefully obvious, it works in an unsurprising way: 

$ python -m unittest test.py
127.0.0.1 - - [18/Mar/2023 14:25:44] code 418, message simulated HTTP 418
127.0.0.1 - - [18/Mar/2023 14:25:44] "PUT /some/path HTTP/1.1" 418 -
.
----------------------------------------------------------------------
Ran 1 test in 0.004s

OK

The next piece in ensuring this test-local HTTP server is fit for testing aiohttp is to make the test case async. I first just wrapped my synchronous request in an asyncio thread and added a list to communicate responses and assertions through. This is a little weird but reflects how I am using this put_json fake rather than "real" application code. For my case I had further assertions to make and application hooks to verify behavior through. 

def put_json(url, data, receiver):
    l = urlparse(url)
    conn = http.client.HTTPConnection(l.netloc)
    conn.request("PUT", l.path, body=json.dumps(data).encode())
    resp = conn.getresponse()
    receiver.append(resp)
    return resp

class MyTestcase(unittest.IsolatedAsyncioTestCase):
    async def test_request_error(self):
        receiver = []

        class MyHandler(BaseHTTPRequestHandler):
            def do_PUT(self):
                length = int(self.headers['content-length'])
                body = self.rfile.read(length)
                self.send_error(418, message='simulated HTTP 418')

        with run_test_server(host='127.0.0.1', port=9999, handler=MyHandler):
            await asyncio.to_thread(lambda: put_json('http://127.0.0.1:9999/some/path', {'fizz': 'buzz'}, receiver))
            self.assertEqual(receiver[0].status, 418)

Things still working as expected, synchronous and asynchronous code mixed to no ill effect. Time to throw in aiohttp: 

import aiohttp

async def function_under_test():
    payload = json.dumps({'fizz': 'buzz'}).encode()
    async with aiohttp.ClientSession() as session:
        async with session.put('http://127.0.0.1:9999/some/path',
                               data=payload) as response:
            return response

class MyTestcase(unittest.IsolatedAsyncioTestCase):
    async def test_request_error(self):

        class MyHandler(BaseHTTPRequestHandler):
            def do_PUT(self):
                length = int(self.headers['content-length'])
                body = self.rfile.read(length)
                self.send_error(418, message='simulated HTTP 418')

        with run_test_server(host='127.0.0.1', port=9999, handler=MyHandler):
            resp = await function_under_test()
            self.assertEqual(resp.status, 418)

This is sufficient to answer my motivating question about connection pool maintenance in the face of all manner of byzantine server responses (and a lack of response).

Thoughts

All of the above is obviously a bit pared down from my actual use but that is only for having omitted the real application under test. I have found it pleasantly surprising how easy it was to develop the idea and throw new exceptional behaviors into test cases. Want a server response that is conditional on request data? A server that only produces a response in some circumstances? A malformed JSON response? Each is little more than the 5 lines that define the handler above. More than that though, reading such a test case leaves very little to the imagination which may be my real issue with bigger solutions. Having to read a test case like: 

    async def test_request_fizzbuzz_gone(self):

        class MyHandler(BaseHTTPRequestHandler):
            def do_PUT(self):
                length = int(self.headers['content-length'])
                body = self.rfile.read(length)
                req = json.loads(body.decode())
                if req['fizzbuzz'] % 2 == 0:
                    self.send_error(418, message='simulated HTTP 418')
                else:
                    self.send_error(410, message='simulated HTTP 410')

        with run_test_server(host='127.0.0.1', port=9999, handler=MyHandler):
            resp = await function_under_test()
            self.assertEqual(resp.status, 410)

It may not make a lot of sense as a test case, but I'm not stuck chasing down Dockerfiles or 3rd party libraries to untangle what the server mock is returning or has been configured to do. More than scriptable, the test server is just part of the test code.

One concern I had with embedding an HTTP server into the individual test was that I might run into port conflicts across tests if the context manager somehow held the port. I threw a thousand tests into a test suite and was unable to elicit any such problems, it all just quietly worked. I still think most integration testing is a bad idea but for a few narrow cases something like this seems dramatically simpler and easier to work with. For my case that has meant 1-10% of the code and zero configuration. The execution overhead is negligible for any of the kinds of tests I am doing and my test suite can run faster than something like a Wiremock docker container is likely to even launch.

A big part of Python's popularity comes down to the availability of libraries for any use. Before the explosion in 3rd party libraries though Python was popular for its robust standard library. While it may be passé to use something like http.client when you could install requests aiohttp I think it is valuable to remember it is always available and there is remarkably less churn in the standard library. Even more important though, just doing things yourself can let you radically simplify the problem because there is only one problem to solve. It can feel sometimes like there are simultaneously too many tool-makers and too few people willing to forego tools in favor of just writing the code that solves a problem.