httptest

受golang的httptest包启发的HTTP测试。支持封装异步协程函数 (async def)。

用法

上下文管理器

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

简单的HTTP服务器处理器

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

提供文件

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

异步支持

Python的unittest包尚未包含异步支持。 python/issue32972。它应该在3.8中实现，您可以在这里查看这里。

如果您想快速添加asyncio测试用例，您可以从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()

在您的项目的setup.py中，将dffml添加到tests_require。

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

自动安装

如果您正在制作Python包，您希望将httptest添加到您的setup.py文件中的tests_require部分。

这样，当您运行python setup.py test时，setuptools将为您在本地目录中安装httptest，如果尚未安装。

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

OIDC服务器

启动服务器 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

发起请求

$ 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

缓存服务器

运行缓存服务器，并使用其URL替换您想要拦截的上游URL。

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

检查缓存目录中的缓存对象。

$ 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/目录中的更多示例。

• github_webhook_event_logger.py