"""Pruebas sobre la red de la Sesión A: demultiplexación, UDP y un stop-and-wait real.""" import re import time import unittest import udp_checksum from netlab import HERE, Lab, run LAB = Lab("par", "w09-par", "redes-w09-par") H1, H2, SRV = "10.209.21.10", "10.209.22.10", "10.209.23.10" PORT = "9009" FOLDER = "/tmp/tftp" def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.deploy() # Un archivo chico y otro más grande, aleatorios, para la transferencia. LAB.out("srv", "sh", "-c", f"mkdir -p {FOLDER} && head -c 25600 /dev/urandom > {FOLDER}/25k") def link(*args): done = run(["sh", f"{HERE}/par/enlace.sh", *args], timeout=30) if done.returncode != 0: raise RuntimeError(done.stdout + done.stderr) def client(node, protocol, source_port, message, target_port=PORT, *rest): return LAB.out(node, "/w09/bin/demux_client", protocol, SRV, target_port, str(source_port), message, *rest) class A_Demultiplexing(unittest.TestCase): def tearDown(self): for program in ("udp_demux", "tcp_demux"): LAB.sh("srv", "pkill", program) time.sleep(0.2) def start(self, command): LAB.daemon("srv", f"/w09/bin/{command} >/tmp/demux.log 2>&1") time.sleep(0.5) def log(self): return LAB.out("srv", "cat", "/tmp/demux.log") def test_udp_one_socket_for_every_client(self): self.start(f"udp_demux {PORT}") replies = [client("h1", "udp", 5000, "uno"), client("h2", "udp", 5000, "dos"), client("h1", "udp", 5001, "tres")] self.assertEqual(len(set(replies)), 1, replies) for peer in (f"{H1}:5000", f"{H2}:5000", f"{H1}:5001"): self.assertIn(peer, self.log()) sockets = [line for line in LAB.out("srv", "ss", "-uan").splitlines() if f":{PORT}" in line] self.assertEqual(len(sockets), 1, sockets) def test_connected_udp_socket_matches_the_source_too(self): self.start(f"udp_demux {PORT} {H1} 5000") exact = client("h1", "udp", 5000, "uno") other_host = client("h2", "udp", 5000, "dos") other_port = client("h1", "udp", 5001, "tres") self.assertNotEqual(exact, other_host) self.assertEqual(other_host, other_port) def test_tcp_one_socket_per_connection(self): self.start(f"tcp_demux {PORT}") for node, source_port in (("h1", 5000), ("h2", 5000), ("h1", 5001)): LAB.daemon(node, f"/w09/bin/demux_client tcp {SRV} {PORT} {source_port} hola 3") time.sleep(1.0) sockets = LAB.out("srv", "ss", "-tan") established = [line for line in sockets.splitlines() if line.startswith("ESTAB") and f":{PORT}" in line] self.assertEqual(len(established), 3, sockets) for peer in (f"{H1}:5000", f"{H2}:5000", f"{H1}:5001"): self.assertTrue(any(peer in line for line in established), peer) self.assertEqual(len(set(re.findall(r"socket (\d+) de", self.log()))), 3, self.log()) time.sleep(2.5) def test_closed_ports(self): self.assertIn("Connection refused", client("h1", "tcp", 5002, "x")) self.assertIn("Connection refused", client("h1", "udp-connect", 5003, "x")) # Sin connect, el aviso de ICMP no le llega a la aplicación: solo vence su espera. self.assertIn("temporarily unavailable", client("h1", "udp", 5004, "x")) class B_Checksum(unittest.TestCase): def test_my_calculation_matches_a_real_datagram(self): LAB.daemon("srv", "tcpdump -i eth1 -n -U -w /tmp/c.pcap udp port 9009 >/dev/null 2>&1") time.sleep(1.0) client("h1", "udp", 5000, "hola") time.sleep(1.0) LAB.sh("srv", "pkill", "tcpdump") time.sleep(0.5) dump = LAB.out("srv", "tcpdump", "-r", "/tmp/c.pcap", "-n", "-x", "-c", "1") hexa = "".join(re.findall(r"0x[0-9a-f]+:\s+((?:[0-9a-f]{4} ?)+)", dump)).replace(" ", "") self.assertEqual(hexa[:2], "45", dump) udp = bytes.fromhex(hexa[40:]) self.assertEqual(udp[8:], b"hola") captured = int.from_bytes(udp[6:8], "big") self.assertEqual(captured, udp_checksum.udp_checksum(H1, SRV, 5000, 9009, b"hola")) class C_RealStopAndWait(unittest.TestCase): @classmethod def setUpClass(cls): LAB.daemon("srv", f"python3 /w09/tftp_mini.py serve {FOLDER} --timeout-ms 200 >/tmp/tftp.log 2>&1") time.sleep(1.0) @classmethod def tearDownClass(cls): LAB.sh("srv", "pkill", "-f", "tftp_mini") link("limpio") def get(self, *flags): output = LAB.out("h1", "python3", "/w09/tftp_mini.py", "get", SRV, "25k", "/tmp/out", *flags, timeout=120) match = re.search(r"tasa=([\d.]+) kbit/s duplicados=(\d+) sha256=(\w+)", output) self.assertIsNotNone(match, output) return float(match[1]), int(match[2]), match[3] def original_hash(self): return LAB.out("srv", "sha256sum", f"{FOLDER}/25k")[:16] def test_rate_is_set_by_the_rtt(self): link("100mbit", "30") at_30 = self.get()[0] link("100mbit", "10") at_10 = self.get()[0] expected_30 = 512 * 8 / 30 self.assertLess(at_30, expected_30 * 1.1) self.assertGreater(at_30, expected_30 * 0.6) self.assertGreater(at_10, at_30 * 2) def test_a_faster_link_does_not_help(self): link("100mbit", "30") slow = self.get()[0] link("1gbit", "30") fast = self.get()[0] self.assertLess(fast, slow * 1.3) def test_survives_loss_with_the_right_hash(self): link("100mbit", "10", "10", "10") self.assertEqual(self.get()[2], self.original_hash()) def test_without_block_numbers_a_lost_ack_duplicates_data(self): link("100mbit", "10", "0", "15") for _ in range(3): _, duplicates, digest = self.get("--ignore-block-number") if duplicates: self.assertNotEqual(digest, self.original_hash()) return self.fail("no se perdió ningún ACK en tres transferencias") def test_udp_checksum_turns_damage_into_loss(self): link("100mbit", "10", "0", "0", "10") before = self.csum_errors() self.assertEqual(self.get()[2], self.original_hash()) self.assertGreater(self.csum_errors(), before) def csum_errors(self): lines = [line for line in LAB.out("h1", "cat", "/proc/net/snmp").splitlines() if line.startswith("Udp:")] return int(lines[1].split()[lines[0].split().index("InCsumErrors")]) class D_WindowWithTcp(unittest.TestCase): """TCP como protocolo real de ventana: más ventana, más throughput, hasta llenar el enlace.""" @classmethod def setUpClass(cls): link("100mbit", "30") LAB.daemon("h1", "iperf3 -s -p 5209 >/dev/null 2>&1") time.sleep(1.0) @classmethod def tearDownClass(cls): LAB.sh("h1", "pkill", "iperf3") link("limpio") def throughput(self, *options): output = LAB.out("srv", "iperf3", "-c", H1, "-p", "5209", "-t", "3", "-f", "m", *options, timeout=60) match = re.search(r"([\d.]+) Mbits/sec\s+receiver", output) self.assertIsNotNone(match, output) return float(match[1]) def test_doubling_the_window_helps_and_auto_tuning_beats_stop_and_wait(self): small = self.throughput("-w", "16K") large = self.throughput("-w", "64K") auto = self.throughput() self.assertGreater(large, small * 1.8) self.assertGreater(auto, large) # stop-and-wait con paquetes de 1500 bytes y 30 ms: 0,4 Mbps. self.assertGreater(small, 0.4 * 5) self.assertGreater(auto, 0.4 * 50) if __name__ == "__main__": unittest.main()