"""
Pruebas de la entrada con contraseña (cartera/acceso.py y su parte del
servidor): solo cuando Caudal va en un servidor; en tu equipo sigue igual.

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

from __future__ import annotations

import http.client
import json
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 acceso as modulo  # noqa: E402
from cartera import servidor  # noqa: E402
from cartera.acceso import Acceso, ErrorAcceso, Espera  # noqa: E402

try:
    import cryptography  # noqa: F401
    HAY_CIFRADO = True
except ImportError:
    HAY_CIFRADO = False
SIN_CIFRADO = "falta cryptography: la contraseña guarda una llave y sin él no se puede crear"

CLAVE = "una frase que recuerdo"


@unittest.skipUnless(HAY_CIFRADO, SIN_CIFRADO)
class TestContraseña(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.ahora = 1000.0
        self.a = Acceso(Path(self.tmp.name), reloj=lambda: self.ahora)

    def test_se_crea_una_vez_y_no_se_guarda_en_claro(self):
        self.assertFalse(self.a.configurado())
        self.a.crear(CLAVE)
        self.assertTrue(self.a.configurado())
        guardado = self.a.ruta.read_text(encoding="utf-8")
        self.assertNotIn(CLAVE, guardado)
        self.assertEqual(self.a.ruta.stat().st_mode & 0o777, 0o600)
        with self.assertRaises(ErrorAcceso):
            self.a.crear("otra contraseña distinta")            # no se pisa
        self.a.comprobar(CLAVE)

    def test_una_contraseña_corta_no_vale(self):
        with self.assertRaises(ErrorAcceso):
            self.a.crear("corta")
        self.assertFalse(self.a.configurado())

    def test_la_equivocada_no_entra_y_tras_varios_fallos_hay_que_esperar(self):
        self.a.crear(CLAVE)
        for _ in range(modulo.FALLOS_LIBRES):
            with self.assertRaises(ErrorAcceso) as e:
                self.a.comprobar("no es esta")
            self.assertNotIsInstance(e.exception, Espera)
        with self.assertRaises(Espera):
            self.a.comprobar(CLAVE)                             # ni la buena, hasta que pase la espera
        self.ahora += 2
        self.a.comprobar(CLAVE)
        with self.assertRaises(ErrorAcceso) as e:               # acertar pone el contador a cero
            self.a.comprobar("no es esta")
        self.assertNotIsInstance(e.exception, Espera)

    def test_la_espera_se_dobla_pero_tiene_tope(self):
        self.a.crear(CLAVE)
        for _ in range(modulo.FALLOS_LIBRES + 30):
            self.ahora += modulo.ESPERA_MAXIMA + 1
            with self.assertRaises(ErrorAcceso):
                self.a.comprobar("no es esta")
        self.ahora += modulo.ESPERA_MAXIMA + 1
        self.a.comprobar(CLAVE)

    def test_cambiarla_cierra_las_sesiones_abiertas(self):
        self.a.crear(CLAVE)
        sid, _ = self.a.abrir()
        with self.assertRaises(ErrorAcceso):
            self.a.cambiar("no es esta", "otra contraseña nueva")
        self.a.cambiar(CLAVE, "otra contraseña nueva")
        self.assertIsNone(self.a.sesion(sid))
        self.a.comprobar("otra contraseña nueva")

    def test_si_se_olvida_se_restablece_sin_tocar_nada_mas(self):
        self.a.crear(CLAVE)
        otro = Path(self.tmp.name) / "datos.txt"
        otro.write_text("mis datos")
        nueva = self.a.restablecer()
        self.a.comprobar(nueva)
        with self.assertRaises(ErrorAcceso):
            self.a.comprobar(CLAVE)
        self.assertEqual(otro.read_text(), "mis datos")

    def test_la_contraseña_guarda_una_llave_que_sobrevive_al_cambio_pero_no_al_olvido(self):
        llave = self.a.crear(CLAVE)
        self.assertEqual(len(llave), 32)
        self.assertEqual(self.a.comprobar(CLAVE), llave)
        self.assertNotIn(llave.hex(), self.a.ruta.read_text(encoding="utf-8"))
        self.a.cambiar(CLAVE, "otra contraseña nueva")
        self.assertEqual(self.a.comprobar("otra contraseña nueva"), llave)
        nueva = self.a.restablecer()
        self.assertNotEqual(self.a.comprobar(nueva), llave)                 # sin la vieja no se abre su sobre

    def test_tras_cambiarla_falta_una_copia_nueva_hasta_que_se_hace(self):
        self.a.crear(CLAVE)
        self.assertFalse(self.a.copia_pendiente())               # recién creada no hay copias viejas que valgan menos
        self.a.anotar_copia()
        mia, _ = self.a.abrir()
        otra, _ = self.a.abrir()
        self.a.cambiar(CLAVE, "otra contraseña nueva", conservar=mia)
        self.assertTrue(self.a.copia_pendiente())
        self.assertIsNotNone(self.a.sesion(mia))                 # quien la cambia sigue dentro
        self.assertIsNone(self.a.sesion(otra))                   # los demás, fuera
        self.a.anotar_copia()
        self.assertFalse(self.a.copia_pendiente())
        self.a.restablecer()
        self.assertTrue(self.a.copia_pendiente())

    def test_una_sesion_caduca(self):
        self.a.crear(CLAVE)
        sid, token = self.a.abrir()
        self.assertEqual(self.a.sesion(sid)["token"], token)
        self.ahora += modulo.DURACION + 1
        self.assertIsNone(self.a.sesion(sid))
        self.assertIsNone(self.a.sesion("inventada"))


@unittest.skipUnless(HAY_CIFRADO, SIN_CIFRADO)
class TestServidorConContraseña(unittest.TestCase):
    """Peticiones de verdad contra Caudal arrancado como en un servidor."""

    @classmethod
    def setUpClass(cls):
        cls._tmp = tempfile.TemporaryDirectory()
        cls.app = servidor.arrancar(Path(cls._tmp.name) / "datos", 0, "127.0.0.1")
        threading.Thread(target=cls.app.servidor.serve_forever, daemon=True).start()

    @classmethod
    def tearDownClass(cls):
        cls.app.servidor.shutdown()
        cls.app.servidor.server_close()
        cls.app.cartera.cerrar()
        cls._tmp.cleanup()

    def pedir(self, metodo, ruta, cuerpo=None, **cabeceras):
        c = http.client.HTTPConnection("127.0.0.1", self.app.puerto, timeout=10)
        if cuerpo is not None:
            cabeceras.setdefault("Content-Type", "application/json")
        c.request(metodo, ruta, json.dumps(cuerpo) if cuerpo is not None else None, cabeceras)
        r = c.getresponse()
        texto = r.read().decode()
        c.close()
        try:
            datos = json.loads(texto)
        except ValueError:
            datos = texto
        return r.status, datos, r.getheader("Set-Cookie") or ""

    def test_1_sin_sesion_solo_se_ve_la_puerta(self):
        estado, pagina, _ = self.pedir("GET", "/")
        self.assertEqual(estado, 200)
        self.assertIn('name="token" content=""', pagina)         # la página no lleva nada privado
        estado, datos, _ = self.pedir("GET", "/api/estado")
        self.assertEqual((estado, datos.get("sin_sesion")), (401, True))
        self.assertEqual(self.pedir("POST", "/api/movimiento", {"x": 1})[0], 401)
        estado, datos, _ = self.pedir("GET", "/api/acceso")
        self.assertEqual((estado, datos["configurado"], datos["dentro"], datos["token"]), (200, False, False, None))

    def test_2_crear_la_contraseña_y_entrar(self):
        # Sin la cabecera propia de la aplicación, ni se intenta (otra web no puede ponerla)
        self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": CLAVE})[0], 403)
        self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": "corta"}, **{"X-Caudal": "1"})[0], 400)
        estado, datos, cookie = self.pedir("POST", "/api/acceso/crear", {"clave": CLAVE}, **{"X-Caudal": "1"})
        self.assertEqual(estado, 200)
        self.assertIn("HttpOnly", cookie)
        self.assertIn("SameSite=Strict", cookie)
        type(self).cookie, type(self).token = cookie.split(";")[0], datos["token"]
        # Con la cookie y el token, dentro; con solo la cookie, no
        self.assertEqual(self.pedir("GET", "/api/estado", Cookie=self.cookie, **{"X-Token": self.token})[0], 200)
        self.assertEqual(self.pedir("GET", "/api/estado", Cookie=self.cookie)[0], 401)
        self.assertEqual(self.pedir("GET", "/api/estado", **{"X-Token": self.token})[0], 401)
        estado, datos, _ = self.pedir("GET", "/api/acceso", Cookie=self.cookie)
        self.assertEqual((datos["configurado"], datos["dentro"], datos["token"]), (True, True, self.token))
        self.assertTrue(self.pedir("GET", "/api/estado", Cookie=self.cookie, **{"X-Token": self.token})[1]["servidor"])

    def test_3_no_se_puede_crear_otra_ni_entrar_con_la_equivocada(self):
        self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": "otra distinta más"}, **{"X-Caudal": "1"})[0], 400)
        estado, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": "no es esta"}, **{"X-Caudal": "1"})
        self.assertEqual((estado, cookie), (400, ""))
        estado, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": CLAVE}, **{"X-Caudal": "1"})
        self.assertEqual(estado, 200)
        self.assertNotEqual(datos["token"], self.token)          # cada sesión, su token

    def test_4_salir_cierra_la_sesion_pero_no_apaga_caudal(self):
        cab = {"Cookie": self.cookie, "X-Token": self.token}
        self.assertEqual(self.pedir("POST", "/api/salir", {}, **cab)[0], 200)        # no apaga nada
        self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 200)
        self.assertEqual(self.pedir("POST", "/api/acceso/salir", {}, Cookie=self.cookie, **{"X-Caudal": "1"})[0], 200)
        self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 401)

    def test_5_cambiar_la_contraseña_desde_dentro(self):
        _, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": CLAVE}, **{"X-Caudal": "1"})
        cab = {"Cookie": cookie.split(";")[0], "X-Token": datos["token"]}
        self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": "no", "nueva": "nueva contraseña"}, **cab)[0], 400)
        self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": CLAVE, "nueva": "nueva contraseña"}, **cab)[0], 200)
        estado, datos, _ = self.pedir("GET", "/api/estado", **cab)                    # quien la cambia sigue dentro…
        self.assertEqual((estado, datos["copia_pendiente"]), (200, True))             # …y se le avisa de la copia
        # Un «_sesion» mandado desde fuera no sirve para conservar la sesión de otro
        self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": "nueva contraseña", "nueva": "nueva contraseña", "_sesion": "x"}, **cab)[0], 200)
        self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 200)
        self.assertEqual(self.pedir("POST", "/api/acceso/entrar", {"clave": "nueva contraseña"}, **{"X-Caudal": "1"})[0], 200)


class TestLigero(unittest.TestCase):
    """La interfaz viaja comprimida y el navegador no la vuelve a bajar si no ha cambiado."""

    @classmethod
    def setUpClass(cls):
        cls._tmp = tempfile.TemporaryDirectory()
        cls.app = servidor.arrancar(Path(cls._tmp.name) / "datos", 0, "127.0.0.1")
        threading.Thread(target=cls.app.servidor.serve_forever, daemon=True).start()

    @classmethod
    def tearDownClass(cls):
        cls.app.servidor.shutdown()
        cls.app.servidor.server_close()
        cls.app.cartera.cerrar()
        cls._tmp.cleanup()

    def pedir(self, ruta, **cabeceras):
        c = http.client.HTTPConnection("127.0.0.1", self.app.puerto, timeout=10)
        c.request("GET", ruta, headers=cabeceras)
        r = c.getresponse()
        cuerpo = r.read()
        c.close()
        return r, cuerpo

    def test_la_interfaz_se_comprime_si_el_navegador_lo_admite(self):
        import gzip
        original = (servidor.WEB / "app.js").read_bytes()
        r, cuerpo = self.pedir("/web/app.js", **{"Accept-Encoding": "gzip, deflate"})
        self.assertEqual(r.getheader("Content-Encoding"), "gzip")
        self.assertLess(len(cuerpo), len(original) / 3)
        self.assertEqual(gzip.decompress(cuerpo), original)
        r, cuerpo = self.pedir("/web/app.js")                    # y sin comprimir para quien no lo pida
        self.assertIsNone(r.getheader("Content-Encoding"))
        self.assertEqual(cuerpo, original)

    def test_lo_que_no_ha_cambiado_no_se_vuelve_a_bajar_y_los_datos_nunca_se_guardan(self):
        r, _ = self.pedir("/web/app.js")
        marca = r.getheader("ETag")
        self.assertEqual(r.getheader("Cache-Control"), "no-cache")
        r, cuerpo = self.pedir("/web/app.js", **{"If-None-Match": marca})
        self.assertEqual((r.status, cuerpo), (304, b""))
        r, _ = self.pedir("/web/app.js", **{"If-None-Match": '"otra"'})
        self.assertEqual(r.status, 200)
        r, _ = self.pedir("/api/acceso")
        self.assertEqual(r.getheader("Cache-Control"), "no-store")


class TestEnTuEquipoSigueIgual(unittest.TestCase):
    def test_sin_modo_servidor_no_hay_contraseña_ni_rutas_de_acceso(self):
        with tempfile.TemporaryDirectory() as carpeta:
            app = servidor.arrancar(Path(carpeta) / "datos", 0)
            try:
                self.assertIsNone(app.acceso)
                self.assertFalse(app.estado()["servidor"])
                self.assertFalse((Path(carpeta) / "acceso.json").exists())
                self.assertEqual(app.servidor.server_address[0], "127.0.0.1")
            finally:
                app.servidor.server_close()
                app.cartera.cerrar()


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