🏆Abschlussprojekt: Projekt: ein rohes fetch()

In vier Lektionen hast du fetch(method, target, *, connect) aus vier Teilen gebaut: make_request schreibt die Anfrage, read_head liest bis zur Leerzeile, parse_head macht aus dem Kopf einen Status und Header, und read_body liest genau den Body, oder nach HEAD keinen. Jedes Teil wurde für sich geprüft, und jeder Live-Lauf hat eine Antwort von practice-api gezeigt. Dieses Projekt prüft die ganze Funktion, mit Tests, die du schreibst, gegen Antworten, die an jeder Position geschnitten sind, an der ein Netzwerk sie schneiden könnte.

Was fetch verspricht

Wenn du fertig bist, enthält wire.py ein fetch, das all das tut:

• Es gibt für eine Antwort mit beliebigem Status ein Dictionary mit "version", "status", "reason", "headers" und "body" zurück, für 404 genauso bereitwillig wie für 200;

• es löst ValueError aus für einen fehlerhaften Kopf oder eine fehlerhafte Länge und für einen Body, den es noch nicht abmessen kann: eine Antwort ohne Content-Length oder mit Transfer-Encoding, die Kapitel 9 behandelt;

• es löst ConnectionError aus, wenn der Server schließt, bevor der Kopf oder der Body vollständig ist;

• es liest nie über das Ende des Bodys hinaus und liest nach HEAD keinen Body;

• es hinterlässt die Verbindung auf jedem Weg geschlossen, was der with-Block für dich erledigt, auch wenn darin eine Ausnahme ausgelöst wird.

Wenn du die Lektionen 3 bis 6 durchgearbeitet hast, tut dein fetch all das schon. Wenn Check my work eine Lücke findet, nennt es sie, und du behebst sie in wire.py. Die neue Arbeit in dieser Lektion sind die Tests.

Die Werkzeuge in fakes.py

Öffne fakes.py. Es ist gewöhnliches Python, für diesen Kurs geschrieben, und enthält zwei Funktionen, die deine Tests benutzen.

scripted_socket(chunks) gibt einen Ersatz-Socket zurück. Jedes recv(size) gibt das nächste Stück des Skripts heraus, nie mehr als size Bytes auf einmal, und hebt den Rest eines langen Stücks für den nächsten Aufruf auf. Nach dem letzten Stück gibt recv b"" zurück, wie ein echter Socket, sobald der Server geschlossen hat. Ein None im Skript steht für einen Server, der verstummt ist, die Verbindung aber offen gehalten hat: recv löst dort TimeoutError aus, und genau das tut ein echter Socket mit einem Timeout. Jeder Aufruf von sendall wird in der Liste sock.sent aufgezeichnet.

fake_connect(sock, calls) gibt eine Funktion zurück, die du fetch als connect übergibst. Wenn fetch sie aufruft, hängt sie die Adresse und das Timeout an deine Liste calls an und gibt sock zurück. Vorher verpackt sie sock in contextlib.nullcontext(sock), weil fetch with connect(...) as sock: schreibt und der Ersatz-Socket allein keinen with-Block beginnen kann: nullcontext übergibt sock an den Block und tut nichts, wenn der Block endet. Deine Pinnwand-Tests in Networks with Python haben denselben Trick benutzt.

tests/test_wire.py beginnt schon mit import wire, importiert beide Funktionen, definiert EXAMPLE_REPLY, eine erfundene Antwort in der Form von practice-api mit einem JSON-Body von 13 Bytes, und enthält einen Test.

Meilensteine

Schreibe die Tests in dieser Reihenfolge. Führe nach jedem python -m pytest -q aus.

1. Die ganze Antwort in einem Stück. Skripte EXAMPLE_REPLY als ein Stück, rufe wire.fetch("GET", "/v1/items/1", connect=fake_connect(sock, calls)) auf und prüfe mit assert, was zurückkam: den Aufruf von connect (("practice-api", 8080) und 2.0), die genauen Bytes in sock.sent, den Status, den Header content-length und den Body.

2. Jede Schnittposition. Eine Schleife über jede Position i von 1 bis zur Länge der Antwort baut ein Skript aus zwei Stücken, [EXAMPLE_REPLY[:i], EXAMPLE_REPLY[i:]], und prüft jedes Mal den Body. Das deckt einen Schnitt innerhalb der Statuszeile ab, innerhalb jeder Headerzeile, innerhalb der Leerzeile und innerhalb des Bodys.

3. HEAD. Skripte nur den Kopf der Antwort, bis einschließlich der Leerzeile, gefolgt von None, und rufe fetch mit "HEAD" auf. Prüfe, dass der Body b"" ist und dass content-length noch da ist. Das None ist wichtig: Es ist ein Server, der die Verbindung offen hält, also löst ein fetch, das auf einen Body wartet oder darauf, dass der Server schließt, TimeoutError aus, und der Test schlägt fehl.

4. Beliebige Schreibweise. Skripte eine Antwort, deren Längen-Header cOnTeNt-LeNgTh geschrieben ist, und prüfe, dass ihr Body zurückkommt.

Hier ist die Form von Meilenstein 2, ohne seine Zusicherungen:

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, []))

Drei Regeln halten deine Tests prüfbar. Schreibe jeden Test als einfache test_-Funktion ohne Parameter. Rufe in jedem Test wire.fetch(...) über das Modul auf, und schreibe nie from wire import fetch oder fetch = wire.fetch: Check my work setzt über das Modul andere Versionen von fetch ein, um zu sehen, ob deine Tests es bemerken, und ein Name, der aus dem Modul herauskopiert wurde, behält das Original. Und lass nie einen Test das Netzwerk erreichen: Jeder Aufruf von fetch in deinen Tests übergibt connect=.

Die Tests ausführen

Im Tab Shell:

python -m pytest -q

Jeder Test sollte in deutlich unter einer Sekunde bestehen, weil kein Test auf ein Netzwerk wartet. Wenn einer hängt, erreicht eine Schleife in deinem Test oder in wire.py das recv nicht: Ein Skript kann nicht warten, weil es mit b"" endet oder bei None eine Ausnahme auslöst.

Wenn du möchtest, führe python wire.py noch einmal gegen practice-api aus. Es ist dieselbe Antwort wie in Lektion 5, jetzt von einer Funktion, die du gegen jede Art getestet hast, auf die die Antwort geschnitten werden könnte. Dieser Live-Lauf ist zum Anschauen da; Check my work benutzt ihn nicht, und deine Tests sind ein stärkerer Beleg als jeder einzelne Lauf, weil sie Schnitte abdecken, die ein Netzwerk vielleicht nie zufällig macht, während du zusiehst.

Drücke dann Check my work. fetch kann genau eine Art von Anfrage senden: eine Methode und ein Ziel, die du eintippst, mit einer festen Zeile Host und keinem anderen Header. Kapitel 2 baut Anfragen für jede URL, mit Headern deiner Wahl.