"""Pruebas del laboratorio de OSPF en ospf/ospf.clab.yml."""
import subprocess
import time
import unittest

from netlab import Lab, answer_path, wait_until

LAB = Lab("ospf", "w05-ospf", "redes-w05-ospf")
R1 = "10.205.0.1"
R2 = "10.205.0.2"
ROUTERS = {"r1": 2, "r2": 3, "r3": 3, "r4": 2, "r5": 2}


def setUpModule():
    unittest.addModuleCleanup(LAB.destroy)
    LAB.deploy()


def run_answer(name):
    done = subprocess.run(["sh", answer_path(name)], capture_output=True, text=True, timeout=120)
    if done.returncode != 0:
        raise AssertionError(f"{name} falló: {done.stdout}{done.stderr}")


def all_full():
    for router, count in ROUTERS.items():
        states = LAB.neighbor_states(router)
        if len(states) != count or not all(s.startswith("Full") for s in states.values()):
            return False
    return True


def adjacency_log(node):
    return [line for line in LAB.out(node, "cat", "/tmp/ospfd.log").splitlines() if "AdjChg" in line]


class A_Starting(unittest.TestCase):
    """Se corre antes que todo lo demás: los routers aún no hablan OSPF."""

    def test_1_no_ospf_at_start(self):
        self.assertNotIn("router ospf", LAB.vtysh("r1", "show running-config"))
        self.assertNotEqual(LAB.sh("h1", "ping", "-c", "1", "-W", "1", "10.205.104.10").returncode, 0)

    def test_2_links_use_ethernet_mtu(self):
        self.assertIn("mtu 1500", LAB.out("r1", "ip", "link", "show", "eth1"))

    def test_3_two_routers_reach_full_through_every_state(self):
        for node, router_id in (("r1", R1), ("r2", R2)):
            LAB.configure(node, "router ospf", f"ospf router-id {router_id}",
                          "log-adjacency-changes detail", "network 10.205.12.0/30 area 0")
        self.assertTrue(wait_until(lambda: LAB.full_with("r1", R2) and LAB.full_with("r2", R1), 40))
        log = "\n".join(adjacency_log("r1"))
        # En un enlace point-to-point no se elige DR: de Init se pasa directo a ExStart.
        for change in ("Down -> Init", "Init -> ExStart", "ExStart -> Exchange"):
            self.assertIn(change, log)
        self.assertRegex(log, r"-> Full")


class B_Lab(unittest.TestCase):
    """Los pasos 2, 3 y 5, con lo que ustedes escriben en lab/."""

    @classmethod
    def setUpClass(cls):
        run_answer("ospf.sh")
        if not wait_until(all_full, 60):
            raise RuntimeError("no todas las adyacencias llegaron a Full")
        wait_until(lambda: LAB.route("r1", "10.205.104.0/24"), 30)

    def test_1_hosts_reach_each_other(self):
        self.assertEqual(LAB.sh("h1", "ping", "-c", "2", "-W", "2", "10.205.104.10").returncode, 0)

    def test_2_lan_is_passive(self):
        for node in ("r1", "r4"):
            self.assertIn("No Hellos (Passive interface)",
                          LAB.vtysh(node, "show ip ospf interface eth3"))

    def test_3_hand_table_matches_the_router(self):
        with open(answer_path("r1_table.txt"), encoding="utf-8") as file:
            lines = [line.split() for line in file if line.strip() and not line.startswith("#")]
        expected = {}
        for prefix, cost, hop in lines:
            expected.setdefault(prefix, (int(cost), set()))[1].add(hop)
        for prefix in ("10.205.0.2/32", "10.205.0.3/32", "10.205.0.4/32", "10.205.0.5/32",
                       "10.205.104.0/24"):
            self.assertIn(prefix, expected, "falta en la tabla")
        for prefix, (cost, hops) in expected.items():
            entries = [e for e in LAB.route("r1", prefix) if e["protocol"] == "ospf"]
            self.assertEqual(len(entries), 1, prefix)
            self.assertEqual(entries[0]["metric"], cost, prefix)
            self.assertEqual({hop["ip"] for hop in entries[0]["nexthops"]}, hops, prefix)

    def test_4_changed_cost_moves_both_directions(self):
        run_answer("cost.sh")
        def forward():
            return LAB.out("h1", "traceroute", "-n", "-q", "1", "-w", "1", "10.205.104.10")
        def backward():
            return LAB.out("h4", "traceroute", "-n", "-q", "1", "-w", "1", "10.205.101.10")
        self.assertTrue(wait_until(lambda: "10.205.35." in forward() and "10.205.24." not in forward(), 30),
                        forward())
        self.assertTrue(wait_until(lambda: "10.205.35." in backward() and "10.205.24." not in backward(), 30),
                        backward())

    def test_5_short_timers_recover_quickly_from_a_silent_cut(self):
        # El corte es en r3-r5, que solo lleva el tráfico después del paso 3.
        answer_path("cost.sh")
        run_answer("timers.sh")
        for router, interface in (("r3", "eth3"), ("r5", "eth1"), ("r1", "eth1")):
            self.assertIn("Hello 2s, Dead 8s", LAB.vtysh(router, f"show ip ospf interface {interface}"))
        # Mientras se aplican uno por uno hay vecinos con valores distintos: mejor esperar
        # a que pase un Dead completo y las adyacencias estén firmes.
        time.sleep(10)
        self.assertTrue(wait_until(all_full, 30))
        ends = (("r3", "eth3"), ("r5", "eth1"))
        for node, interface in ends:
            LAB.sh(node, "tc", "qdisc", "add", "dev", interface, "root", "netem", "loss", "100%")
        def repair():
            for node, interface in ends:
                LAB.sh(node, "tc", "qdisc", "del", "dev", interface, "root")
        self.addCleanup(repair)
        def recovered():
            return LAB.sh("h1", "ping", "-c", "1", "-W", "1", "10.205.104.10").returncode == 0
        # Con OSPF sin enterarse el camino sigue roto; con Dead de 8 s debe volver en poco.
        self.assertFalse(recovered())
        self.assertTrue(wait_until(recovered, 30, 0.5))
        route = LAB.vtysh("r1", "show ip route 10.205.104.0/24")
        self.assertRegex(route, r"10\.205\.12\.2")


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