🏆Capstone project: Project: A Raw fetch()

Over four lessons you built fetch(method, target, *, connect) from four pieces: make_request writes the request, read_head reads up to the empty line, parse_head turns the head into a status and headers, and read_body reads exactly the body, or none after HEAD. Each piece was checked on its own, and each live run showed one reply from practice-api. This project proves the whole function, with tests you write, against replies cut at every position a network could cut them.

What fetch promises

When you are done, wire.py holds a fetch that does all of this:

• returns a dictionary with "version", "status", "reason", "headers" and "body" for a reply with any status, 404 as readily as 200;

• raises ValueError for a malformed head or length, and for a body it cannot measure yet: a reply with no Content-Length, or with Transfer-Encoding, which Chapter 9 handles;

• raises ConnectionError when the server closes before the head or the body is complete;

• never reads past the end of the body, and reads no body after HEAD;

• leaves the connection closed on every path, which the with block does for you, also when an exception is raised inside it.

If you followed Lessons 3 to 6, your fetch already does all of this. If Check my work finds a gap, it names it, and you fix it in wire.py. The new work in this lesson is the tests.

The tools in fakes.py

Open fakes.py. It is ordinary Python, written for this course, and it holds two functions your tests use.

scripted_socket(chunks) returns a stand-in socket. Each recv(size) hands out the next chunk of the script, never more than size bytes at a time, keeping the rest of a long chunk for the next call. After the last chunk, recv returns b"", as a real socket does once the server has closed. A None in the script stands for a server that has gone silent but kept the connection open: recv raises TimeoutError there, which is what a real socket with a timeout does. Every sendall call is recorded in the list sock.sent.

fake_connect(sock, calls) returns a function you pass to fetch as connect. When fetch calls it, it appends the address and timeout to your calls list and hands back sock. It wraps sock in contextlib.nullcontext(sock) first, because fetch writes with connect(...) as sock: and the stand-in socket cannot start a with block on its own: nullcontext hands sock to the block and does nothing when the block ends. Your notice board tests in Networks with Python used the same trick.

tests/test_wire.py already starts with import wire, imports both functions, defines EXAMPLE_REPLY, a made-up reply in practice-api’s shape with a 13-byte JSON body, and holds one test.

Milestones

Write the tests in this order. Run python -m pytest -q after each one.

1. The whole reply in one chunk. Script EXAMPLE_REPLY as one chunk, call wire.fetch("GET", "/v1/items/1", connect=fake_connect(sock, calls)), and assert what came back: the connect call (("practice-api", 8080) and 2.0), the exact bytes in sock.sent, the status, the content-length header, and the body.

2. Every split position. A loop over every position i from 1 to the length of the reply builds a two-chunk script, [EXAMPLE_REPLY[:i], EXAMPLE_REPLY[i:]], and checks the body each time. That covers a cut inside the status line, inside every header line, inside the empty line, and inside the body.

3. HEAD. Script only the head of the reply, up to and including the empty line, followed by None, and call fetch with "HEAD". Assert that the body is b"" and that content-length is still there. The None matters: it is a server that keeps the connection open, so a fetch that waits for a body, or for the server to close, raises TimeoutError and the test fails.

4. Any casing. Script a reply whose length header is written cOnTeNt-LeNgTh, and assert that its body comes back.

Here is the shape of milestone 2, without its assertions:

def test_fetch_reads_a_reply_split_anywhere():
    """The reply may arrive in two pieces cut at any byte."""
    for i in range(1, len(EXAMPLE_REPLY)):
        sock = scripted_socket([EXAMPLE_REPLY[:i], EXAMPLE_REPLY[i:]])
        response = wire.fetch("GET", "/v1/items/1", connect=fake_connect(sock, []))

Three rules keep your tests checkable. Write each test as a plain test_ function with no parameters. Call wire.fetch(...) through the module inside each test, and never write from wire import fetch or fetch = wire.fetch: Check my work swaps in other versions of fetch through the module to see whether your tests notice, and a name copied out of the module keeps the original. And never let a test reach the network: every fetch call in your tests passes connect=.

Run the tests

From the Shell tab:

python -m pytest -q

Every test should pass in well under a second, because no test waits for a network. If one hangs, a loop in your test or in wire.py is not reaching recv: a script cannot wait, because it ends in b"" or raises at None.

If you like, run python wire.py once more against practice-api. It is the same reply as in Lesson 5, now from a function you have tested against every way the reply could be cut. That live run is for looking at; Check my work does not use it, and your tests are stronger evidence than any one run, because they cover cuts a network may never happen to make while you watch.

Then press Check my work. fetch can send exactly one kind of request: a method and a target you type, with a fixed Host line and no other header. Chapter 2 builds requests for any URL, with headers of your choosing.