"""Pruebas de OSPF sobre el grafo del capítulo, six/six.clab.yml."""
import json
import os
import re
import subprocess
import time
import unittest

from netlab import HERE, Lab, wait_until

LAB = Lab("six", "w05-six", "redes-w05-six")
NODES = "abcdef"
EDGES = [("a", "b", 3), ("a", "c", 1), ("b", "c", 4), ("b", "d", 2), ("c", "d", 5),
         ("c", "e", 5), ("d", "e", 1), ("d", "f", 7), ("e", "f", 2)]


def number(node):
    return NODES.index(node) + 1


def loopback(node):
    return f"10.205.0.{number(node)}"


def peer_addresses():
    """Dirección de cada extremo de cada enlace: {ip: router}."""
    owners = {}
    for left, right, _ in EDGES:
        owners[f"10.205.{number(left)}{number(right)}.1"] = left
        owners[f"10.205.{number(left)}{number(right)}.2"] = right
    return owners


def neighbor_count(node):
    return sum(node in edge[:2] for edge in EDGES)


def all_full():
    for node in NODES:
        states = LAB.neighbor_states(node)
        if len(states) != neighbor_count(node) or not all(s.startswith("Full") for s in states.values()):
            return False
    return True


def setUpModule():
    unittest.addModuleCleanup(LAB.destroy)
    LAB.deploy()
    if not wait_until(all_full, 60):
        raise RuntimeError("las adyacencias no llegaron a Full")
    wait_until(lambda: LAB.route("a", "10.205.0.6/32"), 30)


def dijkstra_table(source):
    out = subprocess.run([os.path.join(HERE, "dijkstra"), "graphs/six.txt", source],
                         capture_output=True, text=True, cwd=HERE, timeout=10).stdout
    table = {}
    for line in out.split("forwarding table")[1].splitlines()[2:]:
        fields = line.split()
        table[fields[0]] = (int(fields[1]), set(fields[2:]))
    return table


def via(node, destination):
    """El next hop de la forwarding table del kernel hacia una dirección."""
    match = re.search(r"via (\S+)", LAB.out(node, "ip", "route", "get", destination))
    return match.group(1) if match else None


def cut_silently(*ends):
    """Corta un enlace sin que se note en la interfaz: todo se pierde, el carrier sigue."""
    for node, interface in ends:
        LAB.sh(node, "tc", "qdisc", "add", "dev", interface, "root", "netem", "loss", "100%")


def repair(*ends):
    for node, interface in ends:
        LAB.sh(node, "tc", "qdisc", "del", "dev", interface, "root")


BD_LINK = (("b", "eth3"), ("d", "eth1"))
THROUGH_B = "10.205.12.2"
THROUGH_C = "10.205.13.2"


def time_until_reroute(limit):
    start = time.monotonic()
    if not wait_until(lambda: via("a", loopback("d")) == THROUGH_C, limit, 0.2):
        return None
    return time.monotonic() - start


def back_to_normal():
    return wait_until(lambda: LAB.full_with("b", loopback("d")) and LAB.full_with("d", loopback("b"))
                      and via("a", loopback("d")) == THROUGH_B, 60)


class A_MatchesTheGraph(unittest.TestCase):
    def test_routes_match_dijkstra_from_every_node(self):
        owners = peer_addresses()
        for source in NODES:
            for destination, (cost, hops) in dijkstra_table(source).items():
                entries = [e for e in LAB.route(source, f"{loopback(destination)}/32")
                           if e["protocol"] == "ospf"]
                self.assertEqual(len(entries), 1, (source, destination))
                self.assertEqual(entries[0]["metric"], cost, (source, destination))
                found = {owners[hop["ip"]] for hop in entries[0]["nexthops"]}
                self.assertEqual(found, hops, (source, destination))

    def test_tie_is_installed_in_the_kernel_as_ecmp(self):
        table = LAB.out("a", "ip", "route", "show", "10.205.0.5")
        self.assertIn(f"nexthop via {THROUGH_B}", table)
        self.assertIn(f"nexthop via {THROUGH_C}", table)

    def test_every_router_holds_the_same_map(self):
        def lsas(node):
            text = LAB.vtysh(node, "show ip ospf database")
            return sorted((f[0], f[1], f[3], f[4]) for f in
                          (line.split() for line in text.splitlines())
                          if len(f) >= 5 and f[3].startswith("0x8"))
        self.assertTrue(wait_until(lambda: all(lsas(n) == lsas("a") for n in NODES), 20))
        self.assertEqual(len(lsas("a")), 6)

    def test_router_lsa_lists_the_costs_of_its_links(self):
        text = LAB.vtysh("a", f"show ip ospf database router {loopback('b')}")
        self.assertIn("Number of Links: 7", text)  # loopback y 2 por cada uno de sus 3 enlaces
        self.assertIn(f"Neighboring Router ID: {loopback('d')}", text)
        self.assertRegex(text, r"TOS 0 Metric: 2")
        self.assertRegex(text, r"TOS 0 Metric: 4")

    def test_point_to_point_means_no_dr(self):
        self.assertTrue(LAB.neighbor_states("a")[loopback("b")].startswith("Full/-"))
        self.assertIn("Network Type POINTOPOINT", LAB.vtysh("a", "show ip ospf interface eth1"))
        self.assertNotIn("Net Link States", LAB.vtysh("a", "show ip ospf database"))


class B_RoutingVersusForwarding(unittest.TestCase):
    def test_kernel_metric_is_not_the_ospf_cost(self):
        self.assertEqual(LAB.route("a", "10.205.0.4/32")[0]["metric"], 5)
        self.assertIn("metric 20", LAB.out("a", "ip", "route", "show", "10.205.0.4"))

    def test_connected_beats_ospf_for_own_link(self):
        entries = LAB.route("a", "10.205.12.0/30")
        selected = {e["protocol"]: e.get("selected", False) for e in entries}
        self.assertEqual(selected.get("connected"), True)
        self.assertFalse(selected.get("ospf"))

    def test_static_route_wins_and_ospf_keeps_its_own_answer(self):
        self.addCleanup(LAB.configure, "a", "no ip route 10.205.0.4/32 10.205.13.2")
        LAB.configure("a", "ip route 10.205.0.4/32 10.205.13.2")
        self.assertTrue(wait_until(lambda: via("a", loopback("d")) == THROUGH_C, 10))
        entries = {e["protocol"]: e for e in LAB.route("a", "10.205.0.4/32")}
        self.assertTrue(entries["static"].get("selected"))
        self.assertFalse(entries["ospf"].get("selected"))
        self.assertEqual(entries["ospf"]["metric"], 5)
        ospf_table = LAB.vtysh("a", "show ip ospf route")
        self.assertRegex(ospf_table, r"10\.205\.0\.4/32\s+\[5\][^\n]*\n\s+via 10\.205\.12\.2")
        # Sin nombre para ese número en esta imagen: 196 es el protocolo "static" de FRRouting.
        self.assertRegex(LAB.out("a", "ip", "route", "show", "10.205.0.4"), r"proto (196|static)")
        LAB.configure("a", "no ip route 10.205.0.4/32 10.205.13.2")
        self.assertTrue(wait_until(lambda: via("a", loopback("d")) == THROUGH_B, 10))


class C_Failures(unittest.TestCase):
    """El orden importa: primero los temporizadores por defecto, después los cortos."""

    def setUp(self):
        self.assertTrue(back_to_normal())

    def test_1_link_down_is_noticed_at_once(self):
        self.addCleanup(LAB.sh, "b", "ip", "link", "set", "eth3", "up")
        LAB.sh("b", "ip", "link", "set", "eth3", "down")
        self.assertIn("NO-CARRIER", LAB.out("d", "ip", "link", "show", "eth1"))
        elapsed = time_until_reroute(8)
        self.assertIsNotNone(elapsed)
        self.assertLess(elapsed, 8)
        self.assertEqual(LAB.sh("a", "ping", "-c", "2", "-W", "2", "-I", loopback("a"),
                                loopback("d")).returncode, 0)

    def test_2_silent_failure_waits_for_dead_interval(self):
        self.addCleanup(repair, *BD_LINK)
        cut_silently(*BD_LINK)
        self.assertIn("LOWER_UP", LAB.out("d", "ip", "link", "show", "eth1"))
        # Por ahora OSPF no se da cuenta, y lo que va por ese camino se pierde.
        self.assertEqual(via("a", loopback("d")), THROUGH_B)
        self.assertNotEqual(LAB.sh("a", "ping", "-c", "2", "-W", "1", "-I", loopback("a"),
                                   loopback("d")).returncode, 0)
        elapsed = time_until_reroute(70)
        self.assertIsNotNone(elapsed, "no hubo cambio de ruta")
        self.assertGreater(elapsed, 20)
        self.assertEqual(LAB.sh("a", "ping", "-c", "2", "-W", "2", "-I", loopback("a"),
                                loopback("d")).returncode, 0)

    def test_3_short_blip_does_not_drop_a_default_adjacency(self):
        self.addCleanup(repair, *BD_LINK)
        cut_silently(*BD_LINK)
        time.sleep(12)
        repair(*BD_LINK)
        self.assertTrue(LAB.full_with("b", loopback("d")))
        self.assertEqual(via("a", loopback("d")), THROUGH_B)

    def set_short_timers(self):
        for node, interface in BD_LINK:
            LAB.configure(node, f"interface {interface}", "ip ospf hello-interval 2",
                          "ip ospf dead-interval 8")
            self.addCleanup(LAB.configure, node, f"interface {interface}",
                            "no ip ospf hello-interval", "no ip ospf dead-interval")
        self.assertIn("Hello 2s, Dead 8s", LAB.vtysh("b", "show ip ospf interface eth3"))
        self.assertTrue(wait_until(lambda: LAB.full_with("b", loopback("d")), 15))

    def test_4_short_timers_detect_sooner_and_also_trip_on_a_blip(self):
        self.set_short_timers()
        self.addCleanup(repair, *BD_LINK)
        cut_silently(*BD_LINK)
        elapsed = time_until_reroute(30)
        self.assertIsNotNone(elapsed, "no hubo cambio de ruta")
        self.assertLess(elapsed, 15)
        repair(*BD_LINK)
        self.assertTrue(back_to_normal())
        # El mismo corte de 12 s del que el valor por defecto ni se entera.
        cut_silently(*BD_LINK)
        self.assertTrue(wait_until(lambda: not LAB.full_with("b", loopback("d")), 14))
        repair(*BD_LINK)


class D_AdjacencyProblems(unittest.TestCase):
    """Cada caso rompe el enlace a-b, lo diagnostica y lo deja como estaba."""

    def setUp(self):
        self.assertTrue(wait_until(lambda: LAB.full_with("a", loopback("b"))
                                   and LAB.full_with("b", loopback("a")), 60))
        LAB.sh("a", "sh", "-c", "> /tmp/ospfd.log")

    def log_has(self, text, seconds=25):
        return wait_until(lambda: text in LAB.out("a", "cat", "/tmp/ospfd.log"), seconds)

    def test_different_area(self):
        def restore():
            LAB.configure("a", "router ospf", "no network 10.205.12.0/30 area 1",
                          "no network 10.205.13.0/30 area 0", "no network 10.205.0.1/32 area 0",
                          "network 10.205.0.0/16 area 0")
        self.addCleanup(restore)
        LAB.configure("a", "router ospf", "no network 10.205.0.0/16 area 0",
                      "network 10.205.12.0/30 area 1", "network 10.205.13.0/30 area 0",
                      "network 10.205.0.1/32 area 0")
        self.assertTrue(self.log_has("invalid Area ID"))
        self.assertNotIn(loopback("b"), LAB.neighbor_states("a"))
        self.assertIn("Area 0.0.0.1", LAB.vtysh("a", "show ip ospf interface eth1"))
        self.assertIn("Area 0.0.0.0", LAB.vtysh("b", "show ip ospf interface eth1"))

    def test_different_timers(self):
        self.addCleanup(LAB.configure, "a", "interface eth1", "no ip ospf dead-interval")
        LAB.configure("a", "interface eth1", "ip ospf dead-interval 20")
        self.assertTrue(self.log_has("RouterDeadInterval mismatch"))
        # La adyacencia ya formada no cae enseguida: cae cuando vence el Dead de a.
        self.assertTrue(wait_until(lambda: loopback("b") not in LAB.neighbor_states("a"), 30))
        self.assertIn("Dead 20s", LAB.vtysh("a", "show ip ospf interface eth1"))
        self.assertIn("Dead 40s", LAB.vtysh("b", "show ip ospf interface eth1"))

    def set_mask(self, new, old):
        LAB.configure("a", "interface eth1", f"no ip address 10.205.12.1/{old}",
                      f"ip address 10.205.12.1/{new}")

    def test_different_mask_is_ignored_on_point_to_point(self):
        self.addCleanup(self.set_mask, 30, 29)
        self.set_mask(29, 30)
        self.assertTrue(wait_until(lambda: LAB.full_with("a", loopback("b")), 30))
        self.assertNotIn("NetworkMask mismatch", LAB.out("a", "cat", "/tmp/ospfd.log"))

    def test_different_mask_blocks_on_broadcast(self):
        def restore():
            self.set_mask(30, 29)
            for node in ("a", "b"):
                LAB.configure(node, "interface eth1", "ip ospf network point-to-point")
        self.addCleanup(restore)
        for node in ("a", "b"):
            LAB.configure(node, "interface eth1", "no ip ospf network")
        self.set_mask(29, 30)
        self.assertTrue(self.log_has("NetworkMask mismatch"))
        self.assertNotIn(loopback("b"), LAB.neighbor_states("a"))

    def test_different_mtu_stops_at_exstart(self):
        def restore():
            LAB.sh("a", "ip", "link", "set", "eth1", "mtu", "1500")
        self.addCleanup(restore)
        LAB.sh("a", "ip", "link", "set", "eth1", "mtu", "1400")
        LAB.sh("b", "ip", "link", "set", "eth1", "down")
        LAB.sh("b", "ip", "link", "set", "eth1", "up")
        self.assertTrue(self.log_has("MTU 1500 is larger", 40))
        self.assertTrue(wait_until(
            lambda: LAB.neighbor_states("a").get(loopback("b"), "").startswith("ExStart"), 20))
        # Sigue atascado: el intercambio no avanza mientras el MTU no coincida.
        time.sleep(6)
        self.assertTrue(LAB.neighbor_states("a")[loopback("b")].startswith("ExStart"))


class E_DefaultCost(unittest.TestCase):
    def test_default_cost_depends_on_bandwidth(self):
        def restore():
            LAB.configure("a", "interface eth1", "no bandwidth", "ip ospf cost 3")
        self.addCleanup(restore)
        LAB.configure("a", "interface eth1", "no ip ospf cost")
        self.assertIn("Cost: 10", LAB.vtysh("a", "show ip ospf interface eth1"))
        self.assertIn("BW 10000 Mbit", LAB.vtysh("a", "show ip ospf interface eth1"))
        LAB.configure("a", "interface eth1", "bandwidth 1000")
        self.assertTrue(wait_until(lambda: "Cost: 100" in LAB.vtysh("a", "show ip ospf interface eth1"), 5))


if __name__ == "__main__":
    unittest.main()