Test the Conversation Without a Port

You have seen your server echo, close and fail in Run, and Submit has checked each function. None of that is evidence you can rerun yourself after a change. This lesson writes three tests in tests/test_loopback.py that start your real server on a real socket and talk to it, in a fraction of a second, with no port and no listener. Check my work later runs the same tests against a server of its own.

Two sockets, no port

Networks with Python Chapter 4 tested a client with socket.socketpair():

client_end, server_end = socket.socketpair()

It returns two sockets that are already connected to each other, inside one program. Bytes sent into one end come out of the other, and when one end closes, recv on the other returns b"". There is no address to bind, no port to pick, and nothing else on the machine can reach either end. For a WebSocket server, that means a test can hand server_end to serve_connection exactly as the wiring hands it an accepted connection, and play the client on client_end.

The two helpers

Open tests/test_loopback.py. Above the three placeholder tests are two supplied helpers.

fixed(n) returns bytes(range(n)), known bytes in place of random ones. Pass it as random_bytes=fixed and the handshake key and every mask key are predictable, so a failing test fails the same way every time.

start_server(server_end) runs loopback_demo.serve_connection(server_end) on a thread and returns a dict with two entries: "thread", the thread, and "outcome", a dict in which the thread stores serve_connection’s result under "result" when it finishes. Before starting the thread it sets a 1-second timeout on server_end, and it starts the thread with daemon=True, as Chapter 2’s project did, so a thread that is still running can never keep the test run alive. The server’s end is closed by a with block inside the thread.

Three tests

Replace the three placeholders:

1. test_client_and_server_agree: run loopback_demo.client_connection on client_end with two or three texts of your own and random_bytes=fixed, then join the server thread. Assert that the client’s result has your texts as its echoes and the code 1000, and that the server’s result counts your messages with the code 1000.

2. test_server_answers_a_ping_with_the_same_payload: do the client’s handshake yourself with ws_handshake.client_handshake, make a read_exact with loopback_demo.make_read_exact(reader), send a masked ping with a payload of your own (ws_frames.encode_frame with mask_key=fixed(4)), and read the answer with ws_frames.read_message(read_exact, role="client"). Assert it is a pong with your payload. Then end the conversation: send a masked close 1000 and read the server’s answering close.

3. test_server_fails_an_unmasked_frame: after the handshake, send ws_frames.encode_frame(ws_frames.OP_TEXT, "no mask"), which has no mask, read the answer, and assert that it is a close with the code 1002. Join the thread and assert that the server’s result has the code 1002 as well.

Every test starts the same way:

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

Then the client’s side goes inside with client_end:, so the client’s end is closed even if an assertion fails partway, and after that block comes server["thread"].join(timeout=1.0). Assert the server’s result after the join: the thread stores it only when it finishes.

Four rules come from how your tests are checked:

• Give client_end a 1-second timeout, and join every thread with timeout=1.0.

• End each conversation: the client closes, or reads the server’s close, instead of waiting out a timeout. The timeouts are a safety net for a broken server; a working server answers at once.

• Call the server’s functions through the modules, as loopback_demo.client_connection(...) and ws_frames.read_message(...), inside each test. Check my work swaps other versions of the server’s functions into loopback_demo to see whether your tests notice; a name imported with from loopback_demo import would keep the original.

• Assert codes, echoes and payloads, not the wording of a reason the server writes itself, such as the reason for a broken rule, which another correct server may word differently.

These are real sockets and a real thread. Your tests create them every time they run, here in the terminal and when Check my work runs them.

Run them

This lesson has a terminal. From the project folder, run:

python -m pytest -q

With the placeholders still in place, the summary line reports 3 failed. When your tests are written, it reports 3 passed. The demo still runs from the terminal too:

python loopback_demo.py

Then press Check my work. It reads tests/test_loopback.py and runs your tests up to five times in one process: once against a working server of its own, where at least three tests must be collected and all must pass, and once each against four servers with one mistake each, where at least one test must fail. The four mistakes are echoes that are masked, every close answered with 1002, a pong that drops the ping’s payload, and a protocol error answered with 1000. Any run that takes longer than 2 seconds is stopped, which is why each test must end its conversation.

Your server is now proved by tests of your own. The last lesson of this chapter turns loopback_demo.py into a command that serves several clients one after another, within a deadline.