"""Pruebas del laboratorio: las políticas de lab/ sobre los seis AS (se saltan si hay TODO).""" import unittest import spvp from netlab import Lab, answer_path, wait_until LAB = Lab("ases", "w06-ases", "redes-w06-ases") ROUTERS = {f"r{n}": 64600 + n for n in range(1, 7)} PREFIXES = {64600 + n: f"100.64.{n}.0/24" for n in range(1, 7)} GRAPH = spvp.ases_graph() def full_path(node, as_path): return [ROUTERS[node]] + as_path def setUpModule(): # Antes de levantar nada: sin archivos listos no hay qué probar. for name in [f"r{n}.conf" for n in range(1, 7)] + ["prefer.conf"]: answer_path(name) unittest.addModuleCleanup(LAB.destroy) LAB.deploy() if not LAB.wait_for_table(ROUTERS, PREFIXES.values()): raise RuntimeError("la topología no quedó lista") for number in range(1, 7): LAB.load_conf(f"r{number}", answer_path(f"r{number}.conf")) if not wait_until(tables_match_the_prediction, seconds=60, step=2): raise RuntimeError("las tablas no llegaron a lo esperado:\n" + describe_tables()) # Un route-map nuevo tarda unos segundos en aplicarse a lo que ya estaba en la tabla. if not wait_until(every_route_is_tagged, seconds=60, step=2): raise RuntimeError("hay rutas sin comunidad: marquen cada ruta según su origen") def expected(node, origin): best = spvp.predicted_paths(origin)[ROUTERS[node]] return list(best[1:]) def tables_match_the_prediction(): return all(LAB.best_path(node, PREFIXES[origin]) == expected(node, origin) for node in ROUTERS for origin in PREFIXES) def every_route_is_tagged(): for node, asn in ROUTERS.items(): for prefix in PREFIXES.values(): if LAB.best_path(node, prefix) in (None, []): continue if f"Community: {asn}:" not in LAB.vtysh(node, f"show bgp ipv4 {prefix}"): return False return True def describe_tables(): lines = [] for node in ROUTERS: for origin, prefix in PREFIXES.items(): lines.append(f"{node} {prefix}: {LAB.best_path(node, prefix)}, " f"se esperaba {expected(node, origin)}") return "\n".join(lines) class A_Export(unittest.TestCase): def test_small_as_announces_only_its_own_prefix(self): self.assertEqual(LAB.advertised("r1", "100.64.2.129"), {"100.64.1.0/24"}) self.assertEqual(LAB.advertised("r1", "100.64.4.129"), {"100.64.1.0/24"}) def test_no_as_passes_a_peer_or_provider_route_upward(self): self.assertEqual(LAB.advertised("r2", "100.64.5.133"), {"100.64.1.0/24", "100.64.2.0/24"}) self.assertEqual(LAB.advertised("r4", "100.64.6.129"), {"100.64.1.0/24", "100.64.4.0/24"}) self.assertEqual(LAB.advertised("r3", "100.64.6.133"), {"100.64.3.0/24"}) def test_peers_exchange_only_customers_and_own(self): self.assertEqual(LAB.advertised("r5", "100.64.5.130"), {"100.64.1.0/24", "100.64.2.0/24", "100.64.5.0/24"}) self.assertEqual(LAB.advertised("r6", "100.64.5.129"), {"100.64.1.0/24", "100.64.3.0/24", "100.64.4.0/24", "100.64.6.0/24"}) def test_customers_receive_everything(self): self.assertEqual(LAB.advertised("r6", "100.64.6.134") | {"100.64.3.0/24"}, set(PREFIXES.values())) class B_Import(unittest.TestCase): def local_pref(self, node, prefix, as_path): """El LOCAL_PREF de un camino. Si nadie lo cambió, vale 100.""" for path in LAB.bgp(node)[prefix]: if path.get("path") == as_path: return path.get("locPrf", 100) self.fail(f"{node} no tiene el camino {as_path} hacia {prefix}") def test_routes_are_tagged_with_a_community_of_the_receiver(self): self.assertTrue(every_route_is_tagged()) def test_customer_beats_peer_even_with_longer_or_equal_path(self): text = LAB.vtysh("r6", "show bgp ipv4 100.64.1.0/24") self.assertIn("64605 64602 64601", text, "r6 también debe conocer el camino por su peer") self.assertEqual(LAB.best_path("r6", "100.64.1.0/24"), [64604, 64601]) self.assertGreater(self.local_pref("r6", "100.64.1.0/24", "64604 64601"), self.local_pref("r6", "100.64.1.0/24", "64605 64602 64601")) def test_customer_above_peer_above_provider(self): # Cada valor se mira donde esa relación existe: r5 tiene cliente y peer, y r2 proveedor. customer = self.local_pref("r5", "100.64.2.0/24", "64602") peer = self.local_pref("r5", "100.64.6.0/24", "64606") provider = self.local_pref("r2", "100.64.5.0/24", "64605") self.assertGreater(customer, peer) self.assertGreater(peer, provider) class C_Valley(unittest.TestCase): def test_every_best_path_everywhere_is_valley_free_and_matches_the_simulation(self): for node in ROUTERS: for origin, prefix in PREFIXES.items(): path = LAB.best_path(node, prefix) self.assertIsNotNone(path, f"{node} no tiene ruta a {prefix}") self.assertTrue(spvp.is_valley_free(GRAPH, full_path(node, path)), (node, path)) self.assertEqual(path, expected(node, origin), (node, prefix)) def test_the_old_shortcut_is_gone(self): self.assertEqual(LAB.best_path("r2", "100.64.4.0/24"), [64605, 64606, 64604]) LAB.assert_hops(self, "r2", "100.64.4.1", ["100.64.5.133", "100.64.5.130", "100.64.4.1"], source="100.64.2.1") def test_everybody_still_reaches_everybody(self): for node, *arguments in (("h1", "100.64.3.10"), ("h3", "100.64.1.10"), ("r2", "-I", "100.64.2.1", "100.64.4.1")): reached = wait_until(lambda: LAB.ok(node, "ping", "-c", "1", "-W", "2", *arguments), seconds=20) self.assertTrue(reached, node) class D_Prefer(unittest.TestCase): """Paso 5: r1 prefiere salir por r2. Corre al final: cambia la tabla de r1.""" TARGET = "100.64.3.0/24" @classmethod def setUpClass(cls): LAB.load_conf("r1", answer_path("prefer.conf")) wait_until(lambda: LAB.best_path("r1", cls.TARGET) == [64602, 64605, 64606, 64603], seconds=30) def test_longer_path_through_r2_wins(self): self.assertEqual(LAB.best_path("r1", self.TARGET), [64602, 64605, 64606, 64603]) self.assertIn("best (Local Pref)", LAB.vtysh("r1", f"show bgp ipv4 {self.TARGET}")) def test_traceroute_goes_through_r2(self): LAB.assert_hops(self, "h1", "100.64.3.10", ["100.64.1.1", "100.64.2.129"]) def test_the_other_direction_does_not_change(self): # local-preference decide por dónde sale el tráfico de r1, no por dónde le entra. LAB.assert_hops(self, "h3", "100.64.1.10", ["100.64.3.1", "100.64.6.133", "100.64.6.130"]) def test_export_rules_survive_the_change(self): self.assertEqual(LAB.advertised("r1", "100.64.2.129"), {"100.64.1.0/24"}) self.assertEqual(LAB.advertised("r1", "100.64.4.129"), {"100.64.1.0/24"}) def test_backup_takes_over_and_primary_returns(self): self.addCleanup(LAB.out, "r1", "ip", "link", "set", "eth2", "up") LAB.out("r1", "ip", "link", "set", "eth2", "down") self.assertTrue(wait_until(lambda: LAB.hops("h1", "100.64.3.10")[1:2] == ["100.64.4.129"], seconds=30)) LAB.out("r1", "ip", "link", "set", "eth2", "up") self.assertTrue(wait_until(lambda: LAB.hops("h1", "100.64.3.10")[1:2] == ["100.64.2.129"], seconds=60)) if __name__ == "__main__": unittest.main()