"""Pruebas sobre la topología: lo que los apuntes dicen que se mide, se mide. Otros programas de la máquina agregan ruido, así que los márgenes son amplios: se revisan tendencias y órdenes de magnitud, no cifras exactas. """ import re import subprocess import unittest import experiments as ex TOPOLOGY = ex.HERE / "topo" / "path.clab.yml" def destroy(): subprocess.run(["containerlab", "destroy", "-t", str(TOPOLOGY), "--cleanup"], capture_output=True, check=False) subprocess.run(["docker", "network", "rm", "redes-w01"], capture_output=True, check=False) class PathTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.addClassCleanup(destroy) destroy() subprocess.run(["containerlab", "deploy", "-t", str(TOPOLOGY)], capture_output=True, check=True) ex.wait_until_ready() ex.start_iperf_server() def setUp(self): ex.clear_links() def test_ttl_expires_at_each_router(self): _, hops = ex.traceroute() self.assertEqual([address for address, _ in hops], ["10.1.0.1", "10.0.12.2", "10.0.23.3", "10.2.0.2"]) def test_rtt_jumps_at_the_long_link(self): ex.set_link(ex.CABLE, delay_ms=ex.CABLE_DELAY_MS) _, hops = ex.traceroute() rtts = [rtt for _, rtt in hops] self.assertLess(rtts[1], 15) self.assertGreater(rtts[2], 40) self.assertLess(rtts[2], 90) self.assertGreater(rtts[3], 40) def test_size_only_matters_through_transmission(self): rtts = ex.rtt_by_size(["r1-r2"]) self.assertGreater(ex.estimated_rate_mbit(rtts, 1), 2.5) self.assertLess(ex.estimated_rate_mbit(rtts, 1), 10) def test_store_and_forward_adds_one_transmission_per_link(self): one = ex.rtt_by_size(["r1-r2"]) three = ex.rtt_by_size(["r1-r2", "h1-r1", "r3-h2"]) ratio = (three[1472] - three[64]) / (one[1472] - one[64]) self.assertGreater(ratio, 2) self.assertLess(ratio, 5) def test_propagation_does_not_depend_on_size(self): without = ex.rtt_by_size(["r1-r2"]) cable = ex.rtt_by_size(["r1-r2"], cable=True) differences = [cable[size] - without[size] for size in ex.SIZES] for difference in differences: self.assertGreater(difference, 35) self.assertLess(difference, 65) self.assertLess(max(differences) - min(differences), 12) def test_slowest_link_limits_throughput_wherever_it_is(self): for slow in ("r1-r2", "r2-r3", "r3-h2"): with self.subTest(slow=slow): ex.clear_links() for link in ("r1-r2", "r2-r3", "r3-h2"): ex.set_link(link, rate_mbit=50 if link == slow else 200) measured = ex.tcp_throughput_mbit(seconds=4) self.assertGreater(measured, 25) self.assertLess(measured, 52) def test_faster_bottleneck_gives_more_throughput(self): ex.set_link("r1-r2", rate_mbit=20) slow = ex.tcp_throughput_mbit(seconds=3) ex.clear_links() ex.set_link("r1-r2", rate_mbit=100) fast = ex.tcp_throughput_mbit(seconds=3) self.assertLess(slow, 21) self.assertGreater(fast, 2 * slow) def test_queue_fills_and_drops_above_one(self): ex.set_link("r1-r2", rate_mbit=ex.QUEUE_RATE_MBIT, limit=ex.QUEUE_LIMIT) light = ex.udp_load(0.5) heavy = ex.udp_load(1.2) self.assertLess(light["lost"], 2) self.assertLess(light["avg"], 15) self.assertGreater(heavy["lost"], 8) self.assertLess(heavy["lost"], 30) # Buffer lleno: 200 paquetes de 1442 bytes a 50 Mbps son 46 ms. self.assertGreater(heavy["avg"], 30) self.assertLess(heavy["avg"], 90) def test_bursts_queue_at_the_same_intensity(self): ex.set_link("r1-r2", rate_mbit=ex.QUEUE_RATE_MBIT, limit=ex.QUEUE_LIMIT) bursts = ex.udp_load(0.5, burst=100) # Una ráfaga de 100 paquetes de 1442 bytes tarda 23 ms en salir. self.assertGreater(bursts["max"], 10) self.assertLess(bursts["lost"], 2) def test_router_changes_link_and_ttl_but_not_addresses_or_transport(self): paths = ex.capture_web() near_h1 = ex.packet_fields(paths["near-h1"]) near_h2 = ex.packet_fields(paths["near-h2"]) self.assertGreaterEqual(len(near_h1), 6) self.assertEqual(len(near_h1), len(near_h2)) # Se comparan solo los paquetes de h1: en cada punto de captura, un paquete # de ida y uno de vuelta pueden verse en distinto orden. sent_near_h1 = [packet for packet in near_h1 if packet["ip.src"] == "10.1.0.2"] sent_near_h2 = [packet for packet in near_h2 if packet["ip.src"] == "10.1.0.2"] self.assertEqual(len(sent_near_h1), len(sent_near_h2)) from_h1 = list(zip(sent_near_h1, sent_near_h2)) self.assertTrue(from_h1) for before, after in from_h1: self.assertNotEqual(before["eth.src"], after["eth.src"]) self.assertNotEqual(before["eth.dst"], after["eth.dst"]) self.assertEqual(int(before["ip.ttl"]) - int(after["ip.ttl"]), 3) for field in ("ip.src", "ip.dst", "tcp.seq_raw", "tcp.len"): self.assertEqual(before[field], after[field]) def test_layers_agrees_with_tshark(self): paths = ex.capture_web() out = subprocess.run([str(ex.HERE / "layers"), str(paths["near-h1"])], capture_output=True, text=True, check=True).stdout command = ["tshark", "-r", str(paths["near-h1"]), "-T", "fields", "-e", "frame.len", "-e", "ip.hdr_len", "-e", "tcp.hdr_len", "-e", "tcp.len", "-e", "ip.ttl"] reference = subprocess.run(command, capture_output=True, text=True, check=True).stdout.splitlines() lines = out.splitlines() self.assertEqual(len(lines), len(reference)) for line, fields in zip(lines, reference): total, ip_size, tcp_size, app, ttl = fields.split("\t") pattern = rf"total={total}\s+enlace=14 ip={ip_size} tcp={tcp_size} app={app} ttl={ttl} " self.assertTrue(re.search(pattern, line), f"{line!r} no coincide con {fields!r}") if __name__ == "__main__": unittest.main()