🏆Eindproject: Project: een kale fetch()

In vier lessen bouwde je fetch(method, target, *, connect) uit vier onderdelen: make_request schrijft het verzoek, read_head leest tot aan de lege regel, parse_head maakt van de kop een status en headers, en read_body leest precies de body, of geen na HEAD. Elk onderdeel werd los gecontroleerd, en elke live uitvoering liet één antwoord van practice-api zien. Dit project bewijst de hele functie, met tests die je zelf schrijft, tegen antwoorden die geknipt zijn op elke positie waar een netwerk ze zou kunnen knippen.

Wat fetch belooft

Als je klaar bent, bevat wire.py een fetch die dit allemaal doet:

• geeft een dictionary terug met "version", "status", "reason", "headers" en "body" voor een antwoord met elke status, 404 net zo goed als 200;

• werpt ValueError op voor een misvormde kop of lengte, en voor een body die het nog niet kan afmeten: een antwoord zonder Content-Length, of met Transfer-Encoding, dat hoofdstuk 9 afhandelt;

• werpt ConnectionError op als de server sluit voordat de kop of de body compleet is;

• leest nooit voorbij het einde van de body, en leest geen body na HEAD;

• laat de verbinding op elk pad gesloten achter, wat het with-blok voor je doet, ook als er daarbinnen een exception wordt opgeworpen.

Als je les 3 tot en met 6 hebt gevolgd, doet je fetch dit allemaal al. Vindt Check my work een gat, dan noemt het dat, en je herstelt het in wire.py. Het nieuwe werk in deze les zijn de tests.

De hulpmiddelen in fakes.py

Open fakes.py. Het is gewone Python, geschreven voor deze cursus, en het bevat twee functies die je tests gebruiken.

scripted_socket(chunks) geeft een vervangende socket terug. Elke recv(size) geeft het volgende stuk van het script, nooit meer dan size bytes tegelijk, en bewaart de rest van een lang stuk voor de volgende aanroep. Na het laatste stuk geeft recv b"" terug, zoals een echte socket doet zodra de server heeft gesloten. Een None in het script staat voor een server die stil is geworden maar de verbinding heeft opengehouden: daar werpt recv TimeoutError op, en dat is wat een echte socket met een time-out doet. Elke aanroep van sendall wordt vastgelegd in de lijst sock.sent.

fake_connect(sock, calls) geeft een functie terug die je als connect aan fetch meegeeft. Als fetch hem aanroept, voegt hij het adres en de time-out toe aan je lijst calls en geeft hij sock terug. Hij verpakt sock eerst in contextlib.nullcontext(sock), omdat fetch with connect(...) as sock: schrijft en de vervangende socket zelf geen with-blok kan beginnen: nullcontext geeft sock aan het blok en doet niets als het blok eindigt. Je prikbordtests in Networks with Python gebruikten dezelfde truc.

tests/test_wire.py begint al met import wire, importeert beide functies, definieert EXAMPLE_REPLY, een verzonnen antwoord in de vorm van practice-api met een JSON-body van 13 bytes, en bevat één test.

Mijlpalen

Schrijf de tests in deze volgorde. Voer na elke test python -m pytest -q uit.

1. Het hele antwoord in één stuk. Script EXAMPLE_REPLY als één stuk, roep wire.fetch("GET", "/v1/items/1", connect=fake_connect(sock, calls)) aan, en controleer met asserts wat er terugkwam: de connect-aanroep (("practice-api", 8080) en 2.0), de exacte bytes in sock.sent, de status, de header content-length en de body.

2. Elke knippositie. Een lus over elke positie i van 1 tot de lengte van het antwoord bouwt een script van twee stukken, [EXAMPLE_REPLY[:i], EXAMPLE_REPLY[i:]], en controleert elke keer de body. Dat dekt een knip binnen de statusregel, binnen elke headerregel, binnen de lege regel en binnen de body.

3. HEAD. Script alleen de kop van het antwoord, tot en met de lege regel, gevolgd door None, en roep fetch aan met "HEAD". Controleer dat de body b"" is en dat content-length er nog staat. De None doet ertoe: het is een server die de verbinding openhoudt, dus een fetch die op een body wacht, of op het sluiten door de server, werpt TimeoutError op en de test mislukt.

4. Elke schrijfwijze. Script een antwoord waarvan de lengteheader als cOnTeNt-LeNgTh is geschreven, en controleer dat de body terugkomt.

Hier is de vorm van mijlpaal 2, zonder de asserts:

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

Drie regels houden je tests controleerbaar. Schrijf elke test als een gewone test_-functie zonder parameters. Roep wire.fetch(...) binnen elke test via de module aan, en schrijf nooit from wire import fetch of fetch = wire.fetch: Check my work zet via de module andere versies van fetch in om te zien of je tests dat merken, en een naam die uit de module is gekopieerd, houdt het origineel vast. En laat een test nooit het netwerk bereiken: elke aanroep van fetch in je tests geeft connect= mee.

De tests uitvoeren

Vanuit het tabblad Shell:

python -m pytest -q

Elke test zou ruim binnen een seconde moeten slagen, omdat geen enkele test op een netwerk wacht. Blijft er een hangen, dan bereikt een lus in je test of in wire.py recv niet: een script kan niet wachten, omdat het eindigt op b"" of bij None een fout opwerpt.

Als je wilt, voer python wire.py dan nog eens uit tegen practice-api. Het is hetzelfde antwoord als in les 5, nu van een functie die je hebt getest tegen elke manier waarop het antwoord geknipt kon worden. Die live uitvoering is om naar te kijken; Check my work gebruikt hem niet, en je tests zijn sterker bewijs dan welke uitvoering ook, omdat ze knippen dekken die een netwerk misschien nooit toevallig maakt terwijl jij kijkt.

Druk daarna op Check my work. fetch kan precies één soort verzoek versturen: een methode en een doel die je intypt, met een vaste Host-regel en geen andere header. Hoofdstuk 2 bouwt verzoeken voor elke URL, met headers die je zelf kiest.