0%

Ein Server, viele Clients · Übung

Mit Fake-Verbindungen testen

Beitritt, Register und Broadcast funktionieren, und bisher haben das nur die Demo und Submit gezeigt. In dieser Lektion schreibst du eigene Tests in tests/test_scoreboard.py und führst sie im aus. Der Nachrichtentest bleibt für den Rest des Kurses; die drei Servertests beschreiben Version eins der Anzeigetafel, und Kapitel 8 ersetzt sie. Keiner von ihnen öffnet einen oder startet einen Thread: Jeder übergibt dem Server eine Stellvertreter-Verbindung und prüft, was der Stellvertreter aufgezeichnet hat.

Ein Stellvertreter, aus der Demo abgelesen

Öffne scoreboard_demo.py und finde demo_connection. Die baut einen Stellvertreter aus drei Funktionen, die in ihr definiert sind, recv, send und close, und die sich ein namens state teilen. send parst jede Nachricht und hängt sie an state["sent"] an, oder löst ConnectionClosedError aus, wenn der Stellvertreter mit fail_send=True erstellt wurde. Die drei Funktionen werden dann mit einer id und einem request in einem types.SimpleNamespace gesammelt, einem , dessen Attribute das sind, was du ihm übergibst. Der Server ruft darauf connection.send(...) und connection.id genauso auf wie auf einer Verbindung der Bibliothek und merkt den Unterschied nie. Python in der Praxis hat Stellvertreter-Sitzungen auf dieselbe Weise gebaut.

Die Tests brauchen weniger als das. add_viewer, apply_score und broadcast verwenden immer nur id und send, also kann make_connection(conn_id, *, fail_send=False) klein sein:

def make_connection(conn_id, *, fail_send=False):
    sent = []

    def send(text):
        if fail_send:
            raise closed_error()
        sent.append(json.loads(text))

    return types.SimpleNamespace(id=conn_id, send=send, sent=sent)

closed_error() ist oben in der Datei vorgegeben: Es baut den ConnectionClosedError, den ein send auslöst, wenn die Verbindung des Zuschauers ohne Close- geendet hat. Wenn sent am Stellvertreter hängt, kann ein Test danach viewer.sent lesen.

Vier Tests

Jeder Test folgt denselben drei Schritten: eine Tafel und einige Stellvertreter vorbereiten, über einen Aufruf an den Server handeln und zusichern, was die Stellvertreter aufgezeichnet haben und wer in board["viewers"] steht.

  • test_a_viewer_gets_the_board_when_it_joins: Erstelle eine Tafel, ändere ihre Version und die Punkte eines Teams, rufe add_viewer auf und sichere zu, dass der Stellvertreter genau diesen Snapshot aufgezeichnet hat und dass seine ID im Register steht.

  • test_a_score_reaches_every_viewer: Füge zwei Zuschauer hinzu, wende eine Wertung an und sichere zu, dass beide die neue Version und die neuen Punkte empfangen haben.

  • test_a_failed_viewer_is_dropped_and_the_rest_still_hear: Registriere einen mit fail_send=True erstellten Zuschauer vor einem funktionierenden, wende eine Wertung an und sichere zu, dass der gescheiterte das Register verlassen hat und dass der funktionierende das Update trotzdem empfangen hat. Die Reihenfolge ist wichtig: Ein gescheiterter Zuschauer, der als letzter registriert wurde, könnte nicht zeigen, ob die nach ihm weitergelaufen ist.

  • test_parse_message_refuses: Behalte die Zeile @pytest.mark.parametrize, die du in Kapitel 3 kennengelernt hast, und gib ihr mindestens drei Texte, die parse_message ablehnen muss. Der Fall mit true als Punkte steht schon da. Rufe darin parse_message in einem try auf; kehre aus except ValueError: zurück und schreibe nach der try-Anweisung assert False, ... mit dem Text, damit ein Text, der akzeptiert wird, den Test scheitern lässt.

Um einen Stellvertreter ohne Beitritts-Snapshot ins Register zu setzen, weise ihn direkt zu: board["viewers"]["broken"] = broken.

Welche Version des Servers soll test_a_failed_viewer_is_dropped_and_the_rest_still_hear erwischen?

Warum über das Modul

Die Datei importiert die , import scoreboard_server und import scoreboard_messages, und die Tests rufen scoreboard_server.apply_score(...) auf. Lass das so. Check my work führt deine Tests gegen Versionen des Servers aus, die je einen Fehler enthalten, und tauscht dabei die Funktion im Modul aus, während deine Tests laufen, um zu sehen, ob deine Tests es bemerken. Ein Aufruf über das Modul schlägt die Funktion im Moment des Aufrufs nach, bekommt also die ausgetauschte. Ein mit from scoreboard_server import apply_score importierter Name wurde einmal gebunden, als die Datei importiert wurde, und behält die alte Funktion. Aus demselben Grund machst du jeden Aufruf innerhalb einer Testfunktion, nicht oben in der Datei.

Warum rufen die Tests scoreboard_server.apply_score(...) auf, statt apply_score mit from zu importieren?

Führe sie aus

Diese Lektion verwendet das Terminal. Führe die Tests im Projektordner aus:

python -m pytest -q

Jeder Platzhalter scheitert, bis du ihn ersetzt. Auf der Seite hat die fertige Datei 8 passed gemeldet: drei Tests und fünf Texte im parametrisierten Test, von denen jeder als eigener Test zählt. Die Demo läuft weiter wie bisher, und ihre Ausgabe hat sich seit Lektion 4 nicht geändert:

python scoreboard_demo.py

Diese Tests beweisen, was ein Thread sieht: den Beitritts-Snapshot, den Broadcast und das Entfernen eines verschwundenen Zuschauers. Sie beweisen nicht den Wettlauf an der Tür aus Lektion 3, der zwei Threads im genau falschen Moment braucht. Den hat ein eigenes Instrument der Prüfung bewiesen, als du in Lektion 3 auf Submit geklickt hast; deine Tests müssen das nicht.

Wenn jeder Test besteht, klicke auf Check my work. Die nächste Lektion schreibt den Handler des Punktezählers, das letzte Stück von Anzeigetafel v1, und führt ihn mit mehreren Zuschauern auf aus.

Aufgabe

Schreibe in tests/test_scoreboard.py make_connection und ersetze die vier Platzhalter durch Tests. Lass closed_error und die Importe, wie sie sind.

  • make_connection(conn_id, *, fail_send=False): Gib einen types.SimpleNamespace zurück, mit id, einem send, das json.loads(text) an eine anhängt (oder closed_error() auslöst, wenn fail_send wahr ist), und dieser Liste als sent.

  • test_a_viewer_gets_the_board_when_it_joins: Nach scoreboard_server.add_viewer hat der Stellvertreter den Snapshot der Tafel aufgezeichnet, mit ihrer Version und ihren Punkteständen, und seine ID steht in board["viewers"].

  • test_a_score_reaches_every_viewer: zwei Zuschauer, ein scoreboard_server.apply_score, und beide haben den neuen Snapshot empfangen.

  • test_a_failed_viewer_is_dropped_and_the_rest_still_hear: ein scheiternder Zuschauer, registriert vor einem funktionierenden; nach einer Wertung ist der scheiternde aus board["viewers"] verschwunden, und der funktionierende hat das Update empfangen.

  • test_parse_message_refuses: parametrisiert mit mindestens drei Texten, darunter der Fall mit true als Punkte, von denen scoreboard_messages.parse_message jeden mit ablehnen muss.

Rufe in den Tests jede Serverfunktion über scoreboard_server. und scoreboard_messages. auf, niemals mit from ... import. Verwende nur Stellvertreter: keinen , keinen Thread, kein Warten.

Führe python -m pytest -q im aus, bis jeder Test besteht, und klicke dann auf Check my work. Die Prüfung liest deine Testdatei und führt deine Tests dann in diesem Prozess aus: einmal mit einer eigenen funktionierenden Anzeigetafel, bei der mindestens vier Tests gesammelt werden müssen (einer parametrisiert mit mindestens drei Texten) und alle bestehen müssen, und je einmal mit vier Anzeigetafeln, die je einen Fehler enthalten: ein add_viewer, das registriert, ohne den Snapshot zu senden, ein broadcast, das beim ersten verschwundenen Zuschauer aufhört, ein broadcast, das diesen Zuschauer behält, und ein is_whole_number, das True akzeptiert. Gegen jede davon muss mindestens ein Test scheitern. Ein Lauf gegen die funktionierende Anzeigetafel, der länger als 1 Sekunde dauert, wird gestoppt. Die Prüfung liest dein Terminal nicht.