"""
Pruebas de la sincronización entre dos Caudal (cartera/enlace.py): el de un
servidor y otro que se enlaza con él. Con dos instalaciones de verdad, una
atendiendo por la red, y sobre todo los casos en que mezclar podría romper algo.

  python3 -m unittest discover -s pruebas -v
"""

from __future__ import annotations

import os
import socket
import sys
import tempfile
import threading
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from cartera import almacen, btc, enlace, servidor  # noqa: E402
from cartera.datos import huella_datos  # noqa: E402
from cartera.enlace import ErrorEnlace  # noqa: E402
from cartera.operaciones import Cartera, ErrorOperacion  # noqa: E402

try:
    import cryptography  # noqa: F401
    import requests  # noqa: F401
    HAY_TODO = True
except ImportError:
    HAY_TODO = False

CLAVE = "mi contraseña de Caudal"


class DosCaudales(unittest.TestCase):
    """Base: un Caudal en un servidor y otro enlazado con él."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.addCleanup(almacen.poner_llave, None)
        # El servidor (como en Start9) …
        self.srv = servidor.arrancar(Path(self.tmp.name) / "servidor" / "datos", 0, "127.0.0.1")
        threading.Thread(target=self.srv.servidor.serve_forever, daemon=True).start()
        self.addCleanup(self.apagar)
        self.srv.entrar(self.srv.acceso.crear(CLAVE))
        self.url = f"http://127.0.0.1:{self.srv.puerto}"
        # … y el otro Caudal (como el del USB)
        self.usb = servidor.App(Cartera(Path(self.tmp.name) / "usb" / "datos"))
        self.addCleanup(self.usb.cartera.cerrar)
        self.usb.enlace_configurar({"url": self.url, "clave": CLAVE, "nombre": "USB"})

    def apagar(self):
        self.srv.servidor.shutdown()
        self.srv.servidor.server_close()
        if self.srv.cartera is not None:
            self.srv.cartera.cerrar()

    def sincronizar(self, **mas):
        return self.usb.enlace_sincronizar(mas)

    def iguales(self):
        self.assertEqual(huella_datos(self.usb.cartera.datos), huella_datos(self.srv.cartera.datos))

    def fondo(self, app, nombre="Fondo", **mas):
        a = app.cartera.guardar_activo({"nombre": nombre, "tipo": "Fondo", **mas})
        m = app.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-01-10", "cantidad": 10, "precio": 100})
        return a, m

    def nombres(self, app):
        return sorted(a["nombre"] for a in app.cartera.datos.vivos("activos"))


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestEnlazarConCodigo(DosCaudales):
    """Enlazar sin teclear la contraseña: el servidor enseña un código de pocos minutos."""

    def otro(self, nombre="otro"):
        app = servidor.App(Cartera(Path(self.tmp.name) / nombre / "datos"))
        self.addCleanup(app.cartera.cerrar)
        return app

    def test_con_el_codigo_se_enlaza_y_sincroniza_sin_contraseña(self):
        self.fondo(self.srv, "Del servidor")
        codigo = self.srv.enlace_codigo()["codigo"]
        self.assertRegex(codigo, r"^[A-Z2-9]{5}-[A-Z2-9]{5}$")
        movil = self.otro()
        movil.enlace_configurar({"url": self.url, "codigo": codigo.lower().replace("-", " "), "nombre": "Móvil"})   # se teclee como se teclee
        movil.enlace_sincronizar({})
        self.assertEqual(self.nombres(movil), ["Del servidor"])
        self.assertEqual([e["nombre"] for e in self.srv.acceso.enlaces()], ["USB", "Móvil"])

    def test_el_codigo_vale_una_sola_vez(self):
        codigo = self.srv.enlace_codigo()["codigo"]
        self.otro("uno").enlace_configurar({"url": self.url, "codigo": codigo})
        with self.assertRaises(ErrorEnlace) as e:
            self.otro("dos").enlace_configurar({"url": self.url, "codigo": codigo})
        self.assertIn("pide uno nuevo", str(e.exception))

    def test_pedir_otro_anula_el_anterior(self):
        viejo = self.srv.enlace_codigo()["codigo"]
        nuevo = self.srv.enlace_codigo()["codigo"]
        with self.assertRaises(ErrorEnlace):
            self.otro("uno").enlace_configurar({"url": self.url, "codigo": viejo})
        self.otro("dos").enlace_configurar({"url": self.url, "codigo": nuevo})

    def test_caduca(self):
        ahora = [1000.0]
        self.srv.acceso._reloj = lambda: ahora[0]
        codigo = self.srv.enlace_codigo()["codigo"]
        ahora[0] += 601
        with self.assertRaises(ErrorEnlace):
            self.otro().enlace_configurar({"url": self.url, "codigo": codigo})

    def test_varios_fallos_lo_anulan_aunque_luego_se_acierte(self):
        ahora = [1000.0]
        self.srv.acceso._reloj = lambda: ahora[0]
        codigo = self.srv.enlace_codigo()["codigo"]
        for n in range(5):
            ahora[0] += 400 if n else 0                 # dejando pasar la espera que imponen los fallos... sin caducar
            self.srv.acceso._codigo["caduca"] = ahora[0] + 600
            with self.assertRaises(ErrorEnlace):
                self.otro(f"malo{n}").enlace_configurar({"url": self.url, "codigo": "AAAAA-AAAAA"})
        ahora[0] += 400
        with self.assertRaises(ErrorEnlace):
            self.otro("bueno").enlace_configurar({"url": self.url, "codigo": codigo})

    def test_en_el_disco_no_queda_rastro_del_codigo(self):
        codigo = self.srv.enlace_codigo()["codigo"].replace("-", "")
        for f in (Path(self.tmp.name) / "servidor").rglob("*"):
            if f.is_file():
                self.assertNotIn(codigo.encode(), f.read_bytes(), f.name)

    def test_con_el_servidor_bloqueado_no_hay_codigo_que_valga(self):
        codigo = self.srv.enlace_codigo()["codigo"]
        self.srv.llave = None                              # como tras reiniciar con los datos cifrados, sin que nadie haya entrado
        with self.assertRaises(ErrorEnlace):
            self.otro().enlace_configurar({"url": self.url, "codigo": codigo})

    def test_sin_la_cabecera_de_caudal_no_se_atiende(self):
        import requests
        codigo = self.srv.enlace_codigo()["codigo"]
        r = requests.post(f"{self.url}/api/enlace/canje", json={"codigo": codigo}, timeout=10)
        self.assertEqual(r.status_code, 400)
        self.otro().enlace_configurar({"url": self.url, "codigo": codigo})      # y el código sigue valiendo

    def test_el_qr_lleva_direccion_y_codigo(self):
        filas = self.srv.dibujar_qr({"texto": f"caudal://enlazar?u={self.url}&c=ABCDE-FGHJK"})["filas"]
        self.assertEqual({len(f) for f in filas}, {len(filas)})
        self.assertEqual(len(filas) % 4, 1)                 # 17 + 4·versión


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestSincronizar(DosCaudales):
    # ── lo normal ──
    def test_enlazar_no_guarda_la_contraseña_y_la_credencial_es_propia(self):
        guardado = (Path(self.tmp.name) / "usb" / "enlace.json").read_text()
        self.assertNotIn(CLAVE, guardado)
        self.assertEqual([e["nombre"] for e in self.srv.acceso.enlaces()], ["USB"])
        self.assertEqual(self.usb.estado()["enlace"]["url"], self.url)
        self.assertEqual(len(self.srv.estado()["equipos"]), 1)

    def test_con_la_contraseña_equivocada_no_se_enlaza(self):
        otro = servidor.App(Cartera(Path(self.tmp.name) / "otro" / "datos"))
        self.addCleanup(otro.cartera.cerrar)
        with self.assertRaises(enlace.ErrorEnlace):
            otro.enlace_configurar({"url": self.url, "clave": "no es esta nunca"})
        self.assertEqual(otro.estado()["enlace"], {})

    def test_la_primera_vez_sube_todo_y_simular_no_cambia_nada(self):
        self.fondo(self.usb, "Mi fondo")
        s = self.sincronizar(simular=True)
        self.assertTrue(s["simulacro"])
        self.assertEqual((s["resumen"]["llegan_aqui"], s["resumen"]["llegan_alli"]), (2, 0))
        self.assertEqual(self.nombres(self.srv), [])                           # el simulacro no toca nada
        r = self.sincronizar()
        self.assertEqual(self.nombres(self.srv), ["Mi fondo"])
        self.assertEqual(len(self.srv.cartera.datos.vivos("movimientos")), 1)
        self.iguales()
        self.assertEqual(self.sincronizar()["resumen"]["llegan_aqui"], 0)      # repetir no hace nada

    def test_lo_apuntado_en_el_servidor_llega_al_abrir_el_otro(self):
        self.fondo(self.usb, "De antes")
        self.sincronizar()
        self.fondo(self.srv, "Nuevo en el servidor")
        self.srv.cartera.guardar_movimiento({"activo": self.srv.cartera.datos.vivos("activos")[0]["id"], "tipo": "Compra",
                                             "fecha": "2025-06-01", "cantidad": 1, "precio": 110})
        r = self.sincronizar()["resumen"]
        self.assertEqual((r["llegan_alli"], r["llegan_aqui"]), (3, 0))
        self.assertEqual(self.nombres(self.usb), ["De antes", "Nuevo en el servidor"])
        self.iguales()
        # y antes de cambiar lo suyo guardó una copia local
        self.assertTrue(list((Path(self.tmp.name) / "usb" / "backups").glob("*.tar.gz")))

    def test_lo_apuntado_en_los_dos_lados_acaba_en_los_dos(self):
        a, _ = self.fondo(self.usb, "Común")
        self.sincronizar()
        self.usb.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-03-01", "cantidad": 2, "precio": 100})
        self.srv.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-04-01", "cantidad": 3, "precio": 100})
        self.sincronizar()
        for app in (self.usb, self.srv):
            self.assertEqual(sorted(m["cantidad"] for m in app.cartera.datos.vivos("movimientos")), [2, 3, 10])
        self.iguales()

    def test_lo_borrado_en_un_lado_no_resucita(self):
        a, m = self.fondo(self.usb, "Común")
        self.sincronizar()
        self.srv.cartera.borrar_movimiento(m["id"])
        self.sincronizar()
        self.assertEqual(self.usb.cartera.datos.vivos("movimientos"), [])
        self.sincronizar()
        self.assertEqual(self.srv.cartera.datos.vivos("movimientos"), [])
        self.iguales()

    def test_los_ajustes_y_los_justificantes_viajan_en_los_dos_sentidos(self):
        _, m = self.fondo(self.usb, "Con factura")
        self.usb.cartera.adjuntar(m["id"], "factura.pdf", b"%PDF del usb")
        self.usb.fiscal_criterios({"opciones": {"territorio": "navarra"}})
        r = self.sincronizar()
        self.assertEqual(r["justificantes"]["subidos"], 1)
        self.assertEqual(self.srv.estado()["adjuntos_faltan"], [])
        self.assertEqual(self.srv._ajustes()["fiscal"]["territorio"], "navarra")
        _, m2 = self.fondo(self.srv, "Del servidor")
        self.srv.cartera.adjuntar(m2["id"], "otra.pdf", b"%PDF del servidor")
        self.assertEqual(self.sincronizar()["justificantes"]["bajados"], 1)
        self.assertEqual(self.usb.estado()["adjuntos_faltan"], [])

    # ── lo que podría romper la mezcla ──
    def test_dos_inversiones_distintas_con_el_mismo_identificador_no_se_pisan(self):
        self.fondo(self.usb, "Común")
        self.sincronizar()
        del_usb, _ = self.fondo(self.usb, "Solo en el USB")
        del_srv, _ = self.fondo(self.srv, "Solo en el servidor")
        self.assertEqual(del_usb["id"], del_srv["id"])                         # chocan: las dos son «F2»
        s = self.sincronizar(simular=True)["resumen"]
        self.assertEqual([x["nombre"] for x in s["renombradas"]], ["Solo en el USB"])
        self.sincronizar()
        for app in (self.usb, self.srv):
            activos = {a["nombre"]: a["id"] for a in app.cartera.datos.vivos("activos")}
            self.assertEqual(sorted(activos), ["Común", "Solo en el USB", "Solo en el servidor"])
            self.assertEqual(len(set(activos.values())), 3)
            # cada operación sigue a SU inversión
            por_activo = {m["activo"] for m in app.cartera.datos.vivos("movimientos")}
            self.assertEqual(por_activo, set(activos.values()))
        self.iguales()
        self.assertEqual(self.sincronizar()["resumen"]["renombradas"], [])     # y ya no vuelve a chocar

    def test_los_lugares_añadidos_en_cada_lado_se_conservan(self):
        btc = self.usb.cartera.guardar_activo({"nombre": "Bitcoin", "tipo": "Cripto", "codigo": "bitcoin", "unidad": "BTC"})
        self.sincronizar()
        self.usb.cartera.guardar_lugar(btc["id"], {"nombre": "Monedero", "clase": "monedero"})
        self.srv.cartera.guardar_lugar(btc["id"], {"nombre": "Exchange", "clase": "exchange"})
        self.sincronizar()
        for app in (self.usb, self.srv):
            self.assertEqual(sorted(x["nombre"] for x in app.cartera.activo(btc["id"])["lugares"]), ["Exchange", "Monedero"])
        self.iguales()

    def test_la_misma_linea_de_extracto_apuntada_en_los_dos_no_se_duplica(self):
        a, _ = self.fondo(self.usb, "Común")
        self.sincronizar()
        linea = {"activo": a["id"], "tipo": "Compra", "fecha": "2025-05-05", "cantidad": 4, "precio": 100, "extracto": "csv:abc123"}
        self.usb.cartera.guardar_movimiento(dict(linea))
        self.srv.cartera.guardar_movimiento(dict(linea))
        r = self.sincronizar()["resumen"]
        self.assertEqual(r["duplicadas"], 1)
        for app in (self.usb, self.srv):
            self.assertEqual(sorted(m["cantidad"] for m in app.cartera.datos.vivos("movimientos")), [4, 10])
        self.iguales()

    def test_dos_compras_iguales_apuntadas_a_mano_no_se_tocan(self):
        a, _ = self.fondo(self.usb, "Común")
        self.sincronizar()
        compra = {"activo": a["id"], "tipo": "Compra", "fecha": "2025-05-05", "cantidad": 4, "precio": 100}
        self.usb.cartera.guardar_movimiento(dict(compra))
        self.srv.cartera.guardar_movimiento(dict(compra))
        self.assertEqual(self.sincronizar()["resumen"]["duplicadas"], 0)      # sin extracto no se puede saber si son la misma
        self.assertEqual(len(self.srv.cartera.datos.vivos("movimientos")), 3)

    def test_avisa_si_son_dos_carteras_que_no_tienen_nada_que_ver(self):
        self.fondo(self.usb, "Una cartera")
        self.srv.cartera.guardar_activo({"nombre": "Otra cartera", "tipo": "Accion"})
        self.assertTrue(self.sincronizar(simular=True)["resumen"]["carteras_distintas"])

    def test_si_lo_que_llega_no_vale_no_se_mezcla_nada(self):
        self.fondo(self.srv, "Del servidor")
        antes = huella_datos(self.srv.cartera.datos)
        e = enlace.Enlace(self.usb.carpeta)
        bueno = enlace.empaquetar(self.usb.cartera.datos, {})
        for malo in ({**bueno, "version": bueno["version"] + 1},
                     {**bueno, "datos": {**bueno["datos"], "movimientos": [{"id": "m1", "activo": "F1", "tipo": "NoExiste", "fecha": "2025-01-01"}]}},
                     {**bueno, "datos": {**bueno["datos"], "activos": [{"id": "F9", "nombre": "x", "tipo": "Fondo"}] * 2}},
                     {"datos": "esto no es"}):
            with self.assertRaises(enlace.ErrorEnlace):
                e.llamar("mezclar", malo)
        self.assertEqual(huella_datos(self.srv.cartera.datos), antes)

    def test_un_equipo_olvidado_ya_no_puede_sincronizar(self):
        self.fondo(self.usb, "Mi fondo")
        self.assertEqual(self.srv.enlace_olvidar_equipos({})["olvidados"], 1)
        with self.assertRaises(enlace.ErrorEnlace):
            self.sincronizar()
        self.assertEqual(self.nombres(self.srv), [])

    def test_en_la_cartera_de_ejemplo_no_se_sincroniza(self):
        self.usb.ejemplo(True)
        self.addCleanup(lambda: self.usb.ejemplo(False))
        with self.assertRaises(ErrorOperacion):
            self.sincronizar()

    def test_con_el_servidor_cifrado_y_recien_reiniciado_tambien_sincroniza(self):
        self.fondo(self.usb, "Mi fondo")
        self.sincronizar()
        self.srv.cifrado_poner({"activar": True, "clave": CLAVE})
        self.fondo(self.srv, "Nuevo en el servidor")
        # se apaga y se enciende: los datos del servidor quedan cerrados
        puerto = self.srv.puerto
        self.apagar()
        almacen.poner_llave(None)
        self.srv = servidor.arrancar(Path(self.tmp.name) / "servidor" / "datos", puerto, "127.0.0.1")
        threading.Thread(target=self.srv.servidor.serve_forever, daemon=True).start()
        self.assertIsNone(self.srv.cartera)
        self.sincronizar()                                                     # la credencial del equipo los abre
        self.assertEqual(self.nombres(self.usb), ["Mi fondo", "Nuevo en el servidor"])
        self.assertTrue(almacen.esta_cifrado((Path(self.tmp.name) / "servidor" / "datos" / "activos.jsonl").read_bytes()))
        # (Que en el otro equipo sigan en claro no se puede comprobar aquí: en esta prueba los dos
        # Caudal van en el mismo proceso y comparten la llave en memoria; de verdad son dos máquinas.)


class TorFalso(threading.Thread):
    """Un SOCKS5 de mentira: lleva cualquier dirección .onion al servidor de la prueba."""

    def __init__(self, puerto_destino: int):
        super().__init__(daemon=True)
        self.puerto_destino = puerto_destino
        self.pedidas: list[str] = []
        self.sock = socket.socket()
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(8)
        self.puerto = self.sock.getsockname()[1]

    def run(self):
        while True:
            try:
                cliente, _ = self.sock.accept()
            except OSError:
                return
            threading.Thread(target=self._atender, args=(cliente,), daemon=True).start()

    def parar(self):
        self.sock.close()

    def _atender(self, cliente: socket.socket):
        try:
            cliente.recv(3)
            cliente.sendall(b"\x05\x00")
            cabecera = cliente.recv(5)
            nombre = cliente.recv(cabecera[4]).decode()
            cliente.recv(2)
            self.pedidas.append(nombre)
            if not nombre.endswith(".onion") or nombre.startswith("apagada"):
                cliente.sendall(b"\x05\x04\x00\x01" + bytes(6))          # destino inalcanzable
                return
            destino = socket.create_connection(("127.0.0.1", self.puerto_destino))
            cliente.sendall(b"\x05\x00\x00\x01" + bytes(6))

            def pasar(de, a):
                try:
                    while True:
                        trozo = de.recv(65536)
                        if not trozo:
                            break
                        a.sendall(trozo)
                except OSError:
                    pass
                finally:
                    for s in (de, a):
                        try:
                            s.shutdown(socket.SHUT_RDWR)
                        except OSError:
                            pass
            hilo = threading.Thread(target=pasar, args=(destino, cliente), daemon=True)
            hilo.start()
            pasar(cliente, destino)
            hilo.join(5)
            destino.close()
        except (OSError, IndexError):
            pass
        finally:
            cliente.close()


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestFueraDeCasa(DosCaudales):
    """Con una dirección .onion de reserva: se usa cuando la de casa no responde."""

    ONION = "http://caudaldeprueba234567caudaldeprueba234567caudaldeprueba23.onion"

    def setUp(self):
        super().setUp()
        self.tor = TorFalso(self.srv.puerto)
        self.tor.start()
        self.addCleanup(self.tor.parar)
        config = btc.cargar(self.usb.cartera.carpeta)
        config["proxy"] = {"host": "127.0.0.1", "puerto": self.tor.puerto}
        btc.guardar(self.usb.cartera.carpeta, config)
        self.e = enlace.Enlace(self.usb.cartera.carpeta)

    def fuera_de_casa(self):
        """Como llevarse el USB: la dirección de casa deja de responder."""
        cerrado = socket.socket()
        cerrado.bind(("127.0.0.1", 0))
        puerto = cerrado.getsockname()[1]
        cerrado.close()
        self.e.guardar({**self.e.config(), "url": f"http://127.0.0.1:{puerto}"})

    def test_al_ponerla_se_prueba_y_solo_entonces_se_guarda(self):
        r = self.usb.enlace_configurar({"url_tor": self.ONION.replace("http://", "")})     # sin «http://» también vale
        self.assertEqual(r["url_tor"], self.ONION)
        self.assertEqual(self.usb.estado()["enlace"]["url_tor"], self.ONION)
        self.assertEqual(self.tor.pedidas, [self.ONION.split("//")[1]])
        with self.assertRaises(enlace.ErrorEnlace):
            self.usb.enlace_configurar({"url_tor": "http://apagada234567.onion"})         # no responde: no se guarda
        with self.assertRaises(enlace.ErrorEnlace):
            self.usb.enlace_configurar({"url_tor": "https://ejemplo.com"})                # no es una .onion
        self.assertEqual(self.e.config()["url_tor"], self.ONION)
        self.usb.enlace_configurar({"url_tor": ""})
        self.assertNotIn("url_tor", self.e.config())

    def test_en_casa_no_se_toca_tor(self):
        self.usb.enlace_configurar({"url_tor": self.ONION})
        self.tor.pedidas.clear()
        self.fondo(self.usb, "Mi fondo")
        self.sincronizar()
        self.assertEqual(self.tor.pedidas, [])
        self.iguales()

    def test_fuera_de_casa_se_sincroniza_por_tor(self):
        self.usb.enlace_configurar({"url_tor": self.ONION})
        self.fondo(self.usb, "Apuntado de viaje")
        self.fondo(self.srv, "Apuntado en casa")
        self.fuera_de_casa()
        self.tor.pedidas.clear()
        self.assertTrue(self.sincronizar(simular=True)["simulacro"])
        self.sincronizar()
        self.assertTrue(self.tor.pedidas)
        self.assertEqual(self.nombres(self.srv), ["Apuntado de viaje", "Apuntado en casa"])
        self.iguales()

    def test_sin_direccion_de_reserva_fuera_de_casa_lo_dice(self):
        self.fuera_de_casa()
        with self.assertRaisesRegex(enlace.ErrorEnlace, "No llego al servidor"):
            self.sincronizar()

    def test_si_no_se_llega_por_ninguna_lo_dice_y_no_cambia_nada(self):
        self.usb.enlace_configurar({"url_tor": self.ONION})
        self.fondo(self.usb, "Mi fondo")
        self.fuera_de_casa()
        self.e.guardar({**self.e.config(), "url_tor": "http://apagada234567.onion"})
        with self.assertRaisesRegex(enlace.ErrorEnlace, "ni por la dirección de casa ni por la .onion"):
            self.sincronizar()
        self.assertEqual(self.nombres(self.srv), [])

    def test_un_servidor_anterior_que_no_conoce_la_prueba_tambien_vale(self):
        de_verdad = enlace.atender
        enlace.atender = lambda app, orden, cuerpo, prefijos: (_ for _ in ()).throw(enlace.ErrorEnlace(enlace.NO_ENTIENDO)) \
            if orden == "hola" else de_verdad(app, orden, cuerpo, prefijos)
        self.addCleanup(setattr, enlace, "atender", de_verdad)
        self.assertEqual(self.usb.enlace_configurar({"url_tor": self.ONION})["url_tor"], self.ONION)

    def test_tambien_se_puede_enlazar_directamente_por_la_onion(self):
        self.usb.enlace_configurar({"desenlazar": True})
        self.usb.enlace_configurar({"url": self.ONION, "clave": CLAVE, "nombre": "USB de viaje"})
        self.fondo(self.usb, "Mi fondo")
        self.sincronizar()
        self.assertEqual(self.nombres(self.srv), ["Mi fondo"])

    def _puerto_cerrado(self):
        cerrado = socket.socket()
        cerrado.bind(("127.0.0.1", 0))
        puerto = cerrado.getsockname()[1]
        cerrado.close()
        return puerto

    def test_el_qr_con_onion_en_casa_enlaza_por_casa_y_la_deja_de_reserva(self):
        self.usb.enlace_configurar({"desenlazar": True})
        self.tor.pedidas.clear()
        codigo = self.srv.enlace_codigo()["codigo"]
        r = self.usb.enlace_configurar({"url": self.url, "url_tor": self.ONION.replace("http://", ""), "codigo": codigo, "nombre": "Móvil"})
        self.assertEqual(r["url_tor"], self.ONION)
        self.assertEqual(self.tor.pedidas, [])                       # en casa no se toca Tor
        self.assertEqual(self.e.config()["url_tor"], self.ONION)
        self.fuera_de_casa()
        self.fondo(self.usb, "De viaje")
        self.sincronizar()                                           # y fuera de casa ya va por la .onion
        self.assertEqual(self.nombres(self.srv), ["De viaje"])

    def test_el_qr_leido_fuera_de_casa_enlaza_por_la_onion(self):
        self.usb.enlace_configurar({"desenlazar": True})
        codigo = self.srv.enlace_codigo()["codigo"]
        casa = f"http://127.0.0.1:{self._puerto_cerrado()}"
        self.usb.enlace_configurar({"url": casa, "url_tor": self.ONION, "codigo": codigo, "nombre": "Móvil"})
        self.assertTrue(self.tor.pedidas)
        self.assertEqual(self.e.config()["url"], casa)
        self.fondo(self.usb, "Mi fondo")
        self.sincronizar()
        self.assertEqual(self.nombres(self.srv), ["Mi fondo"])

    def test_si_no_llega_por_ninguna_no_queda_enlazado(self):
        self.usb.enlace_configurar({"desenlazar": True})
        codigo = self.srv.enlace_codigo()["codigo"]
        with self.assertRaisesRegex(enlace.ErrorEnlace, "ni por la dirección de casa ni por la .onion"):
            self.usb.enlace_configurar({"url": f"http://127.0.0.1:{self._puerto_cerrado()}", "url_tor": "http://apagada234567.onion",
                                        "codigo": codigo})
        self.assertEqual(self.e.config(), {})
        self.otro_con(codigo)                                        # el código no se ha gastado

    def otro_con(self, codigo):
        app = servidor.App(Cartera(Path(self.tmp.name) / "otro" / "datos"))
        self.addCleanup(app.cartera.cerrar)
        app.enlace_configurar({"url": self.url, "codigo": codigo})

    def test_una_onion_que_no_es_onion_no_se_acepta_al_enlazar(self):
        self.usb.enlace_configurar({"desenlazar": True})
        with self.assertRaisesRegex(enlace.ErrorEnlace, "dirección .onion"):
            self.usb.enlace_configurar({"url": self.url, "url_tor": "https://ejemplo.com", "clave": CLAVE})

    def test_con_la_credencial_retirada_la_onion_no_se_guarda(self):
        self.srv.enlace_olvidar_equipos({})
        with self.assertRaises(enlace.ErrorEnlace):
            self.usb.enlace_configurar({"url_tor": self.ONION})
        self.assertNotIn("url_tor", self.e.config())


def _certificado(carpeta: Path, nombre: str) -> tuple[str, str]:
    """Un certificado firmado por sí mismo, como autoridad y como servidor a la vez."""
    import datetime as dt
    from cryptography import x509
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import ec
    clave = ec.generate_private_key(ec.SECP256R1())
    quien = x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, nombre)])
    ahora = dt.datetime.now(dt.timezone.utc)
    cert = (x509.CertificateBuilder().subject_name(quien).issuer_name(quien).public_key(clave.public_key())
            .serial_number(x509.random_serial_number()).not_valid_before(ahora - dt.timedelta(days=1))
            .not_valid_after(ahora + dt.timedelta(days=2))
            .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
            .sign(clave, hashes.SHA256()))
    ruta_cert, ruta_clave = carpeta / f"{nombre}.crt", carpeta / f"{nombre}.key"
    ruta_cert.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    ruta_clave.write_bytes(clave.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
                                               serialization.NoEncryption()))
    return str(ruta_cert), str(ruta_clave)


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestCertificadoElegidoComoFichero(DosCaudales):
    """En el teléfono no hay una ruta que escribir: el certificado raíz llega con su contenido."""

    def otro(self):
        app = servidor.App(Cartera(Path(self.tmp.name) / "movil" / "datos"))
        self.addCleanup(app.cartera.cerrar)
        return app

    def test_se_guarda_junto_a_los_datos_y_se_usa(self):
        ruta, _ = _certificado(Path(self.tmp.name), "startos")
        pem = Path(ruta).read_text()
        movil = self.otro()
        movil.enlace_configurar({"url": self.url, "codigo": self.srv.enlace_codigo()["codigo"], "certificado_pem": pem})
        e = enlace.Enlace(movil.cartera.carpeta)
        guardado = Path(e.config()["certificado"])
        self.assertEqual(guardado.name, enlace.FICHERO_CERTIFICADO)
        self.assertEqual(guardado.parent, e.ruta.parent)
        self.assertEqual(guardado.read_text().strip(), pem.strip())
        movil.enlace_configurar({"desenlazar": True})          # al desenlazar se va con lo demás
        self.assertFalse(guardado.exists())

    def test_lo_que_no_es_un_certificado_no_se_acepta(self):
        movil = self.otro()
        for malo in ("hola", "-----BEGIN CERTIFICATE-----\nnada\n-----END CERTIFICATE-----"):
            with self.assertRaises(ErrorEnlace):
                movil.enlace_configurar({"url": self.url, "codigo": "ABCDE-FGHJK", "certificado_pem": malo})
        self.assertFalse((enlace.Enlace(movil.cartera.carpeta).ruta.parent / enlace.FICHERO_CERTIFICADO).exists())


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestOnionConHttps(unittest.TestCase):
    """Una .onion con https, como las de StartOS: un mismo puerto reparte varios
    servicios, así que hay que decir en el saludo a cuál se va (SNI)."""

    def setUp(self):
        import http.server
        import json
        import ssl
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        carpeta = Path(self.tmp.name)
        self.suya, clave = _certificado(carpeta, "servidor")
        self.otra, _ = _certificado(carpeta, "otra-autoridad")
        visto = self.visto = {}

        class Manejador(http.server.BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_POST(self):
                visto["host"] = self.headers.get("Host")
                self.rfile.read(int(self.headers["Content-Length"]))
                cuerpo = json.dumps({"ok": True}).encode()
                self.send_response(200)
                self.send_header("Content-Length", str(len(cuerpo)))
                self.send_header("Set-Cookie", "sesion=abc; HttpOnly")
                self.end_headers()
                self.wfile.write(cuerpo)

        self.https = http.server.HTTPServer(("127.0.0.1", 0), Manejador)
        contexto = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        contexto.load_cert_chain(self.suya, clave)
        contexto.sni_callback = lambda s, nombre, c: visto.__setitem__("sni", nombre)
        self.https.socket = contexto.wrap_socket(self.https.socket, server_side=True)
        threading.Thread(target=self.https.serve_forever, daemon=True).start()
        self.addCleanup(self.https.server_close)
        self.addCleanup(self.https.shutdown)
        self.tor = TorFalso(self.https.server_address[1])
        self.tor.start()
        self.addCleanup(self.tor.parar)

    def pedir(self, certificado: str):
        sesion = enlace._SesionTor(("127.0.0.1", self.tor.puerto), certificado)
        r = sesion.post("https://caudaldeprueba234567.onion/api/enlace", json={"orden": "hola"}, headers={}, timeout=(8, 30))
        return r, sesion

    def test_dice_a_que_servicio_va_y_guarda_la_sesion(self):
        r, sesion = self.pedir(self.suya)
        self.assertEqual((r.status_code, r.json()), (200, {"ok": True}))
        self.assertEqual(self.visto, {"sni": "caudaldeprueba234567.onion", "host": "caudaldeprueba234567.onion"})
        self.assertEqual(sesion.cookies, {"sesion": "abc"})

    def test_sin_certificado_raiz_tambien_va_porque_tor_ya_garantiza_el_destino(self):
        self.assertEqual(self.pedir("")[0].status_code, 200)

    def test_con_el_certificado_de_otra_autoridad_no_sigue(self):
        with self.assertRaisesRegex(enlace.ErrorEnlace, "no es de la autoridad"):
            self.pedir(self.otra)


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


class TestDireccionesDelQR(unittest.TestCase):
    """Las direcciones que lleva el QR de enlazar: la de casa (nunca un .local si hay otra) y la .onion."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.datos = Path(self.tmp.name) / "datos"
        self.datos.mkdir()
        viejo = os.environ.pop(enlace.VARIABLE_DIRECCIONES, None)
        self.addCleanup(lambda: os.environ.__setitem__(enlace.VARIABLE_DIRECCIONES, viejo) if viejo is not None
                        else os.environ.pop(enlace.VARIABLE_DIRECCIONES, None))

    def test_de_las_de_startos_elige_la_ip_de_casa_y_la_onion_sin_cifrar(self):
        urls = ["https://servidor.local:50717", "http://servidor.local:50716", "https://10.8.0.5:50717",
                "https://192.168.1.20:50717", "http://192.168.1.20:50716", "https://[fe80::1]:50717",
                "https://abcdefghijklmnop.onion", "http://abcdefghijklmnop.onion"]
        self.assertEqual(enlace.elegir_direcciones(urls), {"casa": "https://192.168.1.20:50717", "tor": "http://abcdefghijklmnop.onion"})

    def test_un_local_solo_si_no_hay_otra(self):
        self.assertEqual(enlace.elegir_direcciones(["https://caudal.local:8443"])["casa"], "https://caudal.local:8443")
        self.assertEqual(enlace.elegir_direcciones(["https://caudal.local:8443", "https://caudal.midominio.es"])["casa"],
                         "https://caudal.midominio.es")
        self.assertEqual(enlace.elegir_direcciones([]), {"casa": "", "tor": ""})
        self.assertEqual(enlace.elegir_direcciones(["http://127.0.0.1:8080", "https://8.8.8.8"]), {"casa": "", "tor": ""})

    def test_las_del_sistema_llegan_por_la_variable(self):
        os.environ[enlace.VARIABLE_DIRECCIONES] = '["https://192.168.1.50:8443", "http://xyz.onion"]'
        d = enlace.direcciones_de_enlace(self.datos)
        self.assertEqual((d["casa"], d["tor"]), ("https://192.168.1.50:8443", "http://xyz.onion"))
        os.environ[enlace.VARIABLE_DIRECCIONES] = "esto no es json"
        self.assertEqual(enlace.direcciones_de_enlace(self.datos)["casa"], "")

    def test_lo_escrito_a_mano_manda_y_vacio_vuelve_a_las_del_sistema(self):
        os.environ[enlace.VARIABLE_DIRECCIONES] = '["https://192.168.1.50:8443", "http://xyz.onion"]'
        d = enlace.poner_direcciones_de_enlace(self.datos, "192.168.1.60:8080", "")
        self.assertEqual((d["casa"], d["tor"]), ("http://192.168.1.60:8080", "http://xyz.onion"))
        d = enlace.poner_direcciones_de_enlace(self.datos, "", "")
        self.assertEqual(d["casa"], "https://192.168.1.50:8443")
        self.assertFalse((self.datos.parent / enlace.FICHERO_DIRECCIONES).exists())

    def test_no_acepta_un_local_ni_una_onion_falsa(self):
        with self.assertRaisesRegex(enlace.ErrorEnlace, ".local"):
            enlace.poner_direcciones_de_enlace(self.datos, "https://umbrel.local:3000", "")
        with self.assertRaisesRegex(enlace.ErrorEnlace, ".onion"):
            enlace.poner_direcciones_de_enlace(self.datos, "", "https://ejemplo.com")
        with self.assertRaisesRegex(enlace.ErrorEnlace, "no puede ser una .onion"):
            enlace.poner_direcciones_de_enlace(self.datos, "http://xyz.onion", "")


@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
class TestDireccionesEnElServidor(DosCaudales):

    def test_el_codigo_viene_con_las_direcciones_y_se_pueden_cambiar(self):
        self.assertEqual(self.srv.enlace_codigo()["direcciones"]["casa"], "")
        self.srv.enlace_direcciones({"casa": "https://192.168.1.50:8443", "tor": "xyz.onion"})
        d = self.srv.enlace_codigo()["direcciones"]
        self.assertEqual((d["casa"], d["tor"]), ("https://192.168.1.50:8443", "http://xyz.onion"))
        with self.assertRaises(ErrorOperacion):
            self.usb.enlace_direcciones({})                         # solo en un servidor