caudal/programa/pruebas/test_core.py

"""
Pruebas de la conexión directa al nodo (cartera/core.py): Bitcoin Core o Knots
por RPC, sin servidor Electrum. Contra un nodo de mentira (core_falso.py) que
comparte cadena con el Electrum de mentira, para comprobar que por los dos
caminos se ve exactamente lo mismo.

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

from __future__ import annotations

import socket
import sys
import threading
import time
import unittest
from pathlib import Path

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

from cartera import btc, bitcoin, core  # noqa: E402
from cartera.electrum import Cliente as ClienteElectrum, ErrorElectrum, conectar  # noqa: E402
from core_falso import CoreFalso  # noqa: E402
from nodo_falso import construir  # noqa: E402
from test_btc import AJENA, SATOSHI, XPUB_TAPROOT, ZPUB, ConNodo, script  # noqa: E402


# El aviso de repaso llega tal cual o envuelto con el nombre del monedero (historial_btc)
REPASO = (core.Repasando, btc.ErrorBitcoin)


class TestDescriptores(unittest.TestCase):
    def test_la_suma_de_control_es_la_del_nodo(self):
        # Los dos ejemplos de la documentación de Bitcoin Core y del BIP 380
        self.assertEqual(core.con_suma("raw(deadbeef)"), "raw(deadbeef)#89f8spxm")
        self.assertEqual(core.con_suma(
            "pkh([d34db33f/44'/0'/0']xpub6ERApfZwUNrhLCkDtcHTcxd75RbzS1ed54G1LkBUHQVHQKqhMkhgbmJbZRkrgZw4koxb5JaHWkY4ALHY2grBGRjaDMzQLcgJvLJuZZvRcEL/1/*)"),
            "pkh([d34db33f/44'/0'/0']xpub6ERApfZwUNrhLCkDtcHTcxd75RbzS1ed54G1LkBUHQVHQKqhMkhgbmJbZRkrgZw4koxb5JaHWkY4ALHY2grBGRjaDMzQLcgJvLJuZZvRcEL/1/*)#ml40v0wf")

    def test_una_zpub_se_le_da_al_nodo_escrita_como_xpub(self):
        estandar = core.clave_estandar(ZPUB)
        self.assertTrue(estandar.startswith("xpub"))
        a, b = bitcoin.ClaveExtendida(ZPUB), bitcoin.ClaveExtendida(estandar)
        self.assertEqual((a.clave, a.cadena), (b.clave, b.cadena))          # la misma clave, otro prefijo

    def test_con_clave_se_piden_los_cuatro_tipos_y_las_dos_ramas(self):
        pedidos = core.requeridos({"clave": ZPUB})
        self.assertEqual(len(pedidos), 8)
        self.assertTrue(any(p["desc"].startswith("wpkh(xpub") and "/1/*)#" in p["desc"] for p in pedidos))
        self.assertTrue(any(p["desc"].startswith("sh(wpkh(xpub") for p in pedidos))
        self.assertEqual(len({p["marca"] for p in pedidos}), 8)

    def test_con_direcciones_sueltas_una_por_direccion(self):
        pedidos = core.requeridos({"direcciones": ["bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu"]})
        self.assertEqual([(p["desc"].split("#")[0], p["rango"]) for p in pedidos],
                         [("addr(bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu)", None)])


class ConCore(ConNodo):
    """La misma cadena inventada, vista además por un Bitcoin Core de mentira."""

    def setUp(self):
        super().setUp()
        self.core = CoreFalso(self.cadena)
        self.addCleanup(self.core.parar)

    def config(self, monederos, **cambios) -> dict:
        c = super().config(monederos)
        c["servidores"] = [self.core.servidor_config(**cambios)]
        btc.guardar(self.carpeta, c)
        return c

    def cliente(self, **cambios) -> core.Cliente:
        return conectar([self.core.servidor_config(**cambios)])[0]

    def cuando_acabe(self, tarea, plazo: float = 5.0):
        """Repite hasta que el nodo deja de repasar (el repaso va en otro hilo)."""
        limite = time.time() + plazo
        while True:
            try:
                return tarea()
            except REPASO as e:
                if "repasando la cadena" not in str(e) or time.time() > limite:
                    raise
                time.sleep(0.02)


class TestLeerDelNodo(ConCore):
    def test_la_primera_vez_avisa_del_repaso_y_despues_lee_como_siempre(self):
        self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100)
        self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
        with self.assertRaisesRegex(REPASO, "repasando la cadena") as aviso:
            self.comprobar()
        self.assertIn("repasando la cadena", str(aviso.exception))
        r = self.cuando_acabe(self.comprobar)
        self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.5)
        self.assertEqual(r["ultima"]["servidor"], "core de prueba")
        # lo nuevo ya no necesita repaso: el nodo lo va viendo él solo
        self.paga(ZPUB, "p2wpkh", 0, 1, 0.25, 200)
        r = self.comprobar()
        self.assertEqual([(p["clase"], p["accion"], p["cantidad"]) for p in r["partes"]], [("entrada", "compra", 0.25)])
        self.assertEqual(self.core.repasos, 1)

    def test_el_monedero_del_nodo_no_tiene_claves_privadas(self):
        self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
        self.cuando_acabe(self.comprobar)
        self.assertEqual(self.core.existentes, {"caudal"})                 # el nodo de mentira rechaza otra cosa
        self.assertNotIn("getdescriptorinfo", self.core.ordenes)
        self.assertLessEqual(set(self.core.ordenes), set(core.METODOS))     # lo que hay que permitir en rpcwhitelist

    def test_por_el_nodo_y_por_electrum_se_ve_exactamente_lo_mismo(self):
        a = self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100)
        self.paga(ZPUB, "p2wpkh", 0, 3, 0.2, 110)
        # un pago a otro con su vuelta, una recompensa de minería y algo todavía sin confirmar
        self.cadena.añadir(construir([(a, 0)], [(10_000_000, AJENA), (39_990_000, script(ZPUB, "p2wpkh", 1, 0))]), 120)
        self.cadena.añadir(construir([], [(312_500_000, script(ZPUB, "p2wpkh", 0, 5))], coinbase=True), 130)
        self.paga(ZPUB, "p2wpkh", 0, 6, 0.01, 0)
        extendida = bitcoin.ClaveExtendida(ZPUB)
        huellas = [d["huella"] for cambio in (0, 1) for d in bitcoin.direcciones(extendida, "p2wpkh", cambio, 0, 25)]
        nodo = self.cliente()
        self.cuando_acabe(lambda: nodo.preparar({"clave": ZPUB}))
        with ClienteElectrum("127.0.0.1", self.nodo.puerto) as electrum:
            self.assertEqual(nodo.historiales(huellas), electrum.historiales(huellas))
            self.assertEqual(nodo.saldos(huellas), electrum.saldos(huellas))
            self.assertEqual(nodo.transacciones([a]), electrum.transacciones([a]))
            self.assertEqual(nodo.cabeceras([100, 130]), electrum.cabeceras([100, 130]))
            self.assertEqual(nodo.altura(), electrum.altura())
            m = {"clave": ZPUB, "tipo": "p2wpkh"}
            self.assertEqual(btc.escanear(nodo, m), btc.escanear(electrum, m))
        self.assertGreater(sum(s["confirmed"] for s in nodo.saldos(huellas)), 3 * SATOSHI)

    def test_un_solo_repaso_vale_para_adivinar_el_tipo_revisar_y_leer(self):
        self.paga(XPUB_TAPROOT, "p2tr", 0, 0, 0.2, 100)
        nodo = self.cliente()
        self.assertEqual(self.cuando_acabe(lambda: btc.detectar(nodo, XPUB_TAPROOT)), ("p2tr", True))
        nodo = self.cliente()
        revision = btc.revisar(nodo, {"clave": XPUB_TAPROOT, "tipo": "p2wpkh"})
        self.assertEqual(revision["tipos"]["p2tr"], {"usadas": 1, "transacciones": 1})
        self.assertEqual(btc.escanear(nodo, {"clave": XPUB_TAPROOT, "tipo": "p2tr"})["sats"], 20_000_000)
        self.assertEqual(self.core.repasos, 1)

    def test_las_direcciones_sueltas_tambien(self):
        direccion = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 2, 1)[0]["direccion"]
        self.paga(ZPUB, "p2wpkh", 0, 2, 0.3, 100)
        self.config([{"nombre": "Papel", "direcciones": direccion, "activo": "C1"}])
        r = self.cuando_acabe(self.comprobar)
        self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.3)
        # añadir otra dirección obliga a repasar solo por ella
        otra = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 4, 1)[0]["direccion"]
        self.paga(ZPUB, "p2wpkh", 0, 4, 0.1, 110)
        nodo = self.cliente()
        r = self.cuando_acabe(lambda: btc.escanear(nodo, {"direcciones": [direccion, otra]}))
        self.assertEqual(r["sats"], 40_000_000)
        self.assertEqual(self.core.repasos, 2)

    def test_avisa_si_el_monedero_tiene_mas_direcciones_de_las_que_sigue_el_nodo(self):
        tope = core.Cliente.tope_rama
        core.Cliente.tope_rama = 20
        try:
            for i in range(20):
                self.paga(ZPUB, "p2wpkh", 0, i, 0.01, 700 + i)
            nodo = self.cliente()
            r = self.cuando_acabe(lambda: btc.escanear(nodo, {"clave": ZPUB, "tipo": "p2wpkh"}))
        finally:
            core.Cliente.tope_rama = tope
        self.assertTrue(r["cortada"])

    def test_una_transaccion_que_no_es_del_monedero_se_le_pide_al_nodo(self):
        ajena = self.cadena.añadir(construir([("cc" * 32, 0)], [(5_000, AJENA)]), 90)
        nodo = self.cliente()
        self.assertEqual(nodo.transacciones([ajena]), [self.cadena.txs[ajena]["raw"]])
        with self.assertRaises(ErrorElectrum) as e:
            nodo.transacciones(["dd" * 32])
        self.assertIn("txindex", str(e.exception))


class TestRepaso(ConCore):
    def test_mientras_repasa_dice_por_donde_va_y_no_ensena_cifras_a_medias(self):
        self.core.repaso_lento = True
        self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100)
        self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
        with self.assertRaisesRegex(REPASO, "repasando la cadena"):
            self.comprobar()
        for _ in range(100):                                   # hasta que el nodo se pone a ello
            if self.core.repasando:
                break
            time.sleep(0.02)
        with self.assertRaisesRegex(REPASO, "repasando la cadena") as aviso:
            self.comprobar()
        self.assertIn("37 %", str(aviso.exception))
        self.assertEqual(self.core.repasos, 1)                 # no lanza otro repaso encima
        self.core.soltar.set()
        self.assertEqual(self.cuando_acabe(self.comprobar)["ultima"]["cuadres"][0]["en_nodo"], 0.5)

    def _a_medias(self, arranque: int) -> dict:
        """Deja el nodo como si Caudal se hubiera cerrado con el repaso en marcha."""
        monedero = {"clave": ZPUB, "tipo": "p2wpkh"}
        pedidos = core.requeridos(monedero)
        nodo = self.cliente()
        nodo._monedero()
        nodo._rpc("importdescriptors", [[{"desc": p["desc"], "timestamp": "now", "range": [0, 999]} for p in pedidos]],
                  monedero=True)
        nodo._etiquetar([p["marca"] for p in pedidos], f"{core.REPASANDO}{arranque}")
        return monedero

    def test_si_caudal_se_cerro_pero_el_nodo_termino_no_repasa_otra_vez(self):
        self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100)
        monedero = self._a_medias(int(self.core.arranque))
        self.core.repasados |= set(self.core.scripts)          # el nodo acabó por su cuenta
        self.assertEqual(btc.escanear(self.cliente(), monedero)["sats"], 50_000_000)
        self.assertEqual(self.core.repasos, 0)
        self.assertEqual(set(self.core.etiquetas.values()), {core.ETIQUETA})

    def test_si_el_nodo_se_reinicio_a_medias_vuelve_a_repasar(self):
        self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100)
        monedero = self._a_medias(int(self.core.arranque))
        self.core.reiniciar()                                   # el repaso se quedó a medias: no ha visto nada
        self.core.arranque += 1000
        with self.assertRaisesRegex(REPASO, "repasando la cadena"):
            btc.escanear(self.cliente(), monedero)
        nodo = self.cliente()
        self.assertEqual(self.cuando_acabe(lambda: btc.escanear(nodo, monedero))["sats"], 50_000_000)
        self.assertEqual(self.core.repasos, 1)

    def test_un_nodo_podado_lo_dice_claro(self):
        self.core.podado = True
        self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
        with self.assertRaisesRegex(btc.ErrorBitcoin, "podado"):
            self.comprobar()
        self.assertEqual(self.core.repasos, 0)


class TestConfiguracionDelNodo(ConCore):
    def test_probar_dice_version_y_altura(self):
        self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 432)
        r = btc.probar(btc.cargar(self.carpeta), self.core.servidor_config())
        self.assertEqual((r["version"], r["altura"], r["podado"]), ("Satoshi:31.0.0", 432, False))
        self.assertNotIn("clave", r["servidor"])

    def test_el_puerto_de_partida_es_el_de_rpc(self):
        s = btc.normalizar_servidor({"host": "minodo.local", "clase": "core", "usuario": "u", "clave": "c"})
        self.assertEqual((s["puerto"], s["clase"], s["usuario"], s["clave"]), (8332, "core", "u", "c"))
        self.assertNotIn("clase", btc.normalizar_servidor({"host": "minodo.local"}))

    def test_con_la_contrasena_mala_o_sin_monedero_lo_dice_claro(self):
        with self.assertRaises(ErrorElectrum) as e:
            self.cliente(clave="otra")
        self.assertIn("usuario y esa contraseña", str(e.exception))
        self.core.con_monedero = False
        with self.assertRaises(ErrorElectrum) as e:
            self.cliente().preparar({"clave": ZPUB})
        self.assertIn("parte de monedero", str(e.exception))

    def test_si_no_es_un_nodo_lo_dice(self):
        otro = socket.socket()                                   # algo que escucha pero no es un nodo
        otro.bind(("127.0.0.1", 0))
        otro.listen(1)
        self.addCleanup(otro.close)

        def contestar():
            try:
                cliente, _ = otro.accept()
                cliente.recv(65536)
                cliente.sendall(b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\n\r\nhola")
                cliente.close()
            except OSError:
                pass
        threading.Thread(target=contestar, daemon=True).start()
        with self.assertRaises(ErrorElectrum) as e:
            conectar([self.core.servidor_config(puerto=otro.getsockname()[1])], espera=2)
        self.assertIn("no contesta como un nodo de Bitcoin", str(e.exception))

    def test_la_contrasena_de_rpc_no_sale_a_la_pantalla_y_se_conserva_al_guardar(self):
        self.config([])
        visto = self.app.btc_estado()["servidores"]
        self.assertEqual([(s["clase"], s["usuario"], s["con_clave"], "clave" in s) for s in visto],
                         [("core", "caudal", True, False)])
        # la pantalla devuelve la lista como la recibió (sin contraseña), con otro nombre
        self.app.btc_servidores({"servidores": [{**visto[0], "nombre": "El de casa"}]})
        guardado = btc.cargar(self.carpeta)["servidores"][0]
        self.assertEqual((guardado["nombre"], guardado["clave"]), ("El de casa", "secreto-de-rpc"))
        self.assertEqual(self.app.btc_probar({"servidor": visto[0]})["version"], "Satoshi:31.0.0")
        # si se cambia de nodo, la contraseña vieja no se le manda al nuevo
        self.app.btc_servidores({"servidores": [{**visto[0], "host": "otro.local"}]})
        self.assertEqual(btc.cargar(self.carpeta)["servidores"][0]["clave"], "")


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