"""Tormenta de broadcast en un ciclo de switches, y spanning tree que la evita."""
import re
import time
import unittest

import labutil

SWITCHES = ("sw1", "sw2", "sw3")


class Loop(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.addClassCleanup(labutil.destroy, "loop")
        labutil.deploy("loop")
        cls.lab = labutil.Lab("loop")
        cls.addClassCleanup(cls.open_loop)
        time.sleep(8)

    @classmethod
    def open_loop(cls):
        cls.lab.sh("sw1", "ip link set eth3 down")

    def rx(self, host="h2"):
        return int(self.lab.out(host, "cat /sys/class/net/eth1/statistics/rx_packets"))

    def port_state(self, switch, port):
        text = self.lab.out(switch, f"bridge link show dev {port}")
        return re.search(r"state (\w+)", text).group(1)

    def flush_arp(self):
        for host in ("h1", "h2"):
            self.lab.sh(host, "ip neigh flush dev eth1")

    def test_1_storm_without_stp(self):
        lab = self.lab
        self.addCleanup(self.open_loop)
        self.assertTrue(lab.ping("h1", "10.203.0.2", 2), "con el ciclo abierto funciona")
        lab.sh("sw1", "ip link set eth3 up", check=True)
        self.flush_arp()
        before = self.rx()
        lab.ping("h1", "10.203.0.2")
        time.sleep(3)
        middle = self.rx()
        time.sleep(1)
        after = self.rx()
        self.assertGreater(middle - before, 100, "un solo ARP se multiplica")
        self.assertGreater(after - middle, 10, "y sigue circulando sin que nadie envíe nada")
        # La tabla queda mal: sw1 cree que h1, conectado a eth1, está por otro puerto.
        self.assertRegex(lab.out("sw1", "bridge fdb show br br0 | grep 02:00:00:00:00:01"),
                         r"dev eth[23]")
        self.open_loop()
        time.sleep(1)
        cut = self.rx()
        time.sleep(1.5)
        self.assertLess(self.rx() - cut, 10, "al abrir el ciclo, la tormenta se apaga")

    def test_2_stp_blocks_one_port_and_recovers_after_a_failure(self):
        lab = self.lab
        for switch in SWITCHES:
            lab.sh(switch, "ip link set br0 type bridge stp_state 1", check=True)
        lab.sh("sw1", "ip link set eth3 up", check=True)
        self.addCleanup(lab.sh, "sw1", "ip link set eth2 up")
        time.sleep(7)

        # sw1 tiene la menor prioridad: es la raíz y todos sus puertos reenvían.
        self.assertIn("root_port 0", lab.out("sw1", "ip -d link show br0"))
        self.assertEqual(self.port_state("sw1", "eth2"), "forwarding")
        self.assertEqual(self.port_state("sw1", "eth3"), "forwarding")
        # sw3 pierde el enlace con sw2 frente a sw2, que tiene menor prioridad.
        self.assertEqual(self.port_state("sw2", "eth2"), "forwarding")
        self.assertEqual(self.port_state("sw3", "eth1"), "blocking")

        self.flush_arp()
        before = self.rx()
        self.assertTrue(lab.ping("h1", "10.203.0.2", 2))
        time.sleep(2)
        self.assertLess(self.rx() - before, 20, "sin tormenta")

        # Se corta el enlace sw1-sw2. El puerto bloqueado espera a max_age y luego pasa por
        # listening y learning: 20 + 2 + 2 s con estos temporizadores.
        lab.sh("sw1", "ip link set eth2 down", check=True)
        start = time.monotonic()
        self.assertFalse(lab.ping("h1", "10.203.0.2", 1), "h2 queda aislado por un momento")
        recovered = None
        while time.monotonic() - start < 60:
            if lab.ping("h1", "10.203.0.2", 1):
                recovered = time.monotonic() - start
                break
        self.assertIsNotNone(recovered, "STP debe activar el puerto bloqueado")
        self.assertGreater(recovered, 10)
        self.assertLess(recovered, 45)
        self.assertEqual(self.port_state("sw3", "eth1"), "forwarding")


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