Das Gespräch ohne Port testen

Du hast in Run gesehen, wie dein Server Echos sendet, schließt und scheitern lässt, und Submit hat jede Funktion geprüft. Nichts davon ist ein Beleg, den du nach einer Änderung selbst erneut ausführen kannst. Diese Lektion schreibt drei Tests in tests/test_loopback.py, die deinen echten Server auf einem echten Socket starten und mit ihm reden, in einem Bruchteil einer Sekunde, ohne Port und ohne Listener. Check my work führt dieselben Tests später gegen einen eigenen Server aus.

Zwei Sockets, kein Port

Networks with Python, Kapitel 4, hat einen Client mit socket.socketpair() getestet:

client_end, server_end = socket.socketpair()

Es gibt zwei Sockets zurück, die bereits miteinander verbunden sind, in einem Programm. Bytes, die in ein Ende gesendet werden, kommen aus dem anderen heraus, und wenn ein Ende schließt, gibt recv auf dem anderen b"" zurück. Es gibt keine Adresse zum Binden, keinen Port zum Wählen, und nichts sonst auf dem Rechner kann eines der beiden Enden erreichen. Für einen WebSocket-Server bedeutet das, dass ein Test server_end an serve_connection übergeben kann, genau wie die Verdrahtung ihm eine angenommene Verbindung übergibt, und auf client_end den Client spielen kann.

Die beiden Hilfsfunktionen

Öffne tests/test_loopback.py. Über den drei Platzhalter-Tests stehen zwei mitgelieferte Hilfsfunktionen.

fixed(n) gibt bytes(range(n)) zurück, bekannte Bytes statt zufälliger. Übergib sie als random_bytes=fixed, und der Handshake-Schlüssel und jeder Maskierungsschlüssel sind vorhersagbar, sodass ein scheiternder Test jedes Mal auf dieselbe Weise scheitert.

start_server(server_end) führt loopback_demo.serve_connection(server_end) in einem Thread aus und gibt ein Dict mit zwei Einträgen zurück: "thread", den Thread, und "outcome", ein Dict, in dem der Thread das Ergebnis von serve_connection unter "result" speichert, wenn er fertig ist. Bevor sie den Thread startet, setzt sie auf server_end ein Timeout von 1 Sekunde, und sie startet den Thread mit daemon=True, wie es das Projekt aus Kapitel 2 getan hat, sodass ein noch laufender Thread den Testlauf nie am Leben halten kann. Das Ende des Servers wird von einem with-Block im Thread geschlossen.

Drei Tests

Ersetze die drei Platzhalter:

1. test_client_and_server_agree: Führe loopback_demo.client_connection auf client_end mit zwei oder drei eigenen Texten und random_bytes=fixed aus und führe dann den join des Server-Threads aus. Sichere zu, dass das Ergebnis des Clients deine Texte als Echos und den Code 1000 hat und dass das Ergebnis des Servers deine Nachrichten mit dem Code 1000 zählt.

2. test_server_answers_a_ping_with_the_same_payload: Führe den Handshake des Clients selbst mit ws_handshake.client_handshake aus, erzeuge ein read_exact mit loopback_demo.make_read_exact(reader), sende einen maskierten Ping mit einer eigenen Payload (ws_frames.encode_frame mit mask_key=fixed(4)) und lies die Antwort mit ws_frames.read_message(read_exact, role="client"). Sichere zu, dass es ein Pong mit deiner Payload ist. Beende dann das Gespräch: Sende einen maskierten Close 1000 und lies den antwortenden Close des Servers.

3. test_server_fails_an_unmasked_frame: Sende nach dem Handshake ws_frames.encode_frame(ws_frames.OP_TEXT, "no mask"), das keine Maske hat, lies die Antwort und sichere zu, dass sie ein Close mit dem Code 1002 ist. Führe den join des Threads aus und sichere zu, dass auch das Ergebnis des Servers den Code 1002 hat.

Jeder Test beginnt auf dieselbe Weise:

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

Dann kommt die Seite des Clients in with client_end:, damit das Ende des Clients auch dann geschlossen wird, wenn eine Zusicherung mittendrin scheitert, und nach diesem Block kommt server["thread"].join(timeout=1.0). Sichere das Ergebnis des Servers nach dem join zu: Der Thread speichert es erst, wenn er fertig ist.

Vier Regeln ergeben sich daraus, wie deine Tests geprüft werden:

• Gib client_end ein Timeout von 1 Sekunde und führe den join jedes Threads mit timeout=1.0 aus.

• Beende jedes Gespräch: Der Client schließt oder liest den Close des Servers, statt ein Timeout abzuwarten. Die Timeouts sind ein Sicherheitsnetz für einen kaputten Server; ein funktionierender Server antwortet sofort.

• Rufe die Funktionen des Servers in jedem Test über die Module auf, als loopback_demo.client_connection(...) und ws_frames.read_message(...). Check my work tauscht andere Versionen der Funktionen des Servers in loopback_demo ein, um zu sehen, ob deine Tests es bemerken; ein Name, der mit from loopback_demo import importiert wurde, würde das Original behalten.

• Sichere Codes, Echos und Payloads zu, nicht den Wortlaut eines Grundes, den der Server selbst schreibt, etwa den Grund für eine verletzte Regel, den ein anderer korrekter Server anders formulieren kann.

Das sind echte Sockets und ein echter Thread. Deine Tests erzeugen sie bei jedem Lauf, hier im Terminal und wenn Check my work sie ausführt.

Sie ausführen

Diese Lektion hat ein Terminal. Führe aus dem Projektordner aus:

python -m pytest -q

Solange die Platzhalter noch da sind, meldet die Zusammenfassungszeile 3 failed. Wenn deine Tests geschrieben sind, meldet sie 3 passed. Die Demo läuft auch weiterhin aus dem Terminal:

python loopback_demo.py

Klicke dann auf Check my work. Es liest tests/test_loopback.py und führt deine Tests bis zu fünfmal in einem Prozess aus: einmal gegen einen eigenen funktionierenden Server, bei dem mindestens drei Tests gesammelt werden und alle bestehen müssen, und je einmal gegen vier Server mit je einem Fehler, bei denen mindestens ein Test scheitern muss. Die vier Fehler sind maskierte Echos, jeder Close mit 1002 beantwortet, ein Pong, der die Payload des Pings weglässt, und ein Protokollfehler, der mit 1000 beantwortet wird. Jeder Lauf, der länger als 2 Sekunden dauert, wird gestoppt, und deshalb muss jeder Test sein Gespräch beenden.

Dein Server ist jetzt durch eigene Tests belegt. Die letzte Lektion dieses Kapitels macht aus loopback_demo.py einen Befehl, der mehrere Clients nacheinander bedient, innerhalb einer Frist.