"""Comprueba lo que dicen los apuntes sobre sockets, en la interfaz loopback."""
import re
import socket
import struct
import subprocess
import threading
import time
import unittest
import hashlib

from netlab import BUILD, NetTestCase, connect, free_port, recv_line, recv_until_close


def run(name, *args, stdin=None, timeout=20):
    return subprocess.run(
        [str(BUILD / name), *map(str, args)], input=stdin, capture_output=True,
        timeout=timeout,
    )


def wc_reply(port, data):
    sock = connect(port)
    with sock:
        sock.sendall(data)
        sock.shutdown(socket.SHUT_WR)
        return recv_until_close(sock).decode()


class ByteOrder(unittest.TestCase):
    def test_network_order_is_big_endian(self):
        out = run("byte_order").stdout.decode()
        self.assertRegex(out, r"htons\(9002\):\s+23 2a")
        self.assertRegex(out, r"htonl\(0x0a020110\):\s+0a 02 01 10")
        self.assertRegex(out, r"cabecera en 3 bytes:\s+07 23 2a")


class CountingService(NetTestCase):
    def test_client_and_server(self):
        server = self.server("wc_server", 0)
        out = run("wc_client", "127.0.0.1", server.port, stdin=b"hola mundo\nsegunda linea\n")
        self.assertEqual(out.returncode, 0)
        self.assertEqual(out.stdout.decode(), "bytes=25 words=4 lines=2\n")

    def test_client_accepts_a_name(self):
        server = self.server("wc_server", 0)
        out = run("wc_client", "localhost", server.port, stdin=b"uno dos\n")
        self.assertEqual(out.stdout.decode(), "bytes=8 words=2 lines=1\n")

    def test_word_split_between_recvs_counts_once(self):
        server = self.server("wc_server", 0)
        sock = self.connect(server.port)
        sock.sendall(b"hola mu")
        time.sleep(0.2)
        sock.sendall(b"ndo\n")
        sock.shutdown(socket.SHUT_WR)
        self.assertEqual(recv_until_close(sock).decode(), "bytes=11 words=2 lines=1\n")

    def test_large_input_arrives_complete(self):
        server = self.server("wc_server", 0)
        data = b"abc def\n" * 200000
        out = run("wc_client", "127.0.0.1", server.port, stdin=data)
        self.assertEqual(out.stdout.decode(), f"bytes={len(data)} words=400000 lines=200000\n")

    def test_server_survives_a_client_that_resets(self):
        server = self.server("wc_server", 0)
        sock = connect(server.port)
        # SO_LINGER en 0: close envía RST en vez de FIN.
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
        sock.sendall(b"medio mensaje")
        sock.close()
        self.assertEqual(wc_reply(server.port, b"x\n"), "bytes=2 words=1 lines=1\n")
        self.assertTrue(server.alive())

    def test_refused_connection_is_reported(self):
        out = run("wc_client", "127.0.0.1", free_port(), stdin=b"x")
        self.assertNotEqual(out.returncode, 0)
        self.assertIn(b"Connection refused", out.stderr)


class Stream(NetTestCase):
    def transfer(self, recv_args, send_args):
        receiver = self.server("chunk_recv", *recv_args, 0)
        sent = run("chunk_send", "127.0.0.1", receiver.port, *send_args)
        self.assertEqual(sent.returncode, 0, sent.stderr)
        receiver.wait()
        return receiver

    def test_bytes_arrive_complete_and_in_order(self):
        for size, count in ((102400, 1), (1000, 100), (1, 2000)):
            with self.subTest(size=size, count=count):
                r = self.transfer([], [size, count])
                r.wait_output(rf"recv: \d+ llamadas, {size * count} bytes")
                r.wait_output(r"fuera de lugar: 0$")

    def test_recv_all_rebuilds_fixed_records(self):
        receiver = self.transfer(["-x", 100], [50, 10])
        receiver.wait_output(r"recv_all: 5 llamadas, 500 bytes, menor 100, mayor 100")

    def test_recv_does_not_see_send_boundaries(self):
        # 2000 envíos de 1 byte seguidos no deberían dar 2000 recv. El kernel
        # decide: la prueba solo exige que se junte algo, una sola vez en cinco.
        for _ in range(5):
            receiver = self.transfer([], [1, 2000])
            calls = int(receiver.wait_output(r"recv: (\d+) llamadas").group(1))
            if calls < 2000:
                return
        self.fail("cada send llegó en su propio recv en todos los intentos")

    def test_recv_all_reports_early_close(self):
        receiver = self.server("chunk_recv", "-x", 100, 0)
        sock = connect(receiver.port)
        sock.sendall(bytes(range(30)))
        sock.close()
        receiver.wait()
        receiver.wait_output(r"recv_all: 1 llamadas, 30 bytes")

    def test_send_may_write_less_than_requested(self):
        out = run("short_send").stdout.decode().splitlines()
        first = re.search(r"pidieron (\d+) bytes y send devuelve (\d+)", out[0])
        self.assertIsNotNone(first, out)
        self.assertLess(int(first.group(2)), int(first.group(1)))
        self.assertGreater(int(first.group(2)), 0)
        self.assertIn("Resource temporarily unavailable", out[1])
        timed = re.search(r"pidieron (\d+) bytes y send devuelve (\d+)", out[2])
        self.assertLess(int(timed.group(2)), int(timed.group(1)))


class Udp(NetTestCase):
    def test_request_and_reply(self):
        server = self.server("time_server", 0)
        for flags in ([], ["-c"]):
            out = run("time_client", *flags, "127.0.0.1", server.port)
            self.assertEqual(out.returncode, 0, out)
            self.assertRegex(out.stdout.decode(), r"hora: \d\d:\d\d:\d\d UTC")

    def test_server_learns_the_client_from_each_datagram(self):
        server = self.server("time_server", 0)
        a = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        b = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.addCleanup(a.close)
        self.addCleanup(b.close)
        for sock in (a, b):
            sock.settimeout(5)
            sock.sendto(b"x", ("127.0.0.1", server.port))
            self.assertRegex(sock.recv(100).decode(), "UTC")
        ports = {a.getsockname()[1], b.getsockname()[1]}
        self.assertEqual(len(ports), 2)
        for port in ports:
            server.wait_output(rf"127\.0\.0\.1:{port}$")

    def test_server_off_without_connect_goes_unnoticed(self):
        out = run("time_client", "-w", 300, "127.0.0.1", free_port(socket.SOCK_DGRAM))
        self.assertEqual(out.returncode, 2, out)
        self.assertIn("enviados 5 bytes", out.stdout.decode())
        self.assertIn("sin respuesta", out.stdout.decode())

    def test_server_off_with_connect_is_refused_after_the_send(self):
        out = run("time_client", "-c", "-w", 300, "127.0.0.1", free_port(socket.SOCK_DGRAM))
        self.assertEqual(out.returncode, 1, out)
        self.assertIn("enviados 5 bytes", out.stdout.decode())
        self.assertIn(b"Connection refused", out.stderr)

    def test_datagrams_keep_their_boundaries(self):
        receiver = self.server("seq_recv", 0, 200, 500)
        sent = run("seq_send", "127.0.0.1", receiver.port, 200, 100, 100)
        self.assertEqual(sent.returncode, 0, sent.stderr)
        receiver.wait()
        receiver.wait_output(r"tamaño de cada recv: 100 bytes, 0 distintos")


class Concurrency(NetTestCase):
    def blocked_client_is_answered(self, program):
        """¿Responde a B mientras A está conectado sin terminar?"""
        server = self.server(program, 0)
        idle = self.connect(server.port)
        idle.sendall(b"medio")
        time.sleep(0.2)
        other = self.connect(server.port)
        other.settimeout(1.0)
        other.sendall(b"uno\n" if program != "tally_server" else b"TOTAL\n")
        if program != "tally_server":
            other.shutdown(socket.SHUT_WR)
        try:
            return bool(other.recv(100))
        except TimeoutError:
            return False

    def test_iterative_server_makes_the_second_client_wait(self):
        self.assertFalse(self.blocked_client_is_answered("wc_server"))

    def test_fork_server_answers_the_second_client(self):
        self.assertTrue(self.blocked_client_is_answered("wc_fork_server"))

    def test_poll_server_answers_the_second_client(self):
        self.assertTrue(self.blocked_client_is_answered("tally_server"))


class Tally(NetTestCase):
    def test_total_is_shared(self):
        server = self.server("tally_server", 0)
        a = self.connect(server.port)
        b = self.connect(server.port)
        a.sendall(b"SUMA 5\n")
        self.assertEqual(recv_line(a), "total 5\n")
        b.sendall(b"SUMA -2\n")
        self.assertEqual(recv_line(b), "total 3\n")
        a.sendall(b"TOTAL\n")
        self.assertEqual(recv_line(a), "total 3\n")

    def test_partial_line_waits_without_blocking_others(self):
        server = self.server("tally_server", 0)
        a = self.connect(server.port)
        b = self.connect(server.port)
        b.sendall(b"SU")
        time.sleep(0.1)
        a.sendall(b"SUMA 2\n")
        self.assertEqual(recv_line(a), "total 2\n")
        b.sendall(b"MA 10\nTOTAL\n")
        self.assertEqual(recv_line(b), "total 12\n")
        self.assertEqual(recv_line(b), "total 12\n")

    def test_bad_command(self):
        server = self.server("tally_server", 0)
        a = self.connect(server.port)
        for line in (b"hola\n", b"SUMA\n", b"SUMA 1 2\n", b"SUMA x\n"):
            a.sendall(line)
            self.assertEqual(recv_line(a), "error\n")

    def test_long_line_closes_only_that_client(self):
        server = self.server("tally_server", 0)
        a = self.connect(server.port)
        b = self.connect(server.port)
        a.sendall(b"x" * 300)
        self.assertEqual(recv_until_close(a), b"")
        b.sendall(b"TOTAL\n")
        self.assertEqual(recv_line(b), "total 0\n")

    def test_client_that_leaves_mid_line_does_not_matter(self):
        server = self.server("tally_server", 0)
        a = connect(server.port)
        b = self.connect(server.port)
        a.sendall(b"SUMA 1")
        a.close()
        b.sendall(b"TOTAL\n")
        self.assertEqual(recv_line(b), "total 0\n")

    def test_client_that_never_reads_is_cut_off(self):
        server = self.server("tally_server", 0)
        slow = connect(server.port)
        self.addCleanup(slow.close)
        other = self.connect(server.port)
        cut = False
        try:
            for _ in range(4000):
                slow.sendall(b"TOTAL\n" * 2000)
        except (BrokenPipeError, ConnectionResetError, TimeoutError):
            cut = True
        self.assertTrue(cut, "el servidor debió cortar al cliente que no lee")
        other.sendall(b"TOTAL\n")
        self.assertEqual(recv_line(other), "total 0\n")


class ReuseAddress(NetTestCase):
    def serve_once(self, port, *flags):
        """Corre reuse_demo hasta que atienda a un cliente. Devuelve bind's error."""
        with subprocess.Popen(
            [str(BUILD / "reuse_demo"), *flags, str(port)],
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
        ) as server:
            line = server.stdout.readline()
            if "Address already in use" in line:
                return False
            return self.finish_once(server, port)

    def finish_once(self, server, port):
        sock = connect(port)
        with sock:
            self.assertEqual(recv_until_close(sock), b"chau\n")
        self.assertEqual(server.wait(timeout=5), 0)
        return True

    def test_time_wait_blocks_restart_without_the_option(self):
        port = free_port()
        self.assertTrue(self.serve_once(port))
        self.assertFalse(self.serve_once(port))

    def test_both_servers_need_the_option(self):
        port = free_port()
        self.assertTrue(self.serve_once(port))
        self.assertFalse(self.serve_once(port, "-r"))

    def test_option_allows_restart_during_time_wait(self):
        port = free_port()
        self.assertTrue(self.serve_once(port, "-r"))
        self.assertTrue(self.serve_once(port, "-r"))

    def test_option_does_not_share_a_port_being_listened_on(self):
        server = self.server("wc_server", 0)  # wc_server usa SO_REUSEADDR
        out = run("reuse_demo", "-r", server.port)
        self.assertEqual(out.returncode, 1)
        self.assertIn(b"Address already in use", out.stderr)


class EndToEnd(NetTestCase):
    """Un intermediario daña los datos entre dos conexiones TCP correctas."""

    def relay_digest(self, data, every):
        sink = socket.socket()
        self.addCleanup(sink.close)
        sink.bind(("127.0.0.1", 0))
        sink.listen(1)
        received = []

        def collect():
            conn, _ = sink.accept()
            with conn:
                received.append(recv_until_close(conn))

        thread = threading.Thread(target=collect, daemon=True)
        thread.start()
        relay = self.server("flaky_relay", 0, "127.0.0.1", sink.getsockname()[1], every)
        sock = connect(relay.port)
        with sock:
            sock.sendall(data)
        self.assertEqual(relay.wait(), 0)
        thread.join(timeout=10)
        return hashlib.sha256(received[0]).hexdigest(), len(received[0])

    def test_tcp_delivers_damaged_bytes_and_only_the_end_notices(self):
        data = bytes(range(251)) * 8000
        digest, size = self.relay_digest(data, 500000)
        self.assertEqual(size, len(data))
        self.assertNotEqual(digest, hashlib.sha256(data).hexdigest())

    def test_healthy_relay_keeps_the_digest(self):
        data = bytes(range(251)) * 8000
        digest, _ = self.relay_digest(data, 0)
        self.assertEqual(digest, hashlib.sha256(data).hexdigest())


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