"""Pruebas del laboratorio. Corren sobre lab/ con `make test-lab`, y sobre solution/ dentro de `make test`, si esa carpeta existe.""" import os import re import socket import struct import subprocess import time import unittest from pathlib import Path from netlab import BUILD, HERE, NetTestCase, connect, free_port, recv_line, recv_until_close LAB_DIR = os.environ.get("LAB_DIR", "solution") BIN = BUILD / LAB_DIR TEXT, COUNTS, ERROR = 1, 2, 3 def run(name, *args, stdin=None, timeout=20): return subprocess.run([str(BIN / name), *map(str, args)], input=stdin, capture_output=True, timeout=timeout) def record(kind, value): return struct.pack("!BH", kind, len(value)) + value def read_exact(sock, n): data = b"" while len(data) < n: chunk = sock.recv(n - len(data)) if not chunk: break data += chunk return data def read_record(sock): header = read_exact(sock, 3) kind, length = struct.unpack("!BH", header) return kind, read_exact(sock, length) @unittest.skipUnless((HERE / LAB_DIR).is_dir(), f"no existe {LAB_DIR}/") class LabCase(NetTestCase): def lab_server(self, name, *args): return self.server(BIN / name, *args) class Step2BigClient(LabCase): def test_100_kb_arrive_and_recv_calls_are_counted(self): server = self.lab_server("wc_counting_server", 0) out = run("wc_big_client", "127.0.0.1", server.port) self.assertEqual(out.returncode, 0, out) self.assertIn("bytes=102400 ", out.stdout.decode()) calls = int(server.wait_output(r"recv: (\d+)").group(1)) # Con un buffer de 4096 bytes, como el de wc_server.c. self.assertGreaterEqual(calls, 25) def test_other_sizes(self): server = self.lab_server("wc_counting_server", 0) out = run("wc_big_client", "127.0.0.1", server.port, 5000) self.assertIn("bytes=5000 ", out.stdout.decode()) class Step3Records(LabCase): def test_one_request(self): server = self.lab_server("wcf_server", 0) sock = self.connect(server.port) sock.sendall(record(TEXT, b"uno dos tres\n")) kind, value = read_record(sock) self.assertEqual(kind, COUNTS) self.assertEqual(struct.unpack("!III", value), (13, 3, 1)) def test_record_split_byte_by_byte(self): server = self.lab_server("wcf_server", 0) sock = self.connect(server.port) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) for byte in record(TEXT, b"hola mundo"): sock.sendall(bytes([byte])) time.sleep(0.005) kind, value = read_record(sock) self.assertEqual(struct.unpack("!III", value), (10, 2, 0)) def test_records_glued_in_one_send(self): server = self.lab_server("wcf_server", 0) sock = self.connect(server.port) sock.sendall(record(TEXT, b"a") + record(TEXT, b"b c") + record(TEXT, b"")) expected = [(1, 1, 0), (3, 2, 0), (0, 0, 0)] for counts in expected: kind, value = read_record(sock) self.assertEqual(kind, COUNTS) self.assertEqual(struct.unpack("!III", value), counts) def test_large_record(self): server = self.lab_server("wcf_server", 0) sock = self.connect(server.port) sock.sendall(record(TEXT, b"x" * 65535)) _, value = read_record(sock) self.assertEqual(struct.unpack("!III", value), (65535, 1, 0)) def test_unknown_kind_gets_an_error_and_a_close(self): server = self.lab_server("wcf_server", 0) sock = self.connect(server.port) sock.sendall(record(99, b"??")) kind, _ = read_record(sock) self.assertEqual(kind, ERROR) self.assertEqual(recv_until_close(sock), b"") def test_client_that_leaves_mid_record_does_not_stop_the_server(self): server = self.lab_server("wcf_server", 0) gone = connect(server.port) gone.sendall(struct.pack("!BH", TEXT, 100) + b"solo un poco") gone.close() sock = self.connect(server.port) sock.sendall(record(TEXT, b"sigue")) kind, _ = read_record(sock) self.assertEqual(kind, COUNTS) def test_client_prints_one_answer_per_line(self): server = self.lab_server("wcf_server", 0) out = run("wcf_client", "127.0.0.1", server.port, stdin=b"uno dos\ntres\n") self.assertEqual(out.returncode, 0, out) self.assertEqual(out.stdout.decode().splitlines(), ["bytes=8 words=2 lines=1", "bytes=5 words=1 lines=1"]) class Step4Udp(LabCase): def test_a_datagram_is_a_message(self): server = self.lab_server("wc_udp_server", 0) for flags in ([], ["-c"]): out = run("wc_udp_client", *flags, "127.0.0.1", server.port, stdin=b"uno dos\n") self.assertEqual(out.returncode, 0, out) self.assertIn("bytes=8 words=2 lines=1", out.stdout.decode()) def test_server_off(self): port = free_port(socket.SOCK_DGRAM) out = run("wc_udp_client", "127.0.0.1", port, stdin=b"hola\n") self.assertEqual(out.returncode, 2, out) self.assertIn("sin respuesta", out.stdout.decode()) def test_server_off_with_connect(self): port = free_port(socket.SOCK_DGRAM) out = run("wc_udp_client", "-c", "127.0.0.1", port, stdin=b"hola\n") self.assertEqual(out.returncode, 1, out) self.assertIn(b"Connection refused", out.stderr) class Step5Room(LabCase): def room(self): server = self.lab_server("room_server", 0) a = self.connect(server.port) b = self.connect(server.port) time.sleep(0.2) return server, a, b def test_line_reaches_the_other_client_and_not_the_sender(self): _, a, b = self.room() a.sendall(b"hola\n") self.assertRegex(recv_line(b), r"^\[\d+\] hola\n$") a.settimeout(0.3) with self.assertRaises(TimeoutError): a.recv(100) def test_both_clients_can_talk_in_any_order(self): _, a, b = self.room() b.sendall(b"uno\n") self.assertIn("uno", recv_line(a)) a.sendall(b"dos\n") self.assertIn("dos", recv_line(b)) b.sendall(b"tres\n") self.assertIn("tres", recv_line(a)) def test_partial_line_waits_for_its_end(self): _, a, b = self.room() a.sendall(b"ho") b.settimeout(0.3) with self.assertRaises(TimeoutError): b.recv(100) b.settimeout(5) a.sendall(b"la\n") self.assertRegex(recv_line(b), r"hola\n$") def test_two_lines_in_one_send(self): _, a, b = self.room() a.sendall(b"uno\ndos\n") self.assertIn("uno", recv_line(b)) self.assertIn("dos", recv_line(b)) def test_room_survives_a_client_leaving(self): server, a, b = self.room() c = connect(server.port) time.sleep(0.1) c.sendall(b"medio") c.close() time.sleep(0.1) a.sendall(b"sigue\n") self.assertIn("sigue", recv_line(b)) if __name__ == "__main__": unittest.main()