""" Pruebas de «Bitcoin desde mi nodo»: derivar direcciones, hablar con un servidor Electrum (de mentira) y proponer las operaciones de las transacciones nuevas. Las claves públicas que se usan aquí son las de los ejemplos oficiales de los BIP 84 y 86 (las del monedero de pruebas «abandon … about», públicas y sin un duro): no hay nada personal en este fichero. python3 -m unittest discover -s pruebas -v """ from __future__ import annotations import datetime as dt import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from cartera import btc, bitcoin # noqa: E402 from cartera.datos import Datos, escribir # noqa: E402 from cartera.electrum import Cliente, ErrorElectrum # noqa: E402 from cartera.operaciones import Cartera # noqa: E402 from cartera.servidor import App # noqa: E402 from nodo_falso import Cadena, NodoFalso, construir # noqa: E402 ZPUB = ("zpub6rFR7y4Q2AijBEqTUquhVz398htDFrtymD9xYYfG1m4wAcvPhXNfE3EfH1r1ADqtfSdVCToUG868Rv" "UUkgDKf31mGDtKsAYz2oz2AGutZYs") XPUB_TAPROOT = ("xpub6BgBgsespWvERF3LHQu6CnqdvfEvtMcQjYrcRzx53QJjSxarj2afYWcLteoGVky7D3UKDP9Qyr" "LprQ3VCECoY49yfdDEHGCtMMj92pReUsQ") AJENA = "76a914000102030405060708090a0b0c0d0e0f1011121388ac" # dirección de otro SATOSHI = 100_000_000 def script(clave: str, tipo: str, cambio: int, indice: int) -> str: return bitcoin.direcciones(bitcoin.ClaveExtendida(clave), tipo, cambio, indice, 1)[0]["script"] class TestCriptografia(unittest.TestCase): def test_ripemd160_vectores_conocidos(self): for mensaje, esperado in [(b"", "9c1185a5c5e9fc54612808977ee8f548b2258d31"), (b"abc", "8eb208f7e05d987a9b044a8e98c6b087f15a0bfc"), (b"message digest", "5d0689ef49d2fae572b881b123a85ffa21595f36"), (b"a" * 1_000_000, "52783243c1697bdbe16d37f97f68f08325dc1528")]: self.assertEqual(bitcoin._ripemd160_puro(mensaje).hex(), esperado) self.assertEqual(bitcoin.ripemd160(mensaje).hex(), esperado) def test_direcciones_del_ejemplo_oficial_bip84(self): z = bitcoin.ClaveExtendida(ZPUB) self.assertEqual((z.tipo_sugerido, z.red), ("p2wpkh", "mainnet")) recibir = bitcoin.direcciones(z, "p2wpkh", 0, 0, 2) cambio = bitcoin.direcciones(z, "p2wpkh", 1, 0, 1) self.assertEqual([d["direccion"] for d in recibir + cambio], ["bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu", "bc1qnjg0jd8228aq7egyzacy8cys3knf9xvrerkf9g", "bc1q8c6fshw2dlwun7ekn9qwf37cu2rn755upcp6el"]) def test_direcciones_del_ejemplo_oficial_bip86_taproot(self): x = bitcoin.ClaveExtendida(XPUB_TAPROOT) d = bitcoin.direcciones(x, "p2tr", 0, 0, 2) + bitcoin.direcciones(x, "p2tr", 1, 0, 1) self.assertEqual([r["direccion"] for r in d], ["bc1p5cyxnuxmeuwuvkwfem96lqzszd02n6xdcjrs20cac6yqjjwudpxqkedrcr", "bc1p4qhjn9zdvkux4e44uhx8tc55attvtyu358kutcqkudyccelu0was9fqzwh", "bc1p3qkhfews2uk44qtvauqyr2ttdsw7svhkl9nkm9s9c3x4ax5h60wqwruhk7"]) def test_los_tres_formatos_de_direccion_van_y_vuelven(self): z = bitcoin.ClaveExtendida(ZPUB) for tipo, empieza in [("p2pkh", "1"), ("p2sh-p2wpkh", "3"), ("p2wpkh", "bc1q"), ("p2tr", "bc1p")]: d = bitcoin.direcciones(z, tipo, 0, 0, 1)[0] self.assertTrue(d["direccion"].startswith(empieza), (tipo, d["direccion"])) self.assertEqual(bitcoin.huella_script(bytes.fromhex(d["script"])), d["huella"]) # Base58Check detecta una errata en la clave with self.assertRaises(ValueError): bitcoin.ClaveExtendida(ZPUB[:-1] + ("A" if ZPUB[-1] != "A" else "B")) def test_avisa_si_le_das_una_clave_privada(self): privada = ("zprvAWgYBBk7JR8Gjrh4UJQ2uJdG1r3WNRRfURiABBE3RvMXYSrRJL62XuezvGdPvG6GFBZduosC" "c1YP5wixPox7zhZLfiUm8aunE96BBa4Kei5") with self.assertRaises(ValueError) as e: bitcoin.ClaveExtendida(privada) self.assertIn("PRIVADA", str(e.exception)) def test_lee_una_transaccion_con_testigos(self): entrada = [("aa" * 32, 0)] salidas = [(12345, script(ZPUB, "p2wpkh", 0, 0))] con = bitcoin.leer_transaccion(construir(entrada, salidas, testigos=True)) sin = bitcoin.leer_transaccion(construir(entrada, salidas, testigos=False)) # El identificador no cuenta los testigos: ha de salir el mismo self.assertEqual(con["txid"], sin["txid"]) self.assertEqual(con["salidas"][0]["satoshis"], 12345) self.assertFalse(con["coinbase"]) self.assertTrue(bitcoin.leer_transaccion(construir([], salidas, coinbase=True))["coinbase"]) def test_fecha_de_la_cabecera_del_bloque(self): cadena = Cadena() self.assertEqual(bitcoin.momento_de_cabecera(cadena.cabecera(10)), 1_735_689_600 + 6000) class ConNodo(unittest.TestCase): """Base: un nodo falso levantado y una cartera de mentira en un directorio temporal.""" def setUp(self): self.cadena = Cadena() self.nodo = NodoFalso(self.cadena) self.nodo.start() self.addCleanup(self.nodo.parar) self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.carpeta = Path(self.tmp.name) / "datos" self.carpeta.mkdir() escribir(self.carpeta, Datos( [{"id": "C1", "nombre": "Bitcoin", "tipo": "Cripto", "codigo": "bitcoin", "unidad": "BTC", "estado": "abierta", "fecha_inicio": "2020-01-01"}], [{"id": "m1", "fecha": "2020-09-14", "activo": "C1", "tipo": "Compra", "cantidad": 0.5, "precio": 9000.0, "importe": 4500.0}], [{"fecha": "2026-01-01", "activo": "C1", "precio": 90000.0, "fuente": "prueba"}], [])) self.cartera = Cartera(self.carpeta) self.addCleanup(self.cartera.cerrar) self.app = App(self.cartera) def config(self, monederos) -> dict: c = btc.cargar(self.carpeta) c["servidores"] = [self.nodo.servidor_config()] c["monederos"] = [btc.normalizar_monedero(m, []) for m in monederos] btc.guardar(self.carpeta, c) return c def comprobar(self) -> dict: return self.app.btc_comprobar() def paga(self, clave, tipo, cambio, indice, btc_cantidad, altura, desde=None, vuelta=None): """Apunta en la cadena falsa una transacción que paga a una dirección nuestra.""" entradas = [desde] if desde else [("bb" * 32, 0)] salidas = [(round(btc_cantidad * SATOSHI), script(clave, tipo, cambio, indice))] if vuelta: salidas.append(vuelta) return self.cadena.añadir(construir(entradas, salidas), altura) class TestCuadreYPropuestas(ConNodo): """La comprobación diaria: el motor del historial enseñando solo lo nuevo.""" def decidir(self, partes, **cambios) -> dict: return self.app.btc_historial_aplicar({"decisiones": [{**p, **cambios} for p in partes]}) def nuevos(self) -> list[dict]: return [m for m in self.cartera.datos.vivos("movimientos") if m["id"] != "m1"] def test_un_monedero_nuevo_no_arrastra_su_pasado(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) r = self.comprobar() self.assertTrue(r["diaria"]) self.assertEqual(r["partes"], []) # el pasado se da por visto… self.assertEqual(r["atrasadas"], 1) # …y espera en el historial completo cuadre = r["ultima"]["cuadres"][0] self.assertEqual(cuadre["en_nodo"], 0.5) self.assertEqual(cuadre["apuntado"], 0.5) self.assertEqual(cuadre["diferencia"], 0.0) # y queda constancia de que ya no es de estreno guardada = btc.cargar(self.carpeta) self.assertFalse(guardada["monederos"][0]["inicial"]) self.assertEqual(len(guardada["monederos"][0]["vistas"]), 1) self.assertEqual(guardada["monederos"][0]["saldo"], 0.5) self.assertEqual(guardada["monederos"][0]["direcciones_usadas"], 1) def test_una_entrada_nueva_se_propone_como_compra_con_su_fecha_y_precio(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 1, 0.25, 200) r = self.comprobar() self.assertEqual(len(r["partes"]), 1) p = r["partes"][0] self.assertEqual((p["clase"], p["accion"], p["cantidad"]), ("entrada", "compra", 0.25)) self.assertEqual(p["fecha"], dt.datetime.fromtimestamp( 1_735_689_600 + 200 * 600, dt.timezone.utc).date().isoformat()) self.assertIsNotNone(p["precio_ref"]) # cierre de ese día self.assertEqual(p["importe_ref"], round(p["precio_ref"] * 0.25, 2)) self.assertTrue(p["confirmada"]) self.assertEqual(p["nombre_activo"], "Bitcoin · Sparrow") # el cuadre de cada lugar, para la revisión; el del activo, para el resumen self.assertEqual((r["cuadres"][0]["clave"], r["cuadres"][0]["en_nodo"]), ("C1@l-sparrow", 0.75)) cuadre = r["ultima"]["cuadres"][0] self.assertEqual((cuadre["en_nodo"], cuadre["apuntado"], cuadre["diferencia"]), (0.75, 0.5, 0.25)) def test_al_guardarla_deja_de_proponerse_y_cuadra(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 1, 0.25, 200) parte = self.comprobar()["partes"][0] r = self.decidir([parte], importe=22500) self.assertEqual((r["guardadas"], r["fallos"]), (1, [])) guardado = self.nuevos()[0] self.assertEqual((guardado["tipo"], guardado["cantidad"], guardado["importe"]), ("Compra", 0.25, 22500.0)) self.assertEqual(guardado["lugar"], "l-sparrow") self.assertTrue(guardado["estimado"]) # el importe en euros hay que revisarlo self.assertEqual(guardado["extracto"], f"btc:{parte['txid']}") # el aviso del resumen se pone al día sin volver a preguntar al nodo ultima = self.app.estado()["btc"]["ultima"] self.assertEqual((ultima["pendientes"], ultima["cuadres"][0]["diferencia"]), (0, 0.0)) segunda = self.comprobar() self.assertEqual(segunda["partes"], []) self.assertEqual(segunda["ultima"]["cuadres"][0]["diferencia"], 0.0) def test_lo_que_dices_que_no_se_apunte_no_vuelve_a_salir(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 150) self.decidir(self.comprobar()["partes"], accion="no_apuntar") self.assertEqual(self.comprobar()["partes"], []) self.assertEqual(self.nuevos(), []) def test_lo_que_dejas_sin_decidir_vuelve_a_salir(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 150) self.assertEqual(len(self.comprobar()["partes"]), 1) self.assertEqual(len(self.comprobar()["partes"]), 1) # y es la misma que se ve en el historial completo self.assertEqual([p["id"] for p in self.app.btc_historial()["partes"]], [p["id"] for p in self.comprobar()["partes"]]) def test_si_decides_solo_una_parte_la_otra_sigue_siendo_novedad(self): origen = self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.cadena.añadir(construir([(origen, 0)], [(round(0.2 * SATOSHI), AJENA), (round(0.29 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 300) partes = self.comprobar()["partes"] self.decidir([p for p in partes if p["clase"] == "salida"], accion="venta", importe=18000) self.assertEqual([p["clase"] for p in self.comprobar()["partes"]], ["comision"]) def test_una_compra_que_ya_tenias_apuntada_se_propone_enlazar(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 0, 0.25, 200) fecha = dt.datetime.fromtimestamp(1_735_689_600 + 200 * 600, dt.timezone.utc).date() mia = self.cartera.guardar_movimiento({"activo": "C1", "tipo": "Compra", "cantidad": 0.25, "fecha": (fecha - dt.timedelta(days=2)).isoformat(), "importe": 21000}) p = self.comprobar()["partes"][0] self.assertEqual((p["accion"], p["enlace"]), ("enlazar", mia["id"])) self.decidir([p]) self.assertEqual(len(self.nuevos()), 1) # no se crea otra enlazada = self.nuevos()[0] self.assertEqual((enlazada["importe"], enlazada["txid"], enlazada["lugar"]), (21000.0, p["txid"], "l-sparrow")) self.assertEqual(self.comprobar()["partes"], []) def test_una_salida_no_se_supone_venta_y_la_comision_va_aparte(self): origen = self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() # gasta 0,5: manda 0,2 fuera y se devuelve 0,29 a su dirección de cambio self.cadena.añadir(construir([(origen, 0)], [(round(0.2 * SATOSHI), AJENA), (round(0.29 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 300) r = self.comprobar() self.assertEqual([(p["clase"], p["accion"], round(p["cantidad"], 8)) for p in r["partes"]], [("salida", "", 0.2), ("comision", "comision", 0.01)]) self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.29) def test_el_bitcoin_minado_se_propone_como_mineria(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.cadena.añadir(construir([], [(round(0.05 * SATOSHI), script(ZPUB, "p2wpkh", 0, 0))], coinbase=True), 400) p = self.comprobar()["partes"][0] self.assertEqual((p["clase"], p["accion"], p["cantidad"]), ("mineria", "mineria", 0.05)) self.decidir([p]) guardado = self.nuevos()[0] self.assertEqual((guardado["tipo"], guardado["cantidad"], guardado.get("precio")), ("Mineria", 0.05, None)) def test_mover_entre_monederos_propios_es_un_traslado_y_su_comision(self): origen = self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}, {"nombre": "Móvil", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C1"}]) self.comprobar() self.cadena.añadir(construir([(origen, 0)], [(round(0.4999 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0))]), 500) r = self.comprobar() self.assertEqual([(p["clase"], p["accion"], round(p["cantidad"], 8)) for p in r["partes"]], [("interno", "traslado", 0.4999), ("comision", "comision", 0.0001)]) self.assertEqual(sorted(r["partes"][0]["monederos"]), ["Móvil", "Sparrow"]) self.assertAlmostEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.4999, places=8) self.assertEqual(self.decidir(r["partes"])["guardadas"], 2) self.assertEqual(sorted(m["tipo"] for m in self.nuevos()), ["Comision", "Traslado"]) # ni compra ni venta: solo se ha ido la comisión self.assertEqual(self.app.estado()["btc"]["ultima"]["cuadres"][0]["diferencia"], 0.0) def test_deshacer_lo_guardado_lo_devuelve_al_historial_no_a_las_novedades(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 150) r = self.decidir(self.comprobar()["partes"], importe=9000) self.app.btc_deshacer(r["importacion"]) self.assertEqual(self.nuevos(), []) despues = self.comprobar() self.assertEqual((despues["partes"], despues["atrasadas"]), ([], 1)) self.assertEqual(len(self.app.btc_historial()["partes"]), 1) def test_las_direcciones_se_buscan_mas_alla_del_hueco_habitual(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 15, 0.1, 590) # la 15 usada empuja la búsqueda más allá self.decidir(self.comprobar()["partes"], accion="no_apuntar") self.paga(ZPUB, "p2wpkh", 0, 30, 0.2, 600) # la 30 ya cae fuera de las 20 primeras r = self.comprobar() self.assertEqual(len(r["partes"]), 1) self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.3) # 0,1 + 0,2 def test_avisa_si_para_de_buscar_antes_de_acabar(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) tope = btc.MAX_DIRECCIONES btc.MAX_DIRECCIONES = 20 # como si el monedero fuera enorme try: for i in range(20): # ninguna tanda queda vacía del todo self.paga(ZPUB, "p2wpkh", 0, i, 0.01, 700 + i) cuadre = self.comprobar()["ultima"]["cuadres"][0] finally: btc.MAX_DIRECCIONES = tope self.assertEqual(cuadre["cortados"], ["Sparrow"]) def test_el_tipo_de_direcciones_se_puede_adivinar(self): self.paga(XPUB_TAPROOT, "p2tr", 0, 0, 0.2, 100) with Cliente("127.0.0.1", self.nodo.puerto) as c: self.assertEqual(btc.detectar_tipo(c, XPUB_TAPROOT), "p2tr") self.assertEqual(btc.detectar_tipo(c, ZPUB), "p2wpkh") # sin historial: el del prefijo def test_una_xpub_no_se_da_por_antigua_se_le_pregunta_al_nodo(self): """Muchos monederos dan una «xpub» y usan direcciones bc1q: el prefijo no dice el tipo.""" self.assertTrue(XPUB_TAPROOT.startswith("xpub")) with Cliente("127.0.0.1", self.nodo.puerto) as c: self.assertEqual(btc.detectar(c, XPUB_TAPROOT), ("p2wpkh", False)) # sin estrenar: bc1q de partida, no «1…» self.paga(XPUB_TAPROOT, "p2wpkh", 0, 0, 0.1, 101) with Cliente("127.0.0.1", self.nodo.puerto) as c: self.assertEqual(btc.detectar(c, XPUB_TAPROOT), ("p2wpkh", True)) self.paga(XPUB_TAPROOT, "p2pkh", 0, 1, 0.1, 102) # y si de verdad usa las antiguas, se ve with Cliente("127.0.0.1", self.nodo.puerto) as c: self.assertEqual(btc.detectar(c, XPUB_TAPROOT)[1], True) class TestConfiguracion(ConNodo): def test_la_clave_no_se_devuelve_entera_a_la_pantalla(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) estado = self.app.btc_estado() m = estado["monederos"][0] self.assertNotIn("clave", m) self.assertTrue(m["clave_corta"].startswith("zpub6rFR7y4Q")) self.assertIn("…", m["clave_corta"]) self.assertEqual([a["id"] for a in estado["activos"]], ["C1"]) def test_editar_el_nombre_no_obliga_a_pegar_la_clave_otra_vez(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) mid = btc.cargar(self.carpeta)["monederos"][0]["id"] self.app.btc_monedero({"monedero": {"id": mid, "nombre": "Frío", "activo": "C1"}}) guardado = btc.cargar(self.carpeta)["monederos"][0] self.assertEqual((guardado["nombre"], guardado["clave"]), ("Frío", ZPUB)) def test_no_admite_dos_veces_la_misma_clave(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) with self.assertRaises(btc.ErrorBitcoin): self.app.btc_monedero({"monedero": {"nombre": "Otro", "clave": ZPUB, "activo": "C1"}}) def test_el_fichero_de_claves_solo_lo_lee_su_dueno(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.assertEqual(btc.ruta_config(self.carpeta).stat().st_mode & 0o777, 0o600) def test_direcciones_de_servidor_escritas_de_cualquier_manera(self): s = btc.normalizar_servidor({"host": "tcp://abc.onion:50001/"}) self.assertEqual((s["host"], s["puerto"], s["tor"]), ("abc.onion", 50001, True)) self.assertFalse(btc.normalizar_servidor({"host": "192.168.1.50", "puerto": 50002, "tls": True})["tor"]) with self.assertRaises(btc.ErrorBitcoin): btc.normalizar_servidor({"host": ""}) def test_probar_el_servidor_dice_version_y_altura(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 777) r = self.app.btc_probar({"servidor": self.nodo.servidor_config()}) self.assertIn("Electrs", r["version"]) self.assertEqual(r["altura"], 777) def test_sin_servidores_o_sin_monederos_avisa_en_claro(self): c = btc.cargar(self.carpeta) btc.guardar(self.carpeta, c) with self.assertRaises(btc.ErrorBitcoin) as e: self.app.btc_comprobar() self.assertIn("monedero", str(e.exception)) def test_si_ningun_servidor_responde_lo_dice_con_los_motivos(self): c = btc.cargar(self.carpeta) c["servidores"] = [{"nombre": "Apagado", "host": "127.0.0.1", "puerto": 1, "tor": False}] c["monederos"] = [btc.normalizar_monedero({"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}, [])] btc.guardar(self.carpeta, c) with self.assertRaises(ErrorElectrum) as e: self.app.btc_comprobar() self.assertIn("Apagado", str(e.exception)) def test_prueba_los_servidores_en_orden_hasta_que_uno_conteste(self): c = btc.cargar(self.carpeta) c["servidores"] = [{"nombre": "Start9", "host": "127.0.0.1", "puerto": 1, "tor": False}, self.nodo.servidor_config("Umbrel")] c["monederos"] = [btc.normalizar_monedero({"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}, [])] btc.guardar(self.carpeta, c) r = self.app.btc_comprobar() self.assertEqual(r["servidor"], "Umbrel") self.assertTrue(any("Start9" in f for f in r["fallos"])) def test_el_servidor_del_propio_aparato_solo_se_propone_si_contesta(self): import os self.addCleanup(os.environ.pop, "CAUDAL_ELECTRUM", None) # varios candidatos: el primero no existe, el segundo es el nodo de la prueba os.environ["CAUDAL_ELECTRUM"] = f"no-existe.invalid:50001, 127.0.0.1:{self.nodo.puerto}" self.assertEqual(self.app.btc_estado()["sugerencia"], {"host": "127.0.0.1", "puerto": self.nodo.puerto, "tor": False, "tls": False}) os.environ["CAUDAL_ELECTRUM"] = "no-existe.invalid:50001" self.assertIsNone(self.app.btc_estado()["sugerencia"]) # ninguno contesta: no se propone nada os.environ["CAUDAL_ELECTRUM"] = f"127.0.0.1:{self.nodo.puerto}" self.config([]) # con un servidor ya puesto, ni se busca antes = len(self.nodo.peticiones) self.assertIsNone(self.app.btc_estado()["sugerencia"]) self.assertEqual(len(self.nodo.peticiones), antes) def test_en_un_servidor_el_fallo_explica_que_quien_conecta_es_el_servidor(self): from cartera import electrum self.addCleanup(setattr, electrum, "EN_SERVIDOR", electrum.EN_SERVIDOR) onion = [{"nombre": "Mi nodo", "host": "abcdefabcdefabcd.onion", "puerto": 50001, "tor": True}] libre = __import__("socket").socket() libre.bind(("127.0.0.1", 0)) proxy = ("127.0.0.1", libre.getsockname()[1]) libre.close() # ahí no hay ningún Tor electrum.EN_SERVIDOR = False with self.assertRaises(ErrorElectrum) as e: electrum.conectar(onion, proxy, espera=2) self.assertIn("en el teléfono, Orbot; en el ordenador, el servicio Tor o el Navegador Tor", str(e.exception)) self.assertNotIn("el servidor donde está Caudal", str(e.exception)) electrum.EN_SERVIDOR = True with self.assertRaises(ErrorElectrum) as e: electrum.conectar(onion, proxy, espera=2) self.assertIn("El servidor donde va Caudal no tiene Tor", str(e.exception)) self.assertIn("la hace el servidor donde está Caudal, no el dispositivo", str(e.exception)) def test_el_resumen_del_estado_general_cuenta_lo_pendiente(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.comprobar() self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 150) self.comprobar() resumen = self.app.estado()["btc"] self.assertEqual(resumen["monederos"], 1) self.assertEqual(resumen["ultima"]["pendientes"], 1) self.assertEqual(resumen["ultima"]["servidor"], "prueba") if __name__ == "__main__": unittest.main() def _escuchando(): import socket as _s srv = _s.socket(); srv.bind(("127.0.0.1", 0)); srv.listen(4) return srv, srv.getsockname()[1] def test_proxy_de_usa_el_tor_que_contesta(monkeypatch): """El de fábrica cede ante el primer Tor que contesta: el que declara el aparato o uno habitual.""" srv, puerto = _escuchando() try: monkeypatch.setattr(btc, "_tor_visto", (0.0, None)) monkeypatch.setattr(btc, "TOR_HABITUALES", ()) monkeypatch.setenv("CAUDAL_TOR", f"127.0.0.1:1,127.0.0.1:{puerto}") # el primero no contesta assert btc.proxy_de({}) == ("127.0.0.1", puerto) assert btc.proxy_de({"proxy": {"host": "127.0.0.1", "puerto": 9050}}) == ("127.0.0.1", puerto) # Uno puesto a mano se respeta assert btc.proxy_de({"proxy": {"host": "10.9.9.9", "puerto": 9150}}) == ("10.9.9.9", 9150) # Sin variable, el Navegador Tor (u otro habitual) también vale monkeypatch.setattr(btc, "_tor_visto", (0.0, None)) monkeypatch.delenv("CAUDAL_TOR") monkeypatch.setattr(btc, "TOR_HABITUALES", (("127.0.0.1", 1), ("127.0.0.1", puerto))) assert btc.proxy_de({}) == ("127.0.0.1", puerto) finally: srv.close() def test_proxy_de_sin_ningun_tor(monkeypatch): monkeypatch.setattr(btc, "_tor_visto", (0.0, None)) monkeypatch.setattr(btc, "TOR_HABITUALES", (("127.0.0.1", 1),)) monkeypatch.setenv("CAUDAL_TOR", "sin-puerto") assert btc.proxy_de({}) == ("127.0.0.1", 9050) def test_lista_de_direcciones(): assert btc.lista_de_direcciones(" a.startos:50001, 10.0.0.1:2 ,mal,:3,x:y") == [("a.startos", 50001), ("10.0.0.1", 2)]