658 lines
38 KiB
Python
658 lines
38 KiB
Python
"""
|
||
Pruebas de la importación del historial completo de bitcoin (historial_btc.py)
|
||
y del precio de referencia (spot_btc.py), contra el nodo Electrum de mentira.
|
||
|
||
Las claves son las de los ejemplos oficiales BIP 84/86: nada personal.
|
||
|
||
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, historial_btc, spot_btc # noqa: E402
|
||
from cartera.datos import Datos, escribir # 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
|
||
from test_btc import AJENA, SATOSHI, XPUB_TAPROOT, ZPUB, script # noqa: E402
|
||
|
||
P2WSH = "0020" + "ab" * 32 # un contrato (lo que se ve al abrir un canal Lightning)
|
||
|
||
|
||
def fecha_de(altura: int) -> str:
|
||
return dt.datetime.fromtimestamp(1_735_689_600 + altura * 600, dt.timezone.utc).date().isoformat()
|
||
|
||
|
||
class TestPrecioDeReferencia(unittest.TestCase):
|
||
def test_la_serie_empaquetada_cubre_desde_2011(self):
|
||
v, dia = spot_btc.precio("2011-09-13")
|
||
self.assertEqual(dia, "2011-09-13")
|
||
self.assertTrue(1 < v < 20)
|
||
v, dia = spot_btc.precio("2021-11-10")
|
||
self.assertTrue(40_000 < v < 70_000)
|
||
|
||
def test_antes_de_la_serie_no_se_inventa_nada(self):
|
||
self.assertEqual(spot_btc.precio("2009-01-03"), (None, None))
|
||
|
||
def test_despues_de_la_serie_usa_los_precios_de_la_cartera(self):
|
||
ultimo = spot_btc.ultimo_dia_serie()
|
||
despues = (dt.date.fromisoformat(ultimo) + dt.timedelta(days=30)).isoformat()
|
||
self.assertEqual(spot_btc.precio(despues), (None, None)) # demasiado lejos
|
||
propios = [{"fecha": despues, "precio": 123456.0}]
|
||
self.assertEqual(spot_btc.precio(despues, propios), (123456.0, despues))
|
||
|
||
|
||
class ConHistorial(unittest.TestCase):
|
||
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": "C2", "nombre": "BTC en Lightning", "tipo": "Cripto", "codigo": "bitcoin",
|
||
"unidad": "BTC", "estado": "abierta", "fecha_inicio": "2020-01-01"}],
|
||
[], [], []))
|
||
self.cartera = Cartera(self.carpeta)
|
||
self.addCleanup(self.cartera.cerrar)
|
||
self.app = App(self.cartera)
|
||
|
||
def config(self, monederos):
|
||
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)
|
||
|
||
def paga(self, clave, tipo, indice, cantidad, altura, cambio=0):
|
||
return self.cadena.añadir(construir([("bb" * 32, altura)],
|
||
[(round(cantidad * SATOSHI), script(clave, tipo, cambio, indice))]), altura)
|
||
|
||
def apuntar(self, **m):
|
||
return self.cartera.guardar_movimiento({"activo": "C1", **m})
|
||
|
||
def historial(self):
|
||
return self.app.btc_historial()
|
||
|
||
def partes(self, clase=None):
|
||
return [p for p in self.historial()["partes"] if clase is None or p["clase"] == clase]
|
||
|
||
def decidir(self, partes, **cambios):
|
||
return self.app.btc_historial_aplicar({"decisiones": [{**p, **cambios} for p in partes]})
|
||
|
||
def movs(self):
|
||
return self.cartera.datos.vivos("movimientos")
|
||
|
||
|
||
class TestHistorial(ConHistorial):
|
||
def test_trae_el_pasado_que_la_comprobacion_diaria_da_por_apuntado(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.assertEqual(self.app.btc_comprobar()["partes"], []) # la diaria no lo ve
|
||
r = self.historial()
|
||
self.assertEqual(len(r["partes"]), 1)
|
||
p = r["partes"][0]
|
||
self.assertEqual((p["clase"], p["accion"], p["cantidad"], p["fecha"]),
|
||
("entrada", "compra", 0.5, fecha_de(100)))
|
||
self.assertIsNotNone(p["precio_ref"]) # de la serie empaquetada
|
||
self.assertEqual(p["importe_ref"], round(p["precio_ref"] * 0.5, 2))
|
||
self.assertEqual(r["cuadres"], [{"clave": "C1@l-sparrow", "activo": "C1", "lugar": "l-sparrow",
|
||
"nombre_activo": "Bitcoin · Sparrow",
|
||
"en_nodo": 0.5, "apuntado": 0.0}])
|
||
|
||
def test_propone_enlazar_con_la_compra_ya_apuntada_sin_tocar_sus_cifras(self):
|
||
self.apuntar(fecha=fecha_de(100 - 144 * 3), tipo="Compra", cantidad=0.5, importe=40000)
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
p = self.partes()[0]
|
||
self.assertEqual(p["accion"], "enlazar")
|
||
compra = self.movs()[0]
|
||
self.assertEqual(p["enlace"], compra["id"])
|
||
r = self.decidir([p])
|
||
self.assertEqual((r["guardadas"], r["creados"], r["enlazados"]), (1, 0, 1))
|
||
m = self.movs()[0]
|
||
self.assertEqual((m["txid"], m["importe"], m["cantidad"]), (p["txid"], 40000, 0.5))
|
||
self.assertEqual(len(self.movs()), 1)
|
||
self.assertEqual(self.partes(), []) # ya no sale
|
||
|
||
def test_enlazar_la_salida_no_se_lleva_por_delante_la_comision(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.3 * SATOSHI), AJENA),
|
||
(round(0.1999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.decidir(self.partes("entrada"), importe=40000)
|
||
self.apuntar(fecha=fecha_de(200), tipo="Venta", cantidad=0.3, importe=27000)
|
||
salida = self.partes("salida")[0]
|
||
self.assertEqual(salida["accion"], "enlazar")
|
||
self.decidir([salida])
|
||
self.assertEqual([p["clase"] for p in self.partes()], ["comision"])
|
||
|
||
def test_una_compra_apuntada_de_otra_cantidad_no_se_enlaza_sola(self):
|
||
self.apuntar(fecha=fecha_de(100), tipo="Compra", cantidad=0.4, importe=30000)
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
p = self.partes()[0]
|
||
self.assertEqual(p["accion"], "compra")
|
||
self.assertEqual([c["exacta"] for c in p["candidatas"]], [False]) # pero se ofrece a mano
|
||
|
||
def test_guardar_marca_la_importacion_y_no_vuelve_a_salir(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
p = self.partes()[0]
|
||
r = self.decidir([p], importe=30000)
|
||
self.assertEqual((r["guardadas"], r["creados"], r["fallos"]), (1, 1, []))
|
||
m = self.movs()[0]
|
||
self.assertEqual((m["tipo"], m["cantidad"], m["importe"]), ("Compra", 0.5, 30000))
|
||
self.assertEqual(m["extracto"], f"btc:{p['txid']}")
|
||
self.assertEqual(m["importacion"], r["importacion"])
|
||
self.assertTrue(m["estimado"])
|
||
self.assertEqual(self.partes(), [])
|
||
self.assertEqual(self.app.btc_comprobar()["partes"], [])
|
||
|
||
def test_lo_que_guardo_la_comprobacion_diaria_no_se_repite(self):
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.app.btc_comprobar()
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.assertEqual(self.decidir(self.app.btc_comprobar()["partes"], importe=30000)["guardadas"], 1)
|
||
self.assertEqual(self.historial()["partes"], [])
|
||
|
||
def test_lo_que_guardo_la_diaria_de_antes_de_unificarse_tampoco(self):
|
||
# Antes guardaba con extracto «btc:<txid>» y sin importación
|
||
txid = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.apuntar(tipo="Compra", cantidad=0.5, importe=30000, fecha=fecha_de(100), extracto=f"btc:{txid}")
|
||
r = self.historial()
|
||
self.assertEqual((r["partes"], r["ya_apuntadas"]), ([], 1))
|
||
|
||
def test_sin_decidir_no_se_guarda_nada(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.3 * SATOSHI), AJENA),
|
||
(round(0.1999 * SATOSHI), script(ZPUB, "p2wpkh", 0, 0, ))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
salida = self.partes("salida")[0]
|
||
self.assertEqual(salida["accion"], "") # una salida no se supone venta
|
||
r = self.decidir([salida])
|
||
self.assertEqual(r["guardadas"], 0)
|
||
self.assertEqual(len(self.partes("salida")), 1)
|
||
|
||
def test_deshacer_borra_lo_creado_quita_enlaces_y_lo_deja_pendiente(self):
|
||
self.apuntar(fecha=fecha_de(100), tipo="Compra", cantidad=0.5, importe=40000)
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.paga(ZPUB, "p2wpkh", 1, 0.25, 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
r = self.decidir(self.partes())
|
||
self.assertEqual((r["creados"], r["enlazados"]), (1, 1))
|
||
self.assertEqual(self.app.btc_importaciones()[0]["id"], r["importacion"])
|
||
d = self.app.btc_deshacer(r["importacion"])
|
||
self.assertEqual((d["borrados"], d["desenlazados"]), (1, 1))
|
||
restantes = self.movs()
|
||
self.assertEqual(len(restantes), 1)
|
||
self.assertNotIn("txid", restantes[0])
|
||
self.assertEqual(restantes[0]["importe"], 40000)
|
||
self.assertEqual(len(self.partes()), 2)
|
||
self.assertEqual(self.app.btc_importaciones(), [])
|
||
|
||
|
||
class TestConjuntoDeMonederos(ConHistorial):
|
||
def test_mover_a_un_monedero_de_otro_activo_es_un_traspaso_con_su_coste(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.2 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.2999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"},
|
||
{"nombre": "Nodo", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C2"}])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in partes],
|
||
[("entrada", 0.5), ("interno", 0.2), ("comision", 0.0001)])
|
||
interno = partes[1]
|
||
self.assertEqual((interno["activo"], interno["destino"], interno["accion"]), ("C1", "C2", "traspaso"))
|
||
r = self.decidir(partes[:1], importe=40000) # compra a 80.000 €/BTC
|
||
r = self.decidir(partes[1:])
|
||
self.assertEqual((r["guardadas"], r["fallos"]), (2, []))
|
||
por_tipo = {m["tipo"]: m for m in self.movs()}
|
||
self.assertEqual(por_tipo["Traspaso_Salida"]["activo"], "C1")
|
||
self.assertEqual(por_tipo["Traspaso_Entrada"]["activo"], "C2")
|
||
self.assertAlmostEqual(por_tipo["Traspaso_Entrada"]["coste_traspaso"], 16000, places=2) # 0,2 × 80.000
|
||
self.assertAlmostEqual(por_tipo["Comision"]["cantidad"], 0.0001, places=8)
|
||
self.assertEqual(self.partes(), [])
|
||
|
||
def test_una_salida_puede_ir_a_un_activo_tuyo_sin_monedero(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.4 * SATOSHI), P2WSH),
|
||
(round(0.0999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.decidir(self.partes("entrada"), importe=40000)
|
||
salida = self.partes("salida")[0]
|
||
self.assertIn("Lightning", salida["pista"])
|
||
r = self.decidir([salida], accion="traspaso", destino="C2")
|
||
self.assertEqual(r["fallos"], [])
|
||
ln = [m for m in self.movs() if m["activo"] == "C2"]
|
||
self.assertEqual([(m["tipo"], m["cantidad"]) for m in ln], [("Traspaso_Entrada", 0.4)])
|
||
|
||
def test_la_mineria_se_propone_y_se_enlaza_con_la_apuntada(self):
|
||
self.apuntar(fecha=fecha_de(300), tipo="Mineria", cantidad=0.00321)
|
||
self.cadena.añadir(construir([], [(321000, script(ZPUB, "p2wpkh", 0, 0)),
|
||
(5 * SATOSHI, AJENA)], coinbase=True), 300)
|
||
self.cadena.añadir(construir([], [(123000, script(ZPUB, "p2wpkh", 0, 1))], coinbase=True), 301)
|
||
self.config([{"nombre": "Minería", "clave": ZPUB, "activo": "C1", "actividad": "Minería"}])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["accion"]) for p in partes],
|
||
[("mineria", "enlazar"), ("mineria", "mineria")])
|
||
self.assertEqual(partes[1]["actividad"], "Minería")
|
||
self.decidir(partes)
|
||
nueva = [m for m in self.movs() if m.get("importacion")][0]
|
||
self.assertEqual((nueva["tipo"], nueva["cantidad"], nueva["actividad"]), ("Mineria", 0.00123, "Minería"))
|
||
|
||
def test_una_transaccion_con_entradas_de_otros_se_marca_mixta(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
ajeno = self.cadena.añadir(construir([("cc" * 32, 0)], [(SATOSHI, AJENA)]), 101)
|
||
self.cadena.añadir(construir([(origen, 0), (ajeno, 0)],
|
||
[(round(0.3 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0)),
|
||
(round(1.19 * SATOSHI), AJENA)]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
salida = self.partes("salida")[0]
|
||
self.assertTrue(salida["mixta"])
|
||
self.assertAlmostEqual(salida["cantidad"], 0.2, places=8) # efecto neto: 0,5 − 0,3
|
||
self.assertEqual(self.partes("comision"), []) # la comisión no es solo tuya
|
||
|
||
def test_no_apuntar_resuelve_la_parte_sin_crear_nada(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
r = self.decidir(self.partes(), accion="no_apuntar")
|
||
self.assertEqual((r["guardadas"], r["creados"]), (1, 0))
|
||
self.assertEqual((self.movs(), self.partes()), ([], []))
|
||
|
||
def test_si_una_falla_las_demas_se_guardan(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.4999 * SATOSHI), AJENA)]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
# Sin apuntar la entrada, vender 0,4999 no puede ser: no había saldo
|
||
r = self.decidir(self.partes("salida"), accion="venta", importe=1000)
|
||
self.assertEqual(r["guardadas"], 0)
|
||
self.assertIn("cantidad", r["fallos"][0]["motivo"])
|
||
r = self.decidir(self.partes("comision"), accion="no_apuntar")
|
||
self.assertEqual(r["guardadas"], 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|
||
|
||
class TestARatos(ConHistorial):
|
||
def test_lo_leido_se_guarda_y_se_puede_seguir_sin_el_nodo(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
primera = self.historial()
|
||
self.nodo.parar() # sin nodo
|
||
segunda = self.app.btc_historial({"leer": False})
|
||
self.assertEqual([p["id"] for p in segunda["partes"]], [p["id"] for p in primera["partes"]])
|
||
self.assertEqual(segunda["servidor"], None)
|
||
self.assertEqual(btc.ruta_config(self.carpeta).with_name("bitcoin_cache.json").stat().st_mode & 0o777, 0o600)
|
||
|
||
def test_las_transacciones_confirmadas_no_se_vuelven_a_pedir(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.historial()
|
||
antes = self.nodo.peticiones.count("blockchain.transaction.get")
|
||
self.paga(ZPUB, "p2wpkh", 1, 0.1, 200)
|
||
self.assertEqual(len(self.partes()), 2)
|
||
self.assertEqual(self.nodo.peticiones.count("blockchain.transaction.get") - antes, 1) # solo la nueva
|
||
|
||
def test_si_cambian_los_monederos_hay_que_volver_a_leer(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.historial()
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"},
|
||
{"nombre": "Móvil", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C1"}])
|
||
with self.assertRaises(btc.ErrorBitcoin) as e:
|
||
self.app.btc_historial({"leer": False})
|
||
self.assertIn("volver a leer", str(e.exception))
|
||
|
||
def test_de_la_mineria_solo_se_guardan_tus_salidas_y_se_gastan_bien(self):
|
||
cb = self.cadena.añadir(construir([], [(5 * SATOSHI, AJENA), (321000, script(ZPUB, "p2wpkh", 0, 0))],
|
||
coinbase=True), 100)
|
||
self.cadena.añadir(construir([(cb, 1)], [(320000, script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in partes], [("mineria", 0.00321), ("comision", 0.00001)])
|
||
from cartera import cache_btc
|
||
guardada = cache_btc.cargar(self.carpeta)["txs"][cb]
|
||
self.assertEqual(len(guardada["salidas"]), 1)
|
||
self.assertTrue(guardada["recortada"])
|
||
|
||
def test_lo_decidido_sin_guardar_se_conserva(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.3 * SATOSHI), AJENA),
|
||
(round(0.1999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
salida = self.partes("salida")[0]
|
||
self.app.btc_borradores({"borradores": {salida["id"]: {"accion": "venta", "importe": 27000}}})
|
||
otra_vez = self.app.btc_historial({"leer": False})
|
||
salida = next(p for p in otra_vez["partes"] if p["clase"] == "salida")
|
||
self.assertEqual(salida["borrador"], {"accion": "venta", "importe": 27000})
|
||
self.decidir(self.partes("entrada"), importe=40000)
|
||
self.decidir([salida], accion="venta", importe=27000)
|
||
self.assertEqual(btc.cargar(self.carpeta)["borradores"], {})
|
||
self.assertEqual(self.historial()["progreso"], {"total": 3, "hechas": 2})
|
||
|
||
def test_si_al_anadir_un_monedero_una_decision_deja_de_encajar_se_avisa(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.2 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.2999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.decidir(self.partes("entrada"), importe=40000)
|
||
self.decidir(self.partes("salida"), accion="venta", importe=18000) # sin el otro monedero parecía una venta
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"},
|
||
{"nombre": "Frío", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C1"}])
|
||
r = self.historial()
|
||
self.assertEqual(len(r["cambiadas"]), 1)
|
||
vieja = r["cambiadas"][0]["antes"][0]
|
||
self.assertEqual((vieja["id"].split(":")[1], vieja["como"]), ("salida", "venta"))
|
||
d = self.app.btc_deshacer_parte({"id": vieja["id"]}["id"])
|
||
self.assertEqual(d["borrados"], 1)
|
||
self.assertFalse([m for m in self.movs() if m["tipo"] == "Venta"])
|
||
r = self.historial()
|
||
self.assertEqual(r["cambiadas"], [])
|
||
|
||
|
||
class TestMonederoCompartido(ConHistorial):
|
||
"""Un monedero del que solo la mitad es tuya: lo del otro dueño no entra en la cartera."""
|
||
|
||
COMUN = {"nombre": "Común", "clave": ZPUB, "activo": "C1", "parte": 0.5}
|
||
MIO = {"nombre": "Frío", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C1"}
|
||
|
||
def saldos(self):
|
||
from cartera import lugares
|
||
return lugares.saldos(self.movs(), "C1")
|
||
|
||
def lugar(self, nombre):
|
||
return next(x["id"] for x in self.cartera.activo("C1")["lugares"] if x["nombre"] == nombre)
|
||
|
||
def cuadres(self):
|
||
return {c["nombre_activo"]: (c["en_nodo"], c["apuntado"]) for c in self.historial()["cuadres"]}
|
||
|
||
def test_de_lo_que_llega_solo_se_apunta_tu_parte(self):
|
||
self.cadena.añadir(construir([], [(1_000_000, script(ZPUB, "p2wpkh", 0, 0)), (SATOSHI, AJENA)],
|
||
coinbase=True), 300)
|
||
self.paga(ZPUB, "p2wpkh", 1, 0.2, 310) # un pago de la pool, no de bloque
|
||
self.config([self.COMUN])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in partes], [("mineria", 0.005), ("entrada", 0.1)])
|
||
r = self.historial()
|
||
self.assertEqual(r["cuadres"][0]["parte"], 0.5)
|
||
self.assertEqual(self.cuadres(), {"Bitcoin · Común": (0.105, 0.0)}) # la mitad de lo que ve el nodo
|
||
self.decidir(partes, accion="mineria")
|
||
self.assertEqual(self.cuadres(), {"Bitcoin · Común": (0.105, 0.105)})
|
||
|
||
def test_medio_satoshi_no_descuadra(self):
|
||
for i, sats in enumerate((100_001, 200_003, 300_005)): # tres pagos impares
|
||
self.cadena.añadir(construir([], [(sats, script(ZPUB, "p2wpkh", 0, i))], coinbase=True), 300 + i)
|
||
self.config([self.COMUN])
|
||
self.decidir(self.partes())
|
||
en_nodo, apuntado = self.cuadres()["Bitcoin · Común"]
|
||
self.assertEqual(en_nodo, 0.003000045)
|
||
self.assertLess(abs(en_nodo - apuntado), 1e-9)
|
||
|
||
def test_un_reparto_a_medias_es_un_traslado_y_lo_del_otro_no_aparece(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 1.0, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.4999 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.4999 * SATOSHI), AJENA)]), 200)
|
||
self.config([self.COMUN, self.MIO])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["accion"], p["cantidad"]) for p in partes],
|
||
[("entrada", "compra", 0.5), ("interno", "traslado", 0.4999), ("comision", "comision", 0.0001)])
|
||
self.decidir(partes[:1], accion="mineria")
|
||
self.assertEqual(self.decidir(partes[1:])["fallos"], [])
|
||
self.assertEqual(sorted(m["tipo"] for m in self.movs()), ["Comision", "Mineria", "Traslado"]) # ni venta ni regalo
|
||
self.assertEqual(self.saldos(), {self.lugar("Frío"): 0.4999})
|
||
self.assertEqual(self.cuadres(), {"Bitcoin · Común": (0.0, 0.0), "Bitcoin · Frío": (0.4999, 0.4999)})
|
||
|
||
def test_un_reparto_con_vuelta_al_monedero_comun(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 1.0, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.2 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.2 * SATOSHI), AJENA),
|
||
(round(0.5998 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([self.COMUN, self.MIO])
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in partes],
|
||
[("entrada", 0.5), ("interno", 0.2), ("comision", 0.0001)])
|
||
self.decidir(partes[:1], accion="mineria")
|
||
self.decidir(partes[1:])
|
||
self.assertEqual(self.cuadres(), {"Bitcoin · Común": (0.2999, 0.2999), "Bitcoin · Frío": (0.2, 0.2)})
|
||
|
||
def test_un_pago_a_un_tercero_es_tuyo_solo_por_tu_parte(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 1.0, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.3 * SATOSHI), AJENA),
|
||
(round(0.6998 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([self.COMUN])
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in self.partes()],
|
||
[("entrada", 0.5), ("salida", 0.15), ("comision", 0.0001)])
|
||
|
||
def test_un_reparto_que_no_es_a_medias_se_propone_por_el_efecto_neto(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 1.0, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.7 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.2998 * SATOSHI), AJENA)]), 200)
|
||
self.config([self.COMUN, self.MIO])
|
||
reparto = [p for p in self.partes() if p["fecha"] == fecha_de(200)]
|
||
self.assertEqual(sorted((p["clase"], p["lugar"], p["cantidad"], p["mixta"]) for p in reparto),
|
||
[("entrada", self.lugar("Frío"), 0.7, True), ("salida", self.lugar("Común"), 0.5, True)])
|
||
|
||
def test_al_marcarlo_compartido_lo_ya_guardado_se_avisa_y_se_deshace_de_una_vez(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 1.0, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.4999 * SATOSHI), script(XPUB_TAPROOT, "p2tr", 0, 0)),
|
||
(round(0.4999 * SATOSHI), AJENA)]), 200)
|
||
self.config([{**self.COMUN, "parte": 1}, self.MIO])
|
||
partes = self.partes()
|
||
self.decidir([p for p in partes if p["clase"] == "entrada"], accion="mineria")
|
||
self.decidir([p for p in partes if p["clase"] == "salida"], accion="pago", importe=40000) # el «regalo»
|
||
self.decidir([p for p in partes if p["clase"] in ("interno", "comision")])
|
||
self.assertEqual(self.historial()["cambiadas"], [])
|
||
c = btc.cargar(self.carpeta)
|
||
c["monederos"] = [btc.normalizar_monedero({"id": m["id"], "clave": m["clave"], "nombre": m["nombre"],
|
||
"activo": "C1", "tipo": m["tipo"],
|
||
"parte": 0.5 if m["nombre"] == "Común" else None}, c["monederos"])
|
||
for m in c["monederos"]]
|
||
btc.guardar(self.carpeta, c)
|
||
cambiadas = self.historial()["cambiadas"]
|
||
antes = sorted((a["id"].split(":")[1], a["como"]) for x in cambiadas for a in x["antes"])
|
||
# el traslado sigue valiendo; la entrada y la comisión eran el doble, y el «regalo» sobra
|
||
self.assertEqual(antes, [("comision", "comision"), ("entrada", "mineria"), ("salida", "pago")])
|
||
r = self.app.btc_deshacer_partes([a["id"] for x in cambiadas for a in x["antes"]])
|
||
self.assertEqual((r["partes"], r["borrados"]), (3, 3))
|
||
self.assertEqual(self.historial()["cambiadas"], [])
|
||
self.decidir(self.partes("entrada"), accion="mineria")
|
||
self.decidir(self.partes("comision"))
|
||
self.assertEqual(self.partes(), [])
|
||
self.assertEqual(sorted(m["tipo"] for m in self.movs()), ["Comision", "Mineria", "Traslado"])
|
||
self.assertEqual(self.cuadres(), {"Bitcoin · Común": (0.0, 0.0), "Bitcoin · Frío": (0.4999, 0.4999)})
|
||
|
||
def test_la_parte_tiene_que_ser_un_porcentaje_con_sentido(self):
|
||
for mala in (0, 1.5, "mitad"):
|
||
with self.assertRaises(btc.ErrorBitcoin):
|
||
btc.normalizar_monedero({**self.COMUN, "parte": mala}, [])
|
||
self.assertEqual(btc.normalizar_monedero({k: v for k, v in self.COMUN.items() if k != "parte"}, [])["parte"], 1.0)
|
||
|
||
|
||
class TestTraza(ConHistorial):
|
||
def test_una_consolidacion_reparte_sus_origenes_a_prorrata(self):
|
||
compra = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
cb = self.cadena.añadir(construir([], [(10_000_000, script(ZPUB, "p2wpkh", 0, 1)), (SATOSHI, AJENA)],
|
||
coinbase=True), 150)
|
||
junta = self.cadena.añadir(construir([(compra, 0), (cb, 0)],
|
||
[(59_990_000, script(XPUB_TAPROOT, "p2tr", 0, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"},
|
||
{"nombre": "Frío", "clave": XPUB_TAPROOT, "tipo": "p2tr", "activo": "C1"}])
|
||
self.decidir(self.partes("entrada"), importe=40000)
|
||
r = self.app.btc_monedas()
|
||
self.assertEqual(len(r["monedas"]), 1)
|
||
m = r["monedas"][0]
|
||
self.assertEqual((m["txid"], m["cantidad"], m["monedero"]), (junta, 0.5999, "Frío"))
|
||
fr = {o["clave"]: o["fraccion"] for o in m["origenes"]}
|
||
self.assertAlmostEqual(fr[compra], 5 / 6, places=5)
|
||
self.assertAlmostEqual(fr[cb], 1 / 6, places=5)
|
||
self.assertEqual(r["origenes"][compra]["importe"], 40000)
|
||
self.assertEqual(r["origenes"][compra]["como"], "compra")
|
||
self.assertEqual(r["origenes"][cb]["clase"], "mineria")
|
||
t = self.app.btc_traza({"txid": junta, "indice": 0})
|
||
self.assertEqual(t["arbol"]["clase"], "paso")
|
||
self.assertEqual(sorted(h["clase"] for h in t["arbol"]["hijos"]), ["mineria", "origen"])
|
||
self.assertEqual(sum(o["sats"] for o in t["origenes"]), 59_990_000)
|
||
|
||
def test_el_cambio_de_un_pago_sigue_siendo_de_su_compra(self):
|
||
compra = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
pago = self.cadena.añadir(construir([(compra, 0)], [(round(0.3 * SATOSHI), AJENA),
|
||
(round(0.1999 * SATOSHI), script(ZPUB, "p2wpkh", 1, 0))]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.historial()
|
||
m = self.app.btc_monedas()["monedas"]
|
||
self.assertEqual([(x["txid"], x["indice"]) for x in m], [(pago, 1)])
|
||
self.assertEqual(m[0]["origenes"], [{"clave": compra, "fraccion": 1.0}])
|
||
|
||
def test_la_parte_de_terceros_de_un_coinjoin_se_ve(self):
|
||
mia = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
ajeno = self.cadena.añadir(construir([("cc" * 32, 0)], [(SATOSHI // 2, AJENA)]), 101)
|
||
cj = self.cadena.añadir(construir([(mia, 0), (ajeno, 0)],
|
||
[(SATOSHI // 2, script(ZPUB, "p2wpkh", 1, 0)), (SATOSHI // 2, AJENA)]), 200)
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.historial()
|
||
fr = {o["clave"]: o["fraccion"] for o in self.app.btc_monedas()["monedas"][0]["origenes"]}
|
||
self.assertAlmostEqual(fr[mia], 0.5)
|
||
self.assertAlmostEqual(fr[f"ajeno:{cj}"], 0.5)
|
||
|
||
def test_sin_haber_leido_el_nodo_lo_dice_en_claro(self):
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
with self.assertRaises(btc.ErrorBitcoin) as e:
|
||
self.app.btc_monedas()
|
||
self.assertIn("Leer el nodo", str(e.exception))
|
||
|
||
|
||
class TestDireccionesSueltas(ConHistorial):
|
||
"""Monederos sin clave extendida (Bolsillo): se vigilan sus direcciones una a una."""
|
||
|
||
def direccion(self, cambio, indice):
|
||
from cartera import bitcoin
|
||
return bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", cambio, indice, 1)[0]["direccion"]
|
||
|
||
def test_lee_cualquier_tipo_de_direccion(self):
|
||
from cartera import bitcoin
|
||
for d, empieza in (("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa", "76a914"),
|
||
("3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy", "a914"),
|
||
("bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4", "0014"),
|
||
("bc1qrp33g0q5c5txsp9arysrx4k6zdkfs4nce4xj0gdcccefvpysxf3qccfmv3", "0020"),
|
||
("bc1p0xlxvlhemja6c4dqv22uapctqupfhlxm9h8z3k2e72q4k9hcz7vqzk5jj0", "5120")):
|
||
script, red = bitcoin.script_de_direccion(d)
|
||
self.assertEqual(red, "mainnet")
|
||
self.assertTrue(script.hex().startswith(empieza), d)
|
||
self.assertEqual(bitcoin.script_de_direccion(self.direccion(0, 0))[0].hex(), script_hex(0, 0))
|
||
for mala in ("bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t5", "hola", "BC1QW508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4"):
|
||
with self.assertRaises(ValueError):
|
||
bitcoin.script_de_direccion(mala)
|
||
|
||
def test_alta_limpia_la_lista_y_avisa_de_erratas(self):
|
||
d0 = self.direccion(0, 0)
|
||
m = btc.normalizar_monedero({"nombre": "Bolsillo", "activo": "C1",
|
||
"direcciones": f"bitcoin:{d0.upper()}?amount=0.1\n{d0}, {self.direccion(0, 1)}"}, [])
|
||
self.assertEqual((m["tipo"], m["direcciones"]), ("sueltas", [d0, self.direccion(0, 1)]))
|
||
self.assertNotIn("clave", m)
|
||
with self.assertRaisesRegex(btc.ErrorBitcoin, "Revisa"):
|
||
btc.normalizar_monedero({"activo": "C1", "direcciones": d0[:-1] + "x"}, [])
|
||
with self.assertRaisesRegex(btc.ErrorBitcoin, "ya está en «Bolsillo»"):
|
||
btc.normalizar_monedero({"activo": "C1", "direcciones": d0}, [m])
|
||
with self.assertRaisesRegex(btc.ErrorBitcoin, "al menos una"):
|
||
btc.normalizar_monedero({"activo": "C1", "direcciones": " "}, [])
|
||
|
||
def test_sin_la_de_cambio_parece_que_se_fue_y_al_anadirla_cuadra(self):
|
||
origen = self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.cadena.añadir(construir([(origen, 0)], [(round(0.4 * SATOSHI), AJENA),
|
||
(round(0.0999 * SATOSHI), script_hex(1, 0))]), 200)
|
||
self.config([{"nombre": "Bolsillo", "activo": "C1", "direcciones": self.direccion(0, 0)}])
|
||
salida = self.partes("salida")
|
||
self.assertAlmostEqual(salida[0]["cantidad"], 0.4999, places=8) # la vuelta parece ajena (+ comisión)
|
||
self.assertEqual(self.historial()["cuadres"][0]["en_nodo"], 0.0)
|
||
|
||
# Se edita desde la aplicación pegando la lista entera, con la de cambio
|
||
mid = btc.cargar(self.carpeta)["monederos"][0]["id"]
|
||
estado = self.app.btc_monedero({"monedero": {
|
||
"id": mid, "nombre": "Bolsillo", "activo": "C1",
|
||
"direcciones": f"{self.direccion(0, 0)}\n{self.direccion(1, 0)}"}})
|
||
self.assertEqual(estado["monederos"][0]["clave_corta"], "2 direcciones sueltas")
|
||
# editar sin mandar direcciones las conserva
|
||
self.app.btc_monedero({"monedero": {"id": mid, "nombre": "Bolsillo (móvil)", "activo": "C1"}})
|
||
self.assertEqual(len(btc.cargar(self.carpeta)["monederos"][0]["direcciones"]), 2)
|
||
|
||
partes = self.historial()["partes"]
|
||
self.assertEqual([(p["clase"], round(p["cantidad"], 8)) for p in partes],
|
||
[("entrada", 0.5), ("salida", 0.4), ("comision", 0.0001)])
|
||
self.assertAlmostEqual(self.historial()["cuadres"][0]["en_nodo"], 0.0999, places=8)
|
||
|
||
def test_un_monedero_de_direcciones_sueltas_tambien_puede_ser_compartido(self):
|
||
self.cadena.añadir(construir([], [(500_000, script(ZPUB, "p2wpkh", 0, 0))], coinbase=True), 300)
|
||
self.config([{"nombre": "Bolsillo", "activo": "C1", "direcciones": self.direccion(0, 0), "parte": "0,5"}])
|
||
self.assertEqual(btc.cargar(self.carpeta)["monederos"][0]["parte"], 0.5)
|
||
partes = self.partes()
|
||
self.assertEqual([(p["clase"], p["cantidad"]) for p in partes], [("mineria", 0.0025)])
|
||
self.decidir(partes)
|
||
c = self.historial()["cuadres"][0]
|
||
self.assertEqual((c["en_nodo"], c["apuntado"], c["parte"]), (0.0025, 0.0025, 0.5))
|
||
# y la comprobación diaria cuadra contra esa misma mitad
|
||
self.assertEqual(self.app.btc_comprobar()["ultima"]["cuadres"][0]["diferencia"], 0.0)
|
||
|
||
def test_la_comprobacion_diaria_tambien_lo_lee(self):
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.25, 100)
|
||
self.config([{"nombre": "Bolsillo", "activo": "C1", "direcciones": self.direccion(0, 0)}])
|
||
r = self.app.btc_comprobar()
|
||
self.assertAlmostEqual(r["cuadres"][0]["en_nodo"], 0.25, places=8)
|
||
|
||
|
||
def script_hex(cambio, indice):
|
||
return script(ZPUB, "p2wpkh", cambio, indice)
|
||
|
||
|
||
class TestRevisarMonederos(ConHistorial):
|
||
"""«Comprobar qué tiene cada monedero»: los cuatro tipos de una clave y las direcciones sin uso."""
|
||
|
||
def test_dice_en_que_tipo_hay_transacciones_y_que_direcciones_no_tienen(self):
|
||
from cartera import bitcoin
|
||
self.paga(ZPUB, "p2wpkh", 0, 0.5, 100)
|
||
self.paga(ZPUB, "p2wpkh", 1, 0.1, 110, cambio=1)
|
||
sin_uso = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 5, 1)[0]["direccion"]
|
||
usada = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 0, 1)[0]["direccion"]
|
||
self.config([{"nombre": "Mal tipo", "clave": ZPUB, "tipo": "p2tr", "activo": "C1"},
|
||
{"nombre": "Bolsillo", "direcciones": f"{usada}\n{sin_uso}", "activo": "C2"}])
|
||
estado = self.app.btc_revisar_monederos()
|
||
mal, bolsillo = estado["monederos"]
|
||
r = mal["revision"]
|
||
self.assertEqual(r["elegido"], "p2tr")
|
||
self.assertEqual({t: v["usadas"] for t, v in r["tipos"].items()},
|
||
{"p2wpkh": 2, "p2tr": 0, "p2sh-p2wpkh": 0, "p2pkh": 0})
|
||
self.assertEqual((bolsillo["revision"]["usadas"], bolsillo["revision"]["total"], bolsillo["revision"]["sin_uso"]),
|
||
(1, 2, [sin_uso]))
|
||
# Se guarda, y se conserva al editar solo el nombre; al cambiar la clave, no
|
||
mid = btc.cargar(self.carpeta)["monederos"][0]["id"]
|
||
self.app.btc_monedero({"monedero": {"id": mid, "nombre": "Otro nombre", "activo": "C1", "tipo": "p2tr"}})
|
||
self.assertIsNotNone(btc.cargar(self.carpeta)["monederos"][0]["revision"])
|
||
|
||
def test_al_unir_lugares_el_monedero_y_el_exchange_recordado_pasan_al_que_se_queda(self):
|
||
self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}])
|
||
self.app._lugares_de_monederos(btc.cargar(self.carpeta))
|
||
sparrow = btc.cargar(self.carpeta)["monederos"][0]["lugar"]
|
||
otro = self.cartera.guardar_lugar("C1", {"nombre": "Sparrow portátil", "clase": "monedero"})["id"]
|
||
ajustes = self.app._ajustes()
|
||
ajustes["destinos_exchange"] = {"Ledger": sparrow}
|
||
self.app._guardar_ajustes(ajustes)
|
||
r = self.app.unir_lugares({"activo": "C1", "desde": sparrow, "hacia": otro})
|
||
self.assertEqual(r["monederos"], 1)
|
||
self.assertEqual(btc.cargar(self.carpeta)["monederos"][0]["lugar"], otro)
|
||
self.assertEqual(self.app._ajustes()["destinos_exchange"], {"Ledger": otro})
|
||
|