"""Las afirmaciones de los apuntes, medidas de verdad sobre dumbbell/dumbbell.clab.yml.

Todo mide tendencias con márgenes amplios: las cifras cambian según la computadora y
según lo que esté corriendo en ella.
"""
import csv
import os
import re
import statistics
import tempfile
import time
import unittest

import bench
import netlab
from netlab import RECEIVER, sh


def setUpModule():
    unittest.addModuleCleanup(netlab.destroy)
    netlab.deploy()
    if not netlab.wait_until(lambda: "bytes from" in sh("s1", f"ping -c 1 -W 2 {RECEIVER}", check=False)
                             and "bytes from" in sh("s2", f"ping -c 1 -W 2 {RECEIVER}", check=False),
                             seconds=60):
        raise RuntimeError("la topología no quedó lista")


def ratios(rows):
    """ssthresh nuevo sobre el cwnd más alto desde el evento anterior."""
    found, last, peak = [], None, 0
    for _, cwnd, ssthresh, _ in rows:
        peak = max(peak, cwnd)
        if ssthresh is not None and ssthresh != last:
            found.append(ssthresh / peak)
            last, peak = ssthresh, cwnd
    return found


class Measured(unittest.TestCase):
    def setUp(self):
        bench.start_servers()
        self.addCleanup(bench.stop_servers)

    def test_kernel_offers_reno_and_cubic(self):
        offered = sh("s1", "cat /proc/sys/net/ipv4/tcp_available_congestion_control").split()
        self.assertIn("reno", offered)
        self.assertIn("cubic", offered)

    def test_single_flow_fills_the_link_and_shows_its_decrease(self):
        bench.set_link()
        for cc, low, high in (("reno", 0.30, 0.62), ("cubic", 0.60, 0.85)):
            result, rows = bench.run_flow(cc, 20)
            self.assertEqual(result["cc"], cc)
            self.assertGreater(result["mbps"], 30, cc)
            self.assertLess(result["mbps"], 52, cc)
            self.assertGreater(len(rows), 50, cc)
            self.assertGreater(min(row[3] for row in rows), 38, "el RTT no puede bajar de 40 ms")
            found = ratios(rows)
            self.assertGreater(len(found), 1, f"{cc}: casi no se vieron reducciones")
            self.assertGreater(statistics.median(found), low, f"{cc}: {found}")
            self.assertLess(statistics.median(found), high, f"{cc}: {found}")

    def test_plot_script_makes_a_png(self):
        bench.set_link()
        _, rows = bench.run_flow("cubic", 3)
        with tempfile.TemporaryDirectory() as folder:
            source = os.path.join(folder, "cubic.csv")
            target = os.path.join(folder, "cwnd.png")
            bench.write_csv(source, rows)
            with open(source, encoding="utf-8") as file:
                self.assertEqual(next(csv.reader(file))[1], "cwnd_segments")
            done = os.system(f"python3 {netlab.HERE}/plot_cwnd.py {target} {source} >/dev/null")
            self.assertEqual(done, 0)
            self.assertGreater(os.path.getsize(target), 1000)

    def test_tcp_info_shows_exponential_slow_start(self):
        bench.set_link()
        receiver = netlab.spawn("rx", "/lab/tcp_info recv 6000")
        time.sleep(0.5)
        output = sh("s1", f"/lab/tcp_info send {RECEIVER} 6000 reno 2 10", timeout=30)
        receiver.communicate(timeout=20)
        self.assertIn("algoritmo: reno", output)
        rows = [line.split() for line in output.splitlines() if not line.startswith("#")]
        self.assertEqual(rows[0][1], "10", "Linux empieza con 10 segmentos")
        self.assertEqual(rows[0][2], "-", "todavía no hay ssthresh")
        # Antes del primer RTT no ha vuelto ningún ACK de datos: la ventana sigue en 10.
        early = [int(row[1]) for row in rows if float(row[0]) < 30]
        self.assertTrue(early and all(value == 10 for value in early), early)
        # Una muestra puede caer a mitad de un RTT, así que no se exigen valores exactos. A los
        # 300 ms, unos 7 RTT, sumar uno por RTT daría 17; doblar da cientos.
        later = [int(row[1]) for row in rows if float(row[0]) >= 300]
        self.assertGreater(later[0], 80, "slow start debe crecer de forma exponencial")
        self.assertTrue(any(row[2] != "-" for row in rows), "nunca hubo pérdida")
        self.assertTrue(any(row[5] == "recovery" for row in rows), "nunca entró a recovery")

    def test_random_loss_hurts_and_matches_the_formula(self):
        bench.set_link(loss=1)
        for cc in ("reno", "cubic"):
            result, _ = bench.run_flow(cc, 15, sample=False)
            self.assertLess(result["mbps"], 15, f"{cc}: con 1 % de pérdida ya no llena el enlace")
            if cc == "reno":
                expected = 1.22 * 1448 * 8 / (0.040 * 0.1) / 1e6
                self.assertGreater(result["mbps"], expected / 3)
                self.assertLess(result["mbps"], expected * 3)

    def test_large_buffer_means_large_delay(self):
        bench.set_link(buffer_kb=2000)
        large = bench.run_bloat("reno", 20)
        bench.set_link(buffer_kb=50)
        small = bench.run_bloat("reno", 20)
        for found in (large, small):
            self.assertLess(found["ping_idle_ms"], 60)
            self.assertGreater(found["ping_received"], 0.6 * found["ping_sent"])
        self.assertGreater(large["ping_load_ms"], large["ping_idle_ms"] + 80)
        self.assertLess(small["ping_load_ms"], small["ping_idle_ms"] + 40)
        self.assertGreater(large["ping_load_ms"], small["ping_load_ms"] * 2)

    def test_same_algorithm_shares_the_link(self):
        bench.set_link()
        one, two = bench.run_pair("reno", "reno", 30)
        self.assertGreater(bench.jain([one["mbps"], two["mbps"]]), 0.7, (one, two))
        self.assertGreater(one["mbps"] + two["mbps"], 30)

    def test_short_rtt_beats_long_rtt_with_reno(self):
        bench.set_link(extra_s2=80)
        short, long = bench.run_pair("reno", "reno", 30)
        self.assertGreater(short["mbps"], 1.5 * long["mbps"], (short, long))

    def test_parallel_connections_take_a_larger_share(self):
        bench.set_link()
        one, many = bench.run_pair("cubic", "cubic", 30, parallel2=4)
        self.assertGreater(many["mbps"], 1.3 * one["mbps"], (one, many))

    def counters(self):
        stats = bench.queue_stats()
        marked = re.search(r"marked (\d+) early (\d+) pdrop (\d+)", stats)
        self.assertIsNotNone(marked, stats)
        return [int(value) for value in marked.groups()]

    def test_ecn_marks_where_red_would_drop(self):
        bench.set_link(queue="red-ecn", buffer_kb=1000, ecn=True)
        self.addCleanup(bench.set_link)
        bench.run_flow("cubic", 12, sample=False)
        marked, early, pdrop = self.counters()
        self.assertGreater(marked, 0, "con ECN, RED marca")
        self.assertEqual(early, 0, "con ECN, RED no bota por anticipado")

        bench.set_link(queue="red-ecn", buffer_kb=1000, ecn=False)
        bench.run_flow("cubic", 12, sample=False)
        marked, early, pdrop = self.counters()
        self.assertEqual(marked, 0, "sin ECN, no hay a quién marcar")
        self.assertGreater(early + pdrop, 0, "sin ECN, la misma cola bota")


class Compiled(unittest.TestCase):
    def test_compiles_without_warnings(self):
        with tempfile.TemporaryDirectory() as folder:
            target = os.path.join(folder, "tcp_info")
            source = os.path.join(netlab.HERE, "tcp_info.c")
            done = os.system(f"gcc -Wall -Wextra -Werror -std=c17 -D_DEFAULT_SOURCE -o {target} {source}")
            self.assertEqual(done, 0)


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