Het gesprek testen zonder poort

Je hebt je server in Run zien echoën, sluiten en falen, en Submit heeft elke functie gecontroleerd. Niets daarvan is bewijs dat je na een wijziging zelf opnieuw kunt draaien. Deze les schrijft drie tests in tests/test_loopback.py die je echte server op een echte socket starten en ermee praten, in een fractie van een seconde, zonder poort en zonder listener. Check my work draait later dezelfde tests tegen een eigen server.

Twee sockets, geen poort

Networks with Python hoofdstuk 4 testte een client met socket.socketpair():

client_end, server_end = socket.socketpair()

Het geeft twee sockets terug die al met elkaar verbonden zijn, binnen één programma. Bytes die in het ene uiteinde gaan, komen uit het andere, en als het ene uiteinde sluit, geeft recv op het andere b"" terug. Er is geen adres om aan te binden, geen poort om te kiezen, en niets anders op de machine kan een van beide uiteinden bereiken. Voor een WebSocket-server betekent dat dat een test server_end aan serve_connection kan geven precies zoals de bedrading er een geaccepteerde verbinding aan geeft, en de client kan spelen op client_end.

De twee hulpfuncties

Open tests/test_loopback.py. Boven de drie plaatshoudertests staan twee meegeleverde hulpfuncties.

fixed(n) geeft bytes(range(n)) terug, bekende bytes in plaats van willekeurige. Geef het mee als random_bytes=fixed en de handshakesleutel en elke maskeersleutel zijn voorspelbaar, zodat een falende test elke keer op dezelfde manier faalt.

start_server(server_end) draait loopback_demo.serve_connection(server_end) op een thread en geeft een dict terug met twee items: "thread", de thread, en "outcome", een dict waarin de thread het resultaat van serve_connection onder "result" opslaat als hij klaar is. Voordat het de thread start, zet het een time-out van 1 seconde op server_end, en het start de thread met daemon=True, zoals het project van hoofdstuk 2 deed, zodat een thread die nog draait de testrun nooit in leven kan houden. Het uiteinde van de server wordt gesloten door een with-blok binnen de thread.

Drie tests

Vervang de drie plaatshouders:

1. test_client_and_server_agree: draai loopback_demo.client_connection op client_end met twee of drie eigen teksten en random_bytes=fixed, en join daarna de serverthread. Assert dat het resultaat van de client je teksten als echo’s heeft en de code 1000, en dat het resultaat van de server je berichten telt met de code 1000.

2. test_server_answers_a_ping_with_the_same_payload: doe de handshake van de client zelf met ws_handshake.client_handshake, maak een read_exact met loopback_demo.make_read_exact(reader), stuur een gemaskeerde ping met een eigen payload (ws_frames.encode_frame met mask_key=fixed(4)), en lees het antwoord met ws_frames.read_message(read_exact, role="client"). Assert dat het een pong met je payload is. Beëindig daarna het gesprek: stuur een gemaskeerde close 1000 en lees de beantwoordende close van de server.

3. test_server_fails_an_unmasked_frame: stuur na de handshake ws_frames.encode_frame(ws_frames.OP_TEXT, "no mask"), dat geen masker heeft, lees het antwoord, en assert dat het een close is met de code 1002. Join de thread en assert dat het resultaat van de server ook de code 1002 heeft.

Elke test begint op dezelfde manier:

client_end, server_end = socket.socketpair()
client_end.settimeout(1.0)
server = start_server(server_end)

Daarna gaat de kant van de client binnen with client_end:, zodat het uiteinde van de client gesloten wordt, ook als een assert halverwege faalt, en na dat blok komt server["thread"].join(timeout=1.0). Assert het resultaat van de server na de join: de thread slaat het pas op als hij klaar is.

Vier regels komen voort uit hoe je tests worden gecontroleerd:

• Geef client_end een time-out van 1 seconde, en join elke thread met timeout=1.0.

• Beëindig elk gesprek: de client sluit, of leest de close van de server, in plaats van een time-out uit te zitten. De time-outs zijn een vangnet voor een kapotte server; een werkende server antwoordt meteen.

• Roep de functies van de server aan via de modules, als loopback_demo.client_connection(...) en ws_frames.read_message(...), binnen elke test. Check my work zet andere versies van de functies van de server in loopback_demo om te zien of je tests het merken; een naam die geïmporteerd is met from loopback_demo import zou het origineel houden.

• Assert codes, echo’s en payloads, niet de formulering van een reden die de server zelf schrijft, zoals de reden voor een gebroken regel, die een andere correcte server anders kan formuleren.

Dit zijn echte sockets en een echte thread. Je tests maken ze elke keer dat ze draaien, hier in de terminal en als Check my work ze draait.

Draai ze

Deze les heeft een terminal. Draai vanuit de projectmap:

python -m pytest -q

Met de plaatshouders nog op hun plek meldt de samenvattingsregel 3 failed. Als je tests geschreven zijn, meldt hij 3 passed. De demo draait ook nog steeds vanuit de terminal:

python loopback_demo.py

Druk daarna op Check my work. Het leest tests/test_loopback.py en draait je tests tot vijf keer in één proces: één keer tegen een eigen werkende server, waarbij minstens drie tests verzameld moeten worden en alle moeten slagen, en één keer elk tegen vier servers met elk één fout, waarbij minstens één test moet falen. De vier fouten zijn echo’s die gemaskeerd zijn, elke close beantwoord met 1002, een pong die de payload van de ping laat vallen, en een protocolfout beantwoord met 1000. Elke run die langer dan 2 seconden duurt, wordt gestopt, en daarom moet elke test zijn gesprek beëindigen.

Je server is nu bewezen door eigen tests. De laatste les van dit hoofdstuk maakt van loopback_demo.py een commando dat meerdere clients na elkaar bedient, binnen een deadline.