"""Pruebas de lo que dicen los apuntes sobre la red fabric/fabric.clab.yml.""" import os import re import statistics import subprocess import time import unittest import scenarios import sdn_ctl from netlab import HERE, answer_path, wait_until from scenarios import CUT_MS, H3, LAB, S1_FROM_L1, S2_FROM_L1 ROUTE = "10.214.2.0/24" def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.destroy() LAB.deploy() if not wait_until(scenarios.ecmp_ready, seconds=90): raise RuntimeError("la topología no quedó lista") class Scenario(unittest.TestCase): def setUp(self): self.addCleanup(self.restore) def restore(self): scenarios.reset() self.assertTrue(wait_until(scenarios.ecmp_ready, seconds=45), "no se volvió al estado inicial") class A_Routes(Scenario): def test_leaf_has_two_next_hops_toward_the_other_leaf(self): route = LAB.out("l1", "ip", "route", "show", ROUTE) self.assertIn(f"nexthop via {S1_FROM_L1} dev eth1", route) self.assertIn(f"nexthop via {S2_FROM_L1} dev eth2", route) def test_both_paths_have_the_same_length_and_the_shared_as_blocks_valleys(self): text = LAB.out("l1", "vtysh", "-c", f"show bgp ipv4 unicast {ROUTE}") self.assertEqual(text.count("65140 65142"), 2, text) self.assertIn("multipath", text) def test_without_maximum_paths_there_is_one_next_hop(self): LAB.sh("l1", "vtysh", "-c", "configure terminal", "-c", "router bgp 65141", "-c", "address-family ipv4 unicast", "-c", "maximum-paths 1", "-c", "end") self.addCleanup(LAB.sh, "l1", "vtysh", "-c", "configure terminal", "-c", "router bgp 65141", "-c", "address-family ipv4 unicast", "-c", "maximum-paths 8", "-c", "end") self.assertTrue(wait_until( lambda: "nexthop" not in LAB.out("l1", "ip", "route", "show", ROUTE), seconds=15)) def test_l3_starts_without_bgp(self): self.assertNotIn("router bgp", LAB.out("l3", "vtysh", "-c", "show running-config")) def test_a_lost_link_only_moves_traffic_to_that_leaf(self): LAB.sh("s1", "ip", "link", "set", "eth2", "down") self.assertTrue(wait_until( lambda: "nexthop" not in LAB.out("l1", "ip", "route", "show", ROUTE), seconds=15)) self.assertIn(f"via {S2_FROM_L1}", LAB.out("l1", "ip", "route", "show", ROUTE)) # s1 no aprende la LAN de l2 a través de l1: el AS 65140 ya está en el AS_PATH. self.assertTrue(wait_until( lambda: "Network not in table" in LAB.out("s1", "vtysh", "-c", f"show bgp ipv4 unicast {ROUTE}"), seconds=15)) class B_Hash(Scenario): def distinct_hops(self, policy): scenarios.set_policy(policy) return {scenarios.next_hop(sport) for sport in range(40000, 40064)} def test_policy_0_ignores_ports(self): self.assertEqual(len(self.distinct_hops(0)), 1) def test_policy_1_uses_ports(self): self.assertEqual(self.distinct_hops(1), {S1_FROM_L1, S2_FROM_L1}) def test_policy_3_uses_ports(self): self.assertEqual(self.distinct_hops(3), {S1_FROM_L1, S2_FROM_L1}) def test_the_sysctl_belongs_to_each_node(self): scenarios.set_policy(1, nodes=("l1",)) self.assertIn("= 1", LAB.out("l1", "sysctl", "net.ipv4.fib_multipath_hash_policy")) self.assertIn("= 0", LAB.out("l2", "sysctl", "net.ipv4.fib_multipath_hash_policy")) def test_policy_0_sends_all_flows_through_one_spine(self): scenarios.set_policy(0) shares = scenarios.measure_flows(flows=16, seconds=2) self.assertGreater(max(shares.values()), 0.95) def test_policy_1_spreads_many_flows(self): scenarios.set_policy(1) shares = scenarios.measure_flows(flows=32, seconds=3) # Con 32 flujos, que uno se lleve menos del 3 % es casi imposible; parejo no tiene que salir. self.assertGreater(min(shares.values()), 0.03) def test_colliding_elephants_get_half_of_what_separated_ones_get(self): together, apart = scenarios.elephants() self.assertLess(sum(together), 130) self.assertGreater(sum(apart), 150) self.assertGreater(sum(apart), 1.3 * sum(together)) def test_route_get_predicts_the_real_path_with_policy_3(self): scenarios.set_policy(3) scenarios.start_iperf_servers("h3", 5201) for hop, spine, first in ((S1_FROM_L1, "s1", 44000), (S2_FROM_L1, "s2", 45000)): cport = scenarios.port_on(hop, 5201, first) before = scenarios.spine_tx() scenarios.iperf_mbits("h1", H3, "--cport", str(cport), seconds=1) after = scenarios.spine_tx() delta = {name: after[name] - before[name] for name in after} other = "s2" if spine == "s1" else "s1" self.assertGreater(delta[spine], 20 * max(delta[other], 1000)) class C_PerPacket(Scenario): def test_splitting_a_flow_reorders_it(self): per_flow, per_packet = scenarios.perpacket() # Por paquete hay más segmentos fuera de orden y más retransmisiones, y TCP rinde mucho menos. self.assertGreater(per_packet[2], per_flow[2]) self.assertGreater(per_packet[1], 100 + 2 * per_flow[1]) self.assertLess(per_packet[0], 0.6 * per_flow[0]) class D_Failure(Scenario): def check_cut(self, rows, low_ms, high_ms): self.assertEqual(len(rows), 16) hit = [row[3] for row in rows if row[3] >= CUT_MS] calm = [row[3] for row in rows if row[3] < CUT_MS] # Con 16 flujos, que todos o ninguno pasen por s1 tiene probabilidad 2 × 2^-16. self.assertTrue(hit, "ningún flujo se cortó") self.assertTrue(calm, "todos los flujos se cortaron") for gap in hit: self.assertGreaterEqual(gap, low_ms) self.assertLessEqual(gap, high_ms) return hit def test_loss_of_carrier_is_almost_instantaneous(self): rows = scenarios.carrier() self.assertEqual(len(rows), 16) self.assertLess(max(row[3] for row in rows), 2000) def test_silent_failure_with_timers_3_9(self): self.check_cut(scenarios.silent(), 4000, 14000) def test_silent_failure_with_timers_1_3(self): self.check_cut(scenarios.silent_short(), 1000, 6000) def test_bfd_beats_the_hold_timer(self): rows = scenarios.silent_bfd() self.assertEqual(len(rows), 16) self.assertLess(max(row[3] for row in rows), 2500) class E_Overlay(Scenario): def test_tenants_with_the_same_addresses_stay_apart(self): scenarios.overlay() for left, right in (("ta1", "ta2"), ("tb1", "tb2")): self.assertIn(", 0% packet loss", LAB.out(left, "ping", "-c", "2", "-W", "2", "10.99.0.2")) neighbor = LAB.out(left, "ip", "neigh", "show", "10.99.0.2") self.assertIn(LAB.out(right, "cat", "/sys/class/net/eth1/address").strip(), neighbor) def test_capture_shows_vni_and_inner_frame(self): scenarios.overlay() LAB.sh("ta1", "ping", "-c", "1", "-W", "2", "10.99.0.2") paths = scenarios.capture_vxlan("test-vni", lambda: LAB.sh("ta1", "ping", "-c", "3", "-i", "0.2", "10.99.0.2")) text = scenarios.read_capture(paths, "-Y", "icmp.type == 8", "-T", "fields", "-e", "ip.src", "-e", "ip.dst", "-e", "udp.dstport", "-e", "vxlan.vni") self.assertIn("10.214.0.1,10.99.0.1", text) self.assertIn("10.214.0.2,10.99.0.2", text) self.assertIn("4789", text) self.assertIn("100", text) def test_mtu_black_hole_and_its_two_fixes(self): scenarios.overlay() LAB.sh("ta1", "ping", "-c", "1", "-W", "2", "10.99.0.2") self.assertTrue(scenarios.tenant_ping(1422)) self.assertFalse(scenarios.tenant_ping(1423)) self.assertLess(scenarios.tenant_tcp(), 1.0) scenarios.tenant_mtu() self.assertGreater(scenarios.tenant_tcp(), 5.0) for tenant in scenarios.TENANTS: LAB.sh(tenant, "ip", "link", "set", "eth1", "mtu", "1500") scenarios.jumbo_underlay() self.assertTrue(wait_until(lambda: scenarios.tenant_ping(1472), seconds=10)) self.assertGreater(scenarios.tenant_tcp(), 5.0) def test_without_lowering_the_mtu_linux_warns_with_icmp(self): learned, rate = scenarios.warn() self.assertIn("mtu 1450", learned) self.assertGreater(rate, 5.0) def test_entropy_needs_the_outer_source_port(self): wide, narrow = scenarios.entropy() self.assertGreater(min(wide.values()), 0.03) self.assertGreater(max(narrow.values()), 0.95) def test_each_inner_flow_gets_its_own_outer_source_port(self): scenarios.overlay() LAB.sh("ta1", "ping", "-c", "1", "-W", "2", "10.99.0.2") ports, _ = scenarios.inner_ports() self.assertGreaterEqual(len(ports), 2) class F_Sdn(Scenario): POLICY = os.path.join(HERE, "policy.txt") def delta_for(self, node, port): before = scenarios.spine_tx() rate, _ = scenarios.iperf_mbits(node, H3, port=port, seconds=1) after = scenarios.spine_tx() return rate, {name: after[name] - before[name] for name in after} def test_rules_choose_the_path_by_port_and_can_drop(self): scenarios.set_policy(1) scenarios.start_iperf_servers("h3", 5201, 5202) sdn_ctl.install(self.POLICY) _, via_s1 = self.delta_for("h1", 5201) _, via_s2 = self.delta_for("h1", 5202) self.assertGreater(via_s1["s1"], 20 * max(via_s1["s2"], 1000)) self.assertGreater(via_s2["s2"], 20 * max(via_s2["s1"], 1000)) blocked = LAB.out("h2", "iperf3", "-c", H3, "-p", "5201", "-t", "1", "--connect-timeout", "1500") self.assertIn("unable to connect", blocked.lower()) self.assertRegex(LAB.out("l1", "nft", "list", "table", "ip", "w14steer"), r"drop|counter packets [1-9]") def test_a_pinned_flow_stays_broken_after_bgp_converges(self): scenarios.set_policy(1) scenarios.start_iperf_servers("h3", 5201, 5202) sdn_ctl.install(self.POLICY) scenarios.cut_silently("s1") self.assertTrue(wait_until( lambda: "nexthop" not in LAB.out("l1", "ip", "route", "show", ROUTE), seconds=20)) pinned = LAB.out("h1", "iperf3", "-c", H3, "-p", "5201", "-t", "1", "--connect-timeout", "1500") self.assertIn("unable to connect", pinned.lower()) scenarios.unsteer() scenarios.iperf_mbits("h1", H3, seconds=1) self.assertGreater(scenarios.iperf_mbits("h1", H3, seconds=1)[0], 1) class G_Collective(Scenario): def test_one_slow_host_delays_everybody(self): start, slow = scenarios.collective() for node in ("h1", "h2", "h3", "h4"): self.assertEqual(len(start[node]), 10) self.assertEqual(len(slow[node]), 10) # h3 no toca el enlace lento (h4 hacia h1), y aun así se atrasa con el grupo. self.assertGreater(statistics.median(slow["h3"]), 3 * statistics.median(start["h3"])) class H_Lossy(Scenario): def test_random_loss_hurts_tcp_more_than_it_takes_from_the_link(self): rates = scenarios.wireless() self.assertGreater(rates["sin pérdida"], 60) self.assertLess(rates["1 % de pérdida"], 0.5 * rates["sin pérdida"]) # El enlace que repara sus pérdidas (emulado con jitter) rinde mucho más que el que no. self.assertGreater(rates["sin pérdida, con 4 ms de jitter"], 2 * rates["1 % de pérdida"]) class I_Lab(Scenario): def setUp(self): self.conf = answer_path("l3-bgp.conf") super().setUp() scenarios.load_l3(self.conf) self.addCleanup(LAB.sh, "l3", "vtysh", "-c", "configure terminal", "-c", "no router bgp 65143") self.assertTrue(wait_until(lambda: LAB.out("l3", "ip", "route", "show", ROUTE).count("nexthop") == 2, seconds=40), LAB.out("l3", "vtysh", "-c", "show bgp summary")) def test_l3_joins_the_fabric(self): text = LAB.out("l1", "ip", "route", "show", "10.214.3.0/24") self.assertEqual(text.count("nexthop"), 2, text) self.assertIn(", 0% packet loss", LAB.out("h5", "ping", "-c", "2", "-W", "2", "10.214.1.10")) self.assertIn(", 0% packet loss", LAB.out("h6", "ping", "-c", "2", "-W", "2", "10.214.2.20")) self.assertIn("65140 65143", LAB.out("l1", "vtysh", "-c", "show bgp ipv4 unicast 10.214.3.0/24")) def test_vxlan_between_hosts(self): answer_path("vxlan.sh") done = scenarios.load_vxlan() self.assertEqual(done.returncode, 0, done.stderr) self.addCleanup(LAB.sh, "h2", "ip", "link", "del", "vx42") self.addCleanup(LAB.sh, "h6", "ip", "link", "del", "vx42") self.assertIn("mtu 1450", LAB.out("h2", "ip", "link", "show", "vx42")) self.assertTrue(wait_until( lambda: ", 0% packet loss" in LAB.out("h2", "ping", "-c", "1", "-W", "1", "10.42.0.2"), seconds=15)) pcap = os.path.join(HERE, "captures", "test-lab.pcap") LAB.daemon("h6", "tcpdump -i eth1 -n -s 200 -U -Z root -w /captures/test-lab.pcap udp port 4789") time.sleep(1.5) LAB.sh("h2", "ping", "-c", "3", "-i", "0.2", "10.42.0.2") time.sleep(1) LAB.sh("h6", "pkill", "tcpdump") time.sleep(0.5) text = subprocess.run(["tshark", "-r", pcap, "-Y", "icmp", "-T", "fields", "-e", "ip.src", "-e", "vxlan.vni"], capture_output=True, text=True, timeout=60).stdout self.assertIn("10.214.1.20,10.42.0.1", text) self.assertIn("42", text) class J_Student_View(unittest.TestCase): def test_lab_files_are_templates(self): # Solo para quien publica: ustedes les borran el TODO al completarlos. if not os.path.isdir(os.path.join(HERE, "solution")): self.skipTest("sin solution/, lab/ es de ustedes") for name in ("l3-bgp.conf", "vxlan.sh"): with open(os.path.join(HERE, "lab", name), encoding="utf-8") as file: self.assertIn("TODO", file.read()) def test_policy_parser_orders_and_rejects(self): rules = sdn_ctl.parse_policy(os.path.join(HERE, "policy.txt")) self.assertEqual([rule[0] for rule in rules], [100, 90, 80]) self.assertEqual(rules[0][2], "drop") table = sdn_ctl.compile_rules(rules) self.assertIn("tcp dport 5201 counter meta mark set 1 accept", table)