httptest

httptest

HTTP testing inspired by golang's httptest package. Supports wrapping asyncio coroutine functions (async def).

Usage

Context Manager

import unittest
import urllib.request

import httptest

class TestHTTPServer(httptest.Handler):

    def do_GET(self):
        contents = "what up".encode()
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.send_header("Content-length", len(contents))
        self.end_headers()
        self.wfile.write(contents)

def main():
    with httptest.Server(TestHTTPServer) as ts:
        with urllib.request.urlopen(ts.url()) as f:
            assert f.read().decode('utf-8') == "what up"

if __name__ == '__main__':
    main()

Simple HTTP Server Handler

import unittest
import urllib.request

import httptest

class TestHTTPServer(httptest.Handler):

    def do_GET(self):
        contents = "what up".encode()
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.send_header("Content-length", len(contents))
        self.end_headers()
        self.wfile.write(contents)

class TestHTTPTestMethods(unittest.TestCase):

    @httptest.Server(TestHTTPServer)
    def test_call_response(self, ts=httptest.NoServer()):
        with urllib.request.urlopen(ts.url()) as f:
            self.assertEqual(f.read().decode('utf-8'), "what up")

if __name__ == '__main__':
    unittest.main()

Serve Files

import pathlib
import unittest
import http.server
import urllib.request

import httptest

FILE_PATH = pathlib.Path(__file__)

class TestHTTPTestMethods(unittest.TestCase):

    @httptest.Server(
        lambda *args: http.server.SimpleHTTPRequestHandler(
            *args, directory=FILE_PATH.parent
        )
    )
    def test_call_response(self, ts=httptest.NoServer()):
        with urllib.request.urlopen(ts.url() + FILE_PATH.name) as f:
            self.assertEqual(f.read().decode('utf-8'), FILE_PATH.read_text())

if __name__ == '__main__':
    unittest.main()

Asyncio Support

Asyncio support for the unittest package hasn't yet landed in Python. python/issue32972. It should land in 3.8, check it out here.

If you want a quick way to add asyncio test cases you can import the helper from intel/dffml.

import sys
import unittest
import urllib.request
if sys.version_info.minor == 3 \
        and sys.version_info.minor <= 7:
    from dffml.util.asynctestcase import AsyncTestCase
else:
    # In Python 3.8
    from unittest import IsolatedAsyncioTestCase as AsyncTestCase

import httptest

class TestHTTPServer(httptest.Handler):

    def do_GET(self):
        contents = "what up".encode()
        self.send_response(200)
        self.send_header("Content-type", "text/plain")
        self.send_header("Content-length", len(contents))
        self.end_headers()
        self.wfile.write(contents)

class TestHTTPTestMethods(AsyncTestCase):

    @httptest.Server(TestHTTPServer)
    async def test_call_response(self, ts=httptest.NoServer()):
        with urllib.request.urlopen(ts.url()) as f:
            self.assertEqual(f.read().decode('utf-8'), "what up")

if __name__ == '__main__':
    unittest.main()

In your project's setup.py, add dffml in tests_require.

setup(
    name='your_package',
    ...
    tests_require=[
        'httptest>=0.1.0',
        'dffml>=0.4.0.post0'
    ]
)

Auto Install

If you're making a python package, you'll want to add httptest to your setup.py file's tests_require section.

This way, when your run python setup.py test setuptools will install httptest for you in a package local directory, if it's not already installed.

setup(
    name='your_package',
    ...
    tests_require=[
        'httptest>=0.1.0'
    ]
)

OIDC Server

Start server httptest-oidc

$ python -m httptest.oidc \
    --issuer https://this-service.example.com \
    --audience https://relying-party.example.com \
    --addr 0.0.0.0 \
    --port 8000 \
    --subject test-subject \
    --private-key-pem-path private-key.pem \
    --token-path token.jwt

Make requests

$ curl -H "Authorization: Bearer $(curl https://this-service.example.com/token | jq -r token)" -v https://relying-party.example.com
$ curl -H "Authorization: Bearer $(cat token.jwt)" -v https://relying-party.example.com

Cache Server

Run the caceh server and use it's URL in place of the upstream URL whatever you want to intercept on

$ httptest-cache --state-dir .cache/httptest --addr 0.0.0.0 --port 7000 http://localhost:8000
Serving on http://localhost:7000

Inspect cached objects in the cache dir

$ python -c 'import sys, pathlib, pickle, pprint; pprint.pprint(pickle.loads(pathlib.Path(sys.argv[-1]).read_bytes()).headers)' .cache/httptest/f31bc77712e808fffdab85a33631e414f25715588b1a026d6b8a4e0171b67e99859ab71b1933c93b0078d1e47da9a929.request.pickle
{'Accept': '*/*',
 'Accept-encoding': 'gzip',
 'Connection': 'close',
 'Content-length': '8159',
 'Content-type': 'application/json',
 'Host': 'localhost:45709',
 'Request-hmac': '1700084205.d96d4f546acedddc142b1168642a74c738685d1ac4aa07984e9a1850bb73ddee',
 'User-agent': 'GitHub-Hookshot/dc69923',
 'X-as': '',
 'X-country': '',
 'X-forwarded-for': '10.56.101.48',
 'X-forwarded-proto': 'https',
 'X-github-delivery': '12dac8d6-83ff-11ee-97c9-119c09045ae0',
 'X-github-event': 'push',
 'X-github-hook-id': '443288828',
 'X-github-hook-installation-target-id': '621131680',
 'X-github-hook-installation-target-type': 'repository',
 'X-github-request-id': '1271:336E:974AC:15C2D2:655539ED',
 'X-glb-edge-region': 'iad',
 'X-glb-edge-site': 'ash1-iad',
 'X-glb-via': 'hostname=glb-proxy-1c66317.ash1-iad.github.net site=ash1-iad '
              'region=iad service=kube-public t=1700084205.902',
 'X-haproxy-ssl-fc-cipher': 'TLS_AES_128_GCM_SHA256',
 'X-haproxy-ssl-fc-protocol': 'TLSv1.3',
 'X-haproxy-ssl-fc-use-keysize': '128',
 'X-real-ip': '10.56.101.48',
 'X-region': '',
 'X-request-start': 't=1700084205915930',
 'X-ssl-ja3-hash': '7a15285d4efc355608b304698cd7f9ab'}

Examples

See the examples/ directory for more examples.

• github_webhook_event_logger.py