GitLocker: The Coding Marketplace

Description:

httptest 2.1.1

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