"""Tests for http_server.py. Plain asserts, no pytest: `python3 test_http_server.py`. Almost every test here is byte-level on the framing builders, which is the point: framing is a property of the bytes, not of the network. Only the last test opens a socket, and it uses an ephemeral port so it can never collide with the demo's fixed 8127. """ import contextlib import io import math import socket import sys import threading import http_server as hs BODY = hs.BODY class FakeConn: """A socket that hands back a prepared stream, at most bufsize at a time. Which is all `recv` ever promised. Nothing about a TCP segment is modelled here, because nothing about a TCP segment is visible to `recv`. """ def __init__(self, data): self.data = data def recv(self, bufsize): take, self.data = self.data[:bufsize], self.data[bufsize:] return take def head_of(raw): """Everything up to and including the blank line that ends the head.""" return raw.split(b"\r\n\r\n", 1)[0] + b"\r\n\r\n" def body_of(raw): return raw.split(b"\r\n\r\n", 1)[1] def test_body_is_27_bytes(): assert len(BODY) == 27, len(BODY) def test_head_ends_with_a_blank_line(): raw = hs.head("HTTP/1.1 200 OK", "A: 1", "B: 2") assert raw == b"HTTP/1.1 200 OK\r\nA: 1\r\nB: 2\r\n\r\n", raw def test_chunk_is_hex_length_then_bytes(): assert hs.chunk(b"abcdef0123456789") == b"10\r\nabcdef0123456789\r\n" def test_correct_declares_the_true_length(): raw = hs.frame_correct(BODY) assert b"Content-Length: 27\r\n" in head_of(raw) assert body_of(raw) == BODY def test_none_declares_no_framing_at_all(): head = head_of(hs.frame_none(BODY)) assert b"Content-Length" not in head assert b"Transfer-Encoding" not in head def test_short_promises_10_and_sends_27(): raw = hs.frame_short(BODY) assert b"Content-Length: 10\r\n" in head_of(raw) assert len(body_of(raw)) == 27 assert len(body_of(raw)) - 10 == 17 # curl reports excess = 17 def test_long_promises_47_and_sends_27(): raw = hs.frame_long(BODY) assert b"Content-Length: 47\r\n" in head_of(raw) assert len(body_of(raw)) == 27 # 20 bytes short of the promise def test_chunked_terminates_and_noterm_does_not(): good, bad = hs.frame_chunked(BODY), hs.frame_chunked_noterm(BODY) assert good.endswith(b"0\r\n\r\n") assert not bad.endswith(b"0\r\n\r\n") assert good == bad + b"0\r\n\r\n" # the ONLY difference assert body_of(good) == b"d\r\n" + BODY[:13] + b"\r\ne\r\n" + BODY[13:] \ + b"\r\n0\r\n\r\n" def test_close_header_only_still_has_no_length(): head = head_of(hs.frame_close_header_only(BODY)) assert b"Connection: close\r\n" in head assert b"Content-Length" not in head # the header is not the framing def test_http10_changes_the_version_and_nothing_else(): head = head_of(hs.frame_http10(BODY)) assert head.startswith(b"HTTP/1.0 200 OK\r\n") assert b"Content-Length" not in head assert head_of(hs.frame_none(BODY))[8:] == head[8:] # same but the version def test_both_orders_carry_the_same_two_headers(): for build in (hs.frame_both, hs.frame_both_te_first): head = head_of(build(BODY)) assert b"Content-Length: 99\r\n" in head assert b"Transfer-Encoding: chunked\r\n" in head a, b = hs.frame_both(BODY), hs.frame_both_te_first(BODY) assert a != b and body_of(a) == body_of(b) # order differs, bodies don't def test_every_framing_is_a_well_formed_head_plus_something(): assert len(hs.FRAMINGS) == 10 for name, (build, description) in hs.FRAMINGS.items(): raw = build(BODY) assert raw.startswith(b"HTTP/1."), name assert b"\r\n\r\n" in raw, name assert description, name def test_read_request_reassembles_a_head_split_across_recvs(): raw = b"GET /hello HTTP/1.1\r\nHost: x\r\nAccept: */*\r\n\r\n" line, headers, body, sizes = hs.read_request(FakeConn(raw), bufsize=8) assert line == "GET /hello HTTP/1.1" assert headers == {"host": "x", "accept": "*/*"} assert body == b"" assert len(raw) == 45 assert sizes == [8, 8, 8, 8, 8, 5] == [8] * 5 + [45 - 40] assert sum(sizes) == len(raw) and len(sizes) == math.ceil(45 / 8) def test_recv_count_is_the_buffer_bound_not_the_wire(): raw = b"POST /u HTTP/1.1\r\nContent-Length: 20000\r\n\r\n" + b"x" * 20000 for bufsize in (1024, 4096, 65536): _, _, body, sizes = hs.read_request(FakeConn(raw), bufsize) assert body == b"x" * 20000 assert len(sizes) == math.ceil(len(raw) / bufsize), (bufsize, sizes) def test_read_request_reports_no_request_on_immediate_eof(): line, headers, body, sizes = hs.read_request(FakeConn(b"")) assert line is None and headers == {} and body == b"" and sizes == [] def test_body_without_content_length_is_left_unread(): raw = b"POST /u HTTP/1.1\r\nHost: x\r\n\r\nhello" _, _, body, _ = hs.read_request(FakeConn(raw), bufsize=4096) assert body == b"hello" # already in the buffer... raw = b"POST /u HTTP/1.1\r\nHost: x\r\n\r\n" _, _, body, _ = hs.read_request(FakeConn(raw + b"hello"), bufsize=len(raw)) assert body == b"" # ...but never asked for def test_a_real_socket_round_trip(): probe = socket.socket() probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] probe.close() quiet = io.StringIO() with contextlib.redirect_stderr(quiet): thread = threading.Thread(target=hs.serve, args=(port, "correct", "fin"), daemon=True) thread.start() client = socket.socket() for _ in range(500): try: client.connect(("127.0.0.1", port)) break except OSError: client = socket.socket() client.sendall(b"GET /hello HTTP/1.1\r\nHost: x\r\n\r\n") got = b"" while True: data = client.recv(4096) if not data: break got += data client.close() assert got == hs.frame_correct(BODY), got assert b"listening on 127.0.0.1:%d" % port in quiet.getvalue().encode() def main(): tests = [value for name, value in sorted(globals().items()) if name.startswith("test_")] for test in tests: test() print("ok %s" % test.__name__) print("\n%d tests passed" % len(tests)) if __name__ == "__main__": main()