Integration Testing
Integration Testing
An integration test runs two real pieces together: a function and the filesystem, or a client and an HTTP handler. The boring default is still a process that starts, checks, and exits. Do not leave serve_forever running. Prefer urllib against a function when there is no socket yet. When you do need a server, bind port 0, start a thread, request, then shutdown.
Mental model
Unit tests pin one function. Integration tests pin a seam you could not fake honestly: JSON on disk, an HTTP status line, a temp directory layout.
tempfile.TemporaryDirectory (or pytest’s tmp_path) is the desk folder. http.server.HTTPServer is the socket. urllib.request.urlopen is the client. Join the thread. Close the server. The test process must return.
If the handler is a function, call the function. A TCP server is for “we actually speak HTTP,” not for “I heard integration tests need ports.”
Worked examples
Case 1: urllib against a builder — no socket
Save as ticket_url.py. The client that will later hit the network should still be testable as string in, string out.
# ticket_url.py
from urllib.parse import urlencode, urljoin
from urllib.request import Request
def ticket_request(base, ticket_id):
url = urljoin(base, "ticket")
query = urlencode({"id": ticket_id})
return Request(f"{url}?{query}", method="GET")
def main():
req = ticket_request("http://desk.local/", "T-11")
print(req.full_url)
print(req.get_method())
if __name__ == "__main__":
main()Run:
uv run python ticket_url.pyOutput:
http://desk.local/ticket?id=T-11
GET
# test_ticket_url.py
from ticket_url import ticket_request
def test_ticket_request_query():
req = ticket_request("http://desk.local/", "T-11")
assert req.full_url == "http://desk.local/ticket?id=T-11"
assert req.get_method() == "GET"Run:
uv run --with pytest pytest test_ticket_url.py -qOutput (duration varies):
. [100%]
1 passed in 0.01s
This is the test you write first. The socket test is next, not instead.
Case 2: A local server that shuts down
Save as desk_http.py. Port 0 means “pick a free one.” The thread runs serve_forever. urlopen hits it. shutdown unblocks the thread. join waits. The process exits.
# desk_http.py
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
from urllib.request import urlopen
class TicketHandler(BaseHTTPRequestHandler):
def do_GET(self):
body = b'{"id": "T-11", "table": 4}'
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format, *args):
return
def serve():
server = HTTPServer(("127.0.0.1", 0), TicketHandler)
thread = Thread(target=server.serve_forever)
thread.start()
return server, thread
def fetch_ticket(url):
with urlopen(url, timeout=2) as resp:
return resp.status, resp.read().decode()
def main():
server, thread = serve()
host, port = server.server_address
try:
status, body = fetch_ticket(f"http://{host}:{port}/ticket")
print(status)
print(body)
finally:
server.shutdown()
thread.join(timeout=2)
if __name__ == "__main__":
main()Run:
uv run python desk_http.pyOutput:
200
{"id": "T-11", "table": 4}
# test_desk_http.py
from desk_http import fetch_ticket, serve
def test_ticket_endpoint():
server, thread = serve()
host, port = server.server_address
try:
status, body = fetch_ticket(f"http://{host}:{port}/ticket")
assert status == 200
assert "T-11" in body
finally:
server.shutdown()
thread.join(timeout=2)Run:
uv run --with pytest pytest test_desk_http.py -qOutput (duration varies):
. [100%]
1 passed in 0.05s
finally runs on failure too. Without it, a failed assert leaves a thread and a socket behind.
Case 3: A temp-dir ticket store
Save as ticket_store.py. JSON files under a directory. The integration is “filesystem plus encode.”
# ticket_store.py
import json
from pathlib import Path
def save_ticket(root, ticket):
path = Path(root) / f"{ticket['id']}.json"
path.write_text(json.dumps(ticket) + "\n")
return path
def load_ticket(root, ticket_id):
path = Path(root) / f"{ticket_id}.json"
return json.loads(path.read_text())
def main():
import tempfile
with tempfile.TemporaryDirectory() as root:
save_ticket(root, {"id": "T-11", "table": 4})
print(load_ticket(root, "T-11"))
if __name__ == "__main__":
main()Run:
uv run python ticket_store.pyOutput:
{'id': 'T-11', 'table': 4}
# test_ticket_store.py
from ticket_store import load_ticket, save_ticket
def test_round_trip(tmp_path):
save_ticket(tmp_path, {"id": "T-11", "table": 4})
ticket = load_ticket(tmp_path, "T-11")
assert ticket["table"] == 4
assert (tmp_path / "T-11.json").is_file()Run:
uv run --with pytest pytest test_ticket_store.py -qOutput (duration varies):
. [100%]
1 passed in 0.01s
tmp_path is already unique per test. You do not need a second TemporaryDirectory inside the test.
The trap
This fragment never returns. Do not run it. serve_forever blocks the main thread. There is no shutdown, no urlopen, no join.
# do_not_run_forever.py
from http.server import BaseHTTPRequestHandler, HTTPServer
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.end_headers()
def main():
HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()
if __name__ == "__main__":
main()The fix is Case 2: a thread, a request, shutdown, join. Bind 127.0.0.1 and port 0. Leave port 8000 for humans.
The boring rule
- Prefer
urllib(parse, request objects,urlopen) against a function until a socket is the point. - Bind
127.0.0.1and port0. Set a timeout onurlopen. shutdownandjoininfinally. The process must exit.tmp_path/TemporaryDirectoryfor desk files. Do not use the repo as a store.- One real seam per test is enough. You do not need a Docker network to check JSON on disk.
Try this
- Make
TicketHandlerreturn 404 unless the path is/ticket. Assert both status codes. - Add
load_ticketraisingFileNotFoundErrorwhen the JSON is missing. Test it withtmp_path. - Point
ticket_requestat the live server from Case 2 (urlopen(req)). Keep the shutdown infinally. - Change
save_ticketto reject an id that contains/. Test that it does not write outsideroot.