433 lines
24 KiB
Python
433 lines
24 KiB
Python
"""
|
|
Pruebas de la fase 3: fusión de copias y sincronización cifrada.
|
|
|
|
Usan GPG de verdad con una clave desechable (sin contraseña) en un GNUPGHOME
|
|
temporal y un repositorio git local en lugar del servidor. Nada sale del equipo.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import datos as D # noqa: E402
|
|
from cartera.datos import Datos, escribir, fusionar, huella_datos, leer, version_de # noqa: E402
|
|
from cartera.operaciones import Cartera # noqa: E402
|
|
from cartera.restaurar import restaurar # noqa: E402
|
|
from cartera.sincronizar import (CIFRADO, ErrorSync, Sincronizador, # noqa: E402
|
|
elegir_ajustes, escribir_ajuste, leer_ajustes)
|
|
|
|
|
|
def activo(aid, **kw):
|
|
return {"id": aid, "nombre": f"Activo {aid}", "tipo": "Fondo", "estado": "abierta",
|
|
"modificado": "2026-01-01T00:00:00Z", **kw}
|
|
|
|
|
|
def mov(mid, **kw):
|
|
return {"id": mid, "fecha": "2026-01-01", "activo": "F1", "tipo": "Compra", "cantidad": 1.0,
|
|
"precio": 10.0, "modificado": "2026-01-01T00:00:00Z", **kw}
|
|
|
|
|
|
class TestFusion(unittest.TestCase):
|
|
def test_une_actualiza_y_respeta_borrados(self):
|
|
local = Datos([activo("F1")], [mov("a"), mov("b", borrado=True, modificado="2026-03-01T00:00:00Z")], [], [])
|
|
remoto = Datos([activo("F1", nombre="Renombrado", modificado="2026-02-01T00:00:00Z")],
|
|
[mov("b", cantidad=5.0, modificado="2026-02-01T00:00:00Z"), mov("c")], [], [])
|
|
r, cuenta = fusionar(local, remoto)
|
|
self.assertEqual(r.activos[0]["nombre"], "Renombrado")
|
|
self.assertEqual({m["id"]: m.get("borrado", False) for m in r.movimientos},
|
|
{"a": False, "b": True, "c": False}) # el borrado (más reciente) gana
|
|
self.assertEqual(cuenta, {"de_remoto": 1, "de_local": 1, "actualizados_desde_remoto": 1,
|
|
"actualizados_desde_local": 1})
|
|
|
|
def test_simetrica_e_idempotente(self):
|
|
a = Datos([activo("F1", nombre="A")], [mov("x", cantidad=2.0)], [], [])
|
|
b = Datos([activo("F1", nombre="B")], [mov("x", cantidad=3.0), mov("y")], [], []) # mismo «modificado»
|
|
ab, _ = fusionar(a, b)
|
|
ba, _ = fusionar(b, a)
|
|
self.assertEqual(huella_datos(ab), huella_datos(ba))
|
|
otra, cuenta = fusionar(ab, ab)
|
|
self.assertEqual(huella_datos(otra), huella_datos(ab))
|
|
self.assertEqual(sum(cuenta.values()), 0)
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("gpg") and shutil.which("git"), "hace falta gpg y git")
|
|
class TestSincronizacion(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp = Path(tempfile.mkdtemp(prefix="sync-", dir="/tmp"))
|
|
cls.gnupg = cls.tmp / "g"
|
|
cls.gnupg.mkdir(mode=0o700)
|
|
cls._gnupg_antes = os.environ.get("GNUPGHOME")
|
|
os.environ["GNUPGHOME"] = str(cls.gnupg)
|
|
subprocess.run(["gpg", "--batch", "--passphrase", "", "--quick-gen-key", "Prueba Cartera <prueba@local>",
|
|
"default", "default", "never"], check=True, capture_output=True)
|
|
cls.remoto = cls.tmp / "remoto.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(cls.remoto)], check=True)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
subprocess.run(["gpgconf", "--kill", "gpg-agent"], capture_output=True)
|
|
if cls._gnupg_antes is None:
|
|
os.environ.pop("GNUPGHOME", None)
|
|
else:
|
|
os.environ["GNUPGHOME"] = cls._gnupg_antes
|
|
shutil.rmtree(cls.tmp, ignore_errors=True)
|
|
|
|
def equipo(self, nombre: str, datos: Datos) -> tuple[Cartera, Sincronizador]:
|
|
base = self.tmp / nombre
|
|
escribir(base / "datos", datos)
|
|
cartera = Cartera(base / "datos")
|
|
self.addCleanup(cartera.cerrar)
|
|
sync = Sincronizador(base / "sync")
|
|
claves = sync.claves()
|
|
self.assertEqual(len(claves), 1)
|
|
self.assertFalse(claves[0]["en_tarjeta"]) # clave de prueba en disco, no en una tarjeta
|
|
sync.configurar(str(self.remoto), claves[0]["huella"])
|
|
return cartera, sync
|
|
|
|
def test_dos_equipos_se_sincronizan_cifrado(self):
|
|
usb, s_usb = self.equipo("usb", Datos([activo("F1")], [mov("a")], [], []))
|
|
self.assertTrue(s_usb.estado(usb.datos)["pendiente"])
|
|
|
|
r = s_usb.sincronizar(usb) # primera subida: no hay nada que descifrar
|
|
self.assertEqual((r["descifrado"], r["subido"]), (False, True))
|
|
self.assertFalse(s_usb.estado(usb.datos)["pendiente"])
|
|
contenido = (s_usb.repo / CIFRADO).read_bytes()
|
|
self.assertNotIn(b"Activo F1", contenido) # en el servidor solo hay cifrado
|
|
self.assertFalse(s_usb.sincronizar(usb)["subido"]) # sin cambios: nada que hacer
|
|
|
|
otro, s_otro = self.equipo("otro", Datos()) # equipo nuevo vacío: restaura
|
|
r = s_otro.sincronizar(otro)
|
|
self.assertTrue(r["descifrado"] and r["cambios_locales"])
|
|
self.assertEqual(huella_datos(leer(self.tmp / "otro" / "datos")), huella_datos(usb.datos))
|
|
|
|
copias = []
|
|
otro.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2026-02-01", "cantidad": 2, "precio": 11})
|
|
self.assertTrue(s_otro.sincronizar(otro, copia_seguridad=lambda: copias.append(1))["subido"])
|
|
usb.guardar_movimiento({"activo": "F1", "tipo": "Venta", "fecha": "2026-03-01", "cantidad": 1, "precio": 12})
|
|
r = s_usb.sincronizar(usb, copia_seguridad=lambda: copias.append(1))
|
|
self.assertTrue(r["descifrado"] and r["cambios_locales"] and r["subido"])
|
|
self.assertEqual(len(usb.datos.vivos("movimientos")), 3)
|
|
s_otro.sincronizar(otro)
|
|
self.assertEqual(huella_datos(otro.datos), huella_datos(usb.datos))
|
|
self.assertEqual(copias, [1]) # copia de seguridad antes de tocar el USB
|
|
|
|
comprobacion = s_usb.comprobar(usb.datos)
|
|
self.assertTrue(comprobacion["identica"])
|
|
|
|
def test_por_git_dos_inversiones_distintas_con_el_mismo_numero_no_se_funden(self):
|
|
self.remoto = self.tmp / "choque.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(self.remoto)], check=True)
|
|
a, s_a = self.equipo("choque_a", Datos())
|
|
s_a.sincronizar(a)
|
|
b, s_b = self.equipo("choque_b", Datos())
|
|
s_b.sincronizar(b)
|
|
fa = a.guardar_activo({"tipo": "Fondo", "nombre": "Fondo del USB"})
|
|
fb = b.guardar_activo({"tipo": "Fondo", "nombre": "Fondo del servidor"})
|
|
self.assertEqual(fa["id"], fb["id"]) # el mismo número en las dos copias
|
|
a.guardar_movimiento({"activo": fa["id"], "tipo": "Compra", "fecha": "2026-02-01", "cantidad": 1, "precio": 10})
|
|
s_a.sincronizar(a)
|
|
s_b.sincronizar(b)
|
|
s_a.sincronizar(a)
|
|
nombres = sorted(x["nombre"] for x in b.datos.vivos("activos"))
|
|
self.assertEqual(nombres, ["Fondo del USB", "Fondo del servidor"])
|
|
self.assertEqual(huella_datos(a.datos), huella_datos(b.datos))
|
|
compra = b.datos.vivos("movimientos")[0]
|
|
self.assertEqual(next(x["nombre"] for x in b.datos.activos if x["id"] == compra["activo"]), "Fondo del USB")
|
|
|
|
def test_usb_nuevo_recupera_del_servidor(self):
|
|
remoto = self.tmp / "restaurar.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(remoto)], check=True)
|
|
vacio = self.tmp / "nuevo" / "datos"
|
|
with self.assertRaises(ErrorSync): # aún no hay copia en el servidor
|
|
restaurar(vacio, str(remoto), sync_dir=self.tmp / "nuevo0" / "sync")
|
|
|
|
usb, s_usb = self.equipo("usb-r", Datos([activo("F1")], [mov("a"), mov("b")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
s_usb.sincronizar(usb)
|
|
|
|
r = restaurar(vacio, str(remoto)) # clave elegida sola
|
|
self.assertEqual(r["registros"]["movimientos"], 2)
|
|
self.assertEqual(huella_datos(leer(vacio)), huella_datos(usb.datos))
|
|
with self.assertRaises(ErrorSync): # nunca sobrescribe
|
|
restaurar(vacio, str(remoto))
|
|
|
|
antes = os.getcwd() # ruta relativa, como abrir_app.sh
|
|
os.chdir(self.tmp)
|
|
self.addCleanup(os.chdir, antes)
|
|
restaurar(Path("relativa/datos"), str(remoto))
|
|
self.assertEqual(huella_datos(leer(self.tmp / "relativa" / "datos")), huella_datos(usb.datos))
|
|
os.chdir(antes)
|
|
|
|
# la app del USB nuevo arranca en sincronía: no descifra ni sube nada
|
|
nuevo = Cartera(vacio)
|
|
self.addCleanup(nuevo.cerrar)
|
|
s_nuevo = Sincronizador(vacio.parent / "sync")
|
|
self.assertFalse(s_nuevo.estado(nuevo.datos)["pendiente"])
|
|
self.assertEqual((s_nuevo.sincronizar(nuevo)["descifrado"], s_nuevo.sincronizar(nuevo)["subido"]),
|
|
(False, False))
|
|
|
|
usb.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2026-02-01", "cantidad": 1, "precio": 9})
|
|
simulacro = restaurar(self.tmp / "x" / "datos", str(remoto), comparar_con=self.tmp / "usb-r" / "datos")
|
|
self.assertEqual(simulacro["diferencias"], 1) # el cambio aún no sincronizado
|
|
self.assertFalse((self.tmp / "x" / "datos").exists()) # el simulacro no escribe
|
|
|
|
def test_los_ajustes_viajan_cifrados_y_llegan_al_otro_equipo(self):
|
|
remoto = self.tmp / "ajustes.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(remoto)], check=True)
|
|
usb, s_usb = self.equipo("usb-a", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
escribir_ajuste(self.tmp / "usb-a" / "bitcoin.json", config_btc([monedero("w1", ["tx1"])]))
|
|
escribir_ajuste(self.tmp / "usb-a" / "extractos.json",
|
|
{"cuenta_intereses": {"trade_republic": "CC1"}, "modificado": "2026-01-01T00:00:00"})
|
|
self.assertTrue(s_usb.estado(usb.datos)["pendiente"]) # los ajustes cuentan como cambio
|
|
|
|
s_usb.sincronizar(usb)
|
|
self.assertFalse(s_usb.estado(usb.datos)["pendiente"])
|
|
self.assertNotIn(b"zpub-w1", (s_usb.repo / CIFRADO).read_bytes()) # en el servidor, cifrado
|
|
|
|
otro, s_otro = self.equipo("otro-a", Datos())
|
|
s_otro.configurar(str(remoto), s_otro.config()["huella"])
|
|
r = s_otro.sincronizar(otro)
|
|
self.assertEqual(sorted(r["ajustes"]), ["bitcoin.json", "extractos.json"])
|
|
recibido = leer_ajustes(self.tmp / "otro-a")
|
|
self.assertEqual(recibido["bitcoin.json"]["monederos"][0]["clave"], "zpub-w1")
|
|
self.assertEqual(recibido["extractos.json"]["cuenta_intereses"], {"trade_republic": "CC1"})
|
|
self.assertEqual((self.tmp / "otro-a" / "bitcoin.json").stat().st_mode & 0o777, 0o600)
|
|
|
|
# Un monedero nuevo en el otro equipo llega al USB, y borrarlo allí no lo resucita
|
|
escribir_ajuste(self.tmp / "otro-a" / "bitcoin.json",
|
|
config_btc([monedero("w1", ["tx1"]), monedero("w2")], "2026-06-01T00:00:00"))
|
|
s_otro.sincronizar(otro)
|
|
s_usb.sincronizar(usb)
|
|
self.assertEqual([m["id"] for m in leer_ajustes(self.tmp / "usb-a")["bitcoin.json"]["monederos"]],
|
|
["w1", "w2"])
|
|
escribir_ajuste(self.tmp / "usb-a" / "bitcoin.json",
|
|
config_btc([monedero("w1", ["tx1"])], "2026-07-01T00:00:00"))
|
|
s_usb.sincronizar(usb)
|
|
s_otro.sincronizar(otro)
|
|
self.assertEqual([m["id"] for m in leer_ajustes(self.tmp / "otro-a")["bitcoin.json"]["monederos"]], ["w1"])
|
|
|
|
comprobacion = s_usb.comprobar(usb.datos)
|
|
self.assertTrue(comprobacion["identica"])
|
|
self.assertTrue(comprobacion["ajustes"]["bitcoin.json"]["en_la_copia"])
|
|
|
|
def repositorio(self, nombre: str) -> Path:
|
|
remoto = self.tmp / f"{nombre}.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(remoto)], check=True)
|
|
return remoto
|
|
|
|
def test_una_copia_de_formato_antiguo_se_migra_antes_de_fusionar(self):
|
|
remoto = self.repositorio("formato-viejo")
|
|
viejo, s_viejo = self.equipo("app-vieja", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_viejo.configurar(str(remoto), s_viejo.config()["huella"])
|
|
s_viejo.sincronizar(viejo) # sube en el formato actual
|
|
|
|
def moneda(tablas):
|
|
for m in tablas["movimientos"]:
|
|
m.setdefault("moneda", "EUR")
|
|
siguiente = D.VERSION + 1
|
|
with mock.patch.object(D, "VERSION", siguiente), mock.patch.dict(D.MIGRACIONES, {siguiente: moneda}):
|
|
nuevo, s_nuevo = self.equipo("app-nueva", Datos([activo("F1")], [mov("b", moneda="USD")], [], []))
|
|
s_nuevo.configurar(str(remoto), s_nuevo.config()["huella"])
|
|
r = s_nuevo.sincronizar(nuevo)
|
|
self.assertTrue(r["descifrado"] and r["subido"])
|
|
self.assertEqual({m["id"]: m["moneda"] for m in nuevo.datos.movimientos}, {"a": "EUR", "b": "USD"})
|
|
self.assertEqual(version_de(self.tmp / "app-nueva" / "datos"), siguiente)
|
|
self.assertTrue(s_nuevo.comprobar(nuevo.datos)["identica"]) # lo subido ya va en el formato nuevo
|
|
|
|
def test_una_copia_de_una_app_mas_nueva_no_se_usa(self):
|
|
remoto = self.repositorio("formato-nuevo")
|
|
siguiente = D.VERSION + 1
|
|
with mock.patch.object(D, "VERSION", siguiente), mock.patch.dict(D.MIGRACIONES, {siguiente: lambda t: None}):
|
|
nuevo, s_nuevo = self.equipo("app-nueva-2", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_nuevo.configurar(str(remoto), s_nuevo.config()["huella"])
|
|
s_nuevo.sincronizar(nuevo) # sube en el formato siguiente
|
|
viejo, s_viejo = self.equipo("app-vieja-2", Datos([activo("F1")], [mov("b")], [], []))
|
|
s_viejo.configurar(str(remoto), s_viejo.config()["huella"])
|
|
antes = huella_datos(leer(self.tmp / "app-vieja-2" / "datos"))
|
|
with self.assertRaisesRegex(ErrorSync, "más nueva"):
|
|
s_viejo.sincronizar(viejo)
|
|
self.assertEqual(huella_datos(leer(self.tmp / "app-vieja-2" / "datos")), antes)
|
|
self.assertEqual(huella_datos(viejo.datos), antes)
|
|
|
|
def test_los_justificantes_viajan_cifrados_en_paquetes(self):
|
|
remoto = self.repositorio("adjuntos")
|
|
pdf1, pdf2 = b"%PDF-1.4 extracto de enero\n", b"%PDF-1.4 factura de febrero\n"
|
|
usb, s_usb = self.equipo("adj-usb", Datos([activo("F1")], [mov("a"), mov("b")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
usb.adjuntar("a", "enero.pdf", pdf1)
|
|
r = s_usb.sincronizar(usb)
|
|
self.assertEqual((r["subido"], r["adjuntos"]["subidos"]), (True, 1))
|
|
paquetes = sorted((s_usb.repo / "adjuntos").glob("*.tar.gpg"))
|
|
self.assertEqual(len(paquetes), 1)
|
|
primero = paquetes[0].read_bytes()
|
|
self.assertNotIn(b"extracto de enero", primero) # cifrado
|
|
indice = leer_ajustes(self.tmp / "adj-usb")["adjuntos.json"]
|
|
self.assertNotIn("enero.pdf", json.dumps(indice)) # el índice no lleva nombres
|
|
|
|
self.assertEqual(s_usb.sincronizar(usb)["adjuntos"]["subidos"], 0) # nada nuevo: nada que subir
|
|
usb.adjuntar("b", "febrero.pdf", pdf2)
|
|
usb.adjuntar("a", "copia.pdf", pdf2) # mismo fichero, otra operación
|
|
self.assertEqual(s_usb.sincronizar(usb)["adjuntos"]["subidos"], 1)
|
|
self.assertEqual(len(list((s_usb.repo / "adjuntos").glob("*.tar.gpg"))), 2)
|
|
self.assertEqual(paquetes[0].read_bytes(), primero) # un paquete no se vuelve a tocar
|
|
|
|
otro, s_otro = self.equipo("adj-otro", Datos()) # otro equipo: se los trae
|
|
s_otro.configurar(str(remoto), s_otro.config()["huella"])
|
|
r = s_otro.sincronizar(otro)
|
|
self.assertEqual((r["adjuntos"]["traidos"], r["adjuntos"]["faltan"]), (2, 0))
|
|
self.assertEqual(otro.leer_adjunto("b", hashlib.sha256(pdf2).hexdigest())[1], pdf2)
|
|
self.assertEqual(s_otro.comprobar(otro.datos)["adjuntos_sin_copia"], 0)
|
|
|
|
nuevo = self.tmp / "adj-nuevo" / "datos" # USB nuevo: restaurar
|
|
r = restaurar(nuevo, str(remoto), sync_dir=self.tmp / "adj-nuevo" / "sync")
|
|
self.assertEqual(r["adjuntos"], {"traidos": 2, "faltan": 0})
|
|
self.assertEqual((nuevo / "adjuntos" / hashlib.sha256(pdf1).hexdigest()).read_bytes(), pdf1)
|
|
|
|
def test_un_paquete_que_no_llego_al_servidor_se_vuelve_a_subir(self):
|
|
remoto = self.repositorio("adjuntos-fallo")
|
|
usb, s_usb = self.equipo("adj-fallo", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
s_usb.sincronizar(usb) # sin justificantes todavía
|
|
h = usb.adjuntar("a", "a.pdf", b"%PDF justificante")["adjuntos"][0]["id"]
|
|
# Simula una subida que falló: el índice local dice que está en un
|
|
# paquete que en el servidor no existe
|
|
escribir_ajuste(self.tmp / "adj-fallo" / "adjuntos.json",
|
|
{"paquetes": {"20260101-000000-perdido": [h]}, "modificado": "2099-01-01T00:00:00"})
|
|
r = s_usb.sincronizar(usb)
|
|
self.assertEqual(r["adjuntos"]["subidos"], 1)
|
|
paquetes = leer_ajustes(self.tmp / "adj-fallo")["adjuntos.json"]["paquetes"]
|
|
self.assertNotIn("20260101-000000-perdido", paquetes)
|
|
self.assertEqual(list(paquetes.values()), [[h]])
|
|
|
|
def test_una_copia_antigua_sin_ajustes_no_los_borra(self):
|
|
"""Las copias anteriores a septiembre de 2026 solo llevan los .jsonl."""
|
|
remoto = self.tmp / "vieja.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(remoto)], check=True)
|
|
usb, s_usb = self.equipo("usb-v", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
s_usb.sincronizar(usb) # copia sin ajustes todavía
|
|
|
|
escribir_ajuste(self.tmp / "usb-v" / "bitcoin.json", config_btc([monedero("w1")]))
|
|
(s_usb.carpeta / "estado.json").unlink() # obliga a descifrar la copia vieja
|
|
s_usb.sincronizar(usb)
|
|
self.assertEqual([m["id"] for m in leer_ajustes(self.tmp / "usb-v")["bitcoin.json"]["monederos"]], ["w1"])
|
|
|
|
def test_el_usb_nuevo_recupera_tambien_los_monederos(self):
|
|
remoto = self.tmp / "restaurar-a.git"
|
|
subprocess.run(["git", "init", "--quiet", "--bare", str(remoto)], check=True)
|
|
usb, s_usb = self.equipo("usb-r2", Datos([activo("F1")], [mov("a")], [], []))
|
|
s_usb.configurar(str(remoto), s_usb.config()["huella"])
|
|
escribir_ajuste(self.tmp / "usb-r2" / "bitcoin.json", config_btc([monedero("w1", ["tx1"])]))
|
|
s_usb.sincronizar(usb)
|
|
|
|
nuevo = self.tmp / "nuevo-r2" / "datos"
|
|
r = restaurar(nuevo, str(remoto))
|
|
self.assertEqual(r["ajustes"], ["bitcoin.json"])
|
|
recuperado = leer_ajustes(nuevo.parent)["bitcoin.json"]
|
|
self.assertEqual(recuperado["monederos"][0]["clave"], "zpub-w1")
|
|
self.assertEqual(recuperado["servidores"][0]["direccion"], "nodo.onion:50001")
|
|
self.assertEqual((nuevo.parent / "bitcoin.json").stat().st_mode & 0o777, 0o600)
|
|
|
|
cartera = Cartera(nuevo) # y arranca en sincronía
|
|
self.addCleanup(cartera.cerrar)
|
|
s_nuevo = Sincronizador(nuevo.parent / "sync", ajustes=nuevo.parent)
|
|
self.assertFalse(s_nuevo.estado(cartera.datos)["pendiente"])
|
|
|
|
def test_sin_configurar_o_repositorio_inalcanzable(self):
|
|
cartera = Cartera(self.tmp / "suelto")
|
|
self.addCleanup(cartera.cerrar)
|
|
with self.assertRaises(ErrorSync):
|
|
Sincronizador(self.tmp / "suelto-sync").sincronizar(cartera)
|
|
mal = Sincronizador(self.tmp / "mal-sync")
|
|
mal.configurar(str(self.tmp / "no-existe.git"), mal.claves()[0]["huella"])
|
|
with self.assertRaises(ErrorSync):
|
|
mal.sincronizar(cartera)
|
|
with self.assertRaises(ErrorSync):
|
|
mal.configurar(str(self.remoto), "0" * 40)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
def monedero(mid, vistas=(), **kw):
|
|
return {"id": mid, "nombre": f"Monedero {mid}", "clave": f"zpub-{mid}",
|
|
"vistas": list(vistas), **kw}
|
|
|
|
|
|
def config_btc(monederos, cuando="2026-01-01T00:00:00"):
|
|
return {"servidores": [{"id": "s1", "direccion": "nodo.onion:50001"}],
|
|
"monederos": list(monederos), "modificado": cuando}
|
|
|
|
|
|
class TestElegirAjustes(unittest.TestCase):
|
|
"""Los ajustes no se fusionan campo a campo: gana el más reciente entero."""
|
|
|
|
def test_gana_el_mas_reciente(self):
|
|
viejo = config_btc([monedero("w1")], "2026-01-01T00:00:00")
|
|
nuevo = config_btc([monedero("w2")], "2026-05-01T00:00:00")
|
|
self.assertEqual(elegir_ajustes({"bitcoin.json": viejo}, {"bitcoin.json": nuevo})["bitcoin.json"]["monederos"],
|
|
nuevo["monederos"])
|
|
self.assertEqual(elegir_ajustes({"bitcoin.json": nuevo}, {"bitcoin.json": viejo})["bitcoin.json"]["monederos"],
|
|
nuevo["monederos"])
|
|
|
|
def test_sin_marca_de_fecha_gana_el_local(self):
|
|
local = {"monederos": [monedero("w1")]}
|
|
remoto = {"monederos": [monedero("w2")]}
|
|
r = elegir_ajustes({"bitcoin.json": local}, {"bitcoin.json": remoto})
|
|
self.assertEqual(r["bitcoin.json"]["monederos"][0]["id"], "w1")
|
|
|
|
def test_lo_que_solo_esta_en_un_sitio_se_conserva(self):
|
|
solo_remoto = elegir_ajustes({}, {"bitcoin.json": config_btc([monedero("w1")])})
|
|
self.assertEqual(solo_remoto["bitcoin.json"]["monederos"][0]["id"], "w1") # USB nuevo
|
|
solo_local = elegir_ajustes({"extractos.json": {"cuenta": "CC1"}}, {})
|
|
self.assertEqual(solo_local["extractos.json"], {"cuenta": "CC1"})
|
|
|
|
def test_un_monedero_borrado_no_vuelve(self):
|
|
"""Con un solo equipo es lo normal: borrar y sincronizar."""
|
|
antes = config_btc([monedero("w1"), monedero("w2")], "2026-01-01T00:00:00")
|
|
despues = config_btc([monedero("w1")], "2026-02-01T00:00:00")
|
|
r = elegir_ajustes({"bitcoin.json": despues}, {"bitcoin.json": antes})
|
|
self.assertEqual([m["id"] for m in r["bitcoin.json"]["monederos"]], ["w1"])
|
|
|
|
def test_las_transacciones_ya_apuntadas_se_suman(self):
|
|
local = config_btc([monedero("w1", ["tx1", "tx2"])], "2026-02-01T00:00:00")
|
|
remoto = config_btc([monedero("w1", ["tx2", "tx3"])], "2026-01-01T00:00:00")
|
|
for a, b in ((local, remoto), (remoto, local)):
|
|
r = elegir_ajustes({"bitcoin.json": a}, {"bitcoin.json": b})
|
|
self.assertEqual(r["bitcoin.json"]["monederos"][0]["vistas"], ["tx1", "tx2", "tx3"],
|
|
"sin sumarlas se volverían a proponer operaciones ya guardadas")
|
|
|
|
def test_sumar_las_vistas_no_resucita_un_monedero(self):
|
|
gana = config_btc([monedero("w1", ["tx1"])], "2026-02-01T00:00:00")
|
|
pierde = config_btc([monedero("w1", ["tx9"]), monedero("w2", ["tx8"])], "2026-01-01T00:00:00")
|
|
r = elegir_ajustes({"bitcoin.json": gana}, {"bitcoin.json": pierde})["bitcoin.json"]
|
|
self.assertEqual([m["id"] for m in r["monederos"]], ["w1"])
|
|
self.assertEqual(r["monederos"][0]["vistas"], ["tx1", "tx9"])
|
|
|
|
def test_no_toca_el_original(self):
|
|
local = config_btc([monedero("w1", ["tx1"])], "2026-02-01T00:00:00")
|
|
elegir_ajustes({"bitcoin.json": local}, {"bitcoin.json": config_btc([monedero("w1", ["tx2"])])})
|
|
self.assertEqual(local["monederos"][0]["vistas"], ["tx1"])
|
|
|
|
def test_se_escriben_solo_para_su_dueno(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
ruta = Path(t) / "bitcoin.json"
|
|
escribir_ajuste(ruta, config_btc([monedero("w1")]))
|
|
self.assertEqual(ruta.stat().st_mode & 0o777, 0o600) # las claves públicas delatan el historial
|
|
self.assertEqual(leer_ajustes(Path(t))["bitcoin.json"]["monederos"][0]["id"], "w1")
|