496 lines
26 KiB
Python
496 lines
26 KiB
Python
"""
|
|
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)]
|