259 lines
13 KiB
Python
259 lines
13 KiB
Python
"""
|
|
Trabajar con el mismo Caudal desde varias copias (USB, móvil, servidor).
|
|
|
|
Cada copia genera sola cosas al abrirse (automáticas, cupones, precios,
|
|
saldos de los monederos). Estas pruebas comprueban que, al sincronizar, eso
|
|
no pisa lo que se hizo en la otra copia ni cuela lo que ya no toca.
|
|
|
|
python3 -m unittest pruebas.test_dos_copias -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import btc, datos as D, marcas # noqa: E402
|
|
from cartera.datos import Datos, escribir, fusionar, huella_datos, leer # noqa: E402
|
|
from cartera.enlace import ErrorEnlace, desempaquetar, empaquetar # noqa: E402
|
|
from cartera.operaciones import Cartera # noqa: E402
|
|
from cartera.sincronizar import elegir_ajustes # noqa: E402
|
|
|
|
HOY = dt.date.today()
|
|
|
|
|
|
class DosCopias(unittest.TestCase):
|
|
"""«start9» y «usb» empiezan con los mismos datos."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
base = Path(self._tmp.name)
|
|
self.c9, self.cu = base / "start9" / "datos", base / "usb" / "datos"
|
|
escribir(self.c9, self.datos_iniciales())
|
|
self.start9 = Cartera(self.c9)
|
|
self.addCleanup(self.start9.cerrar)
|
|
|
|
def datos_iniciales(self) -> Datos:
|
|
return Datos()
|
|
|
|
def copiar_al_usb(self):
|
|
escribir(self.cu, leer(self.c9))
|
|
|
|
def abrir_usb(self) -> Cartera:
|
|
c = Cartera(self.cu)
|
|
self.addCleanup(c.cerrar)
|
|
return c
|
|
|
|
def en_los_dos_sentidos(self, a: Datos, b: Datos):
|
|
r1, _ = fusionar(a, b)
|
|
r2, _ = fusionar(b, a)
|
|
self.assertEqual(huella_datos(r1), huella_datos(r2), "las dos copias deben llegar a lo mismo")
|
|
return r1
|
|
|
|
|
|
class TestAutomaticaDeInversionCerrada(DosCopias):
|
|
def datos_iniciales(self):
|
|
return Datos([{"id": "C1", "nombre": "Hosting", "tipo": "Cripto", "estado": "abierta"}], [], [], [])
|
|
|
|
def _gastos(self, datos):
|
|
return sorted(m["fecha"] for m in datos.vivos("movimientos") if m["tipo"] == "Gasto")
|
|
|
|
def test_cerrar_la_inversion_para_sus_automaticas(self):
|
|
self.start9.guardar_recurrente({"activo": "C1", "tipo": "Gasto", "importe": 10, "dia": 1, "desde": "2026-07-01"})
|
|
a = self.start9.activo("C1")
|
|
a.update({"estado": "cerrada", "fecha_cierre": "2026-08-15"})
|
|
self.start9._guardar()
|
|
self.start9.generar_recurrentes(dt.date(2026, 11, 17))
|
|
self.assertEqual(self._gastos(self.start9.datos), ["2026-07-01", "2026-08-01"])
|
|
|
|
def test_lo_que_genero_el_usb_tras_el_cierre_no_llega(self):
|
|
self.start9.guardar_recurrente({"activo": "C1", "tipo": "Gasto", "importe": 10, "dia": 1, "desde": "2026-07-01"})
|
|
self.start9.generar_recurrentes(dt.date(2026, 8, 17))
|
|
self.copiar_al_usb()
|
|
a = self.start9.activo("C1")
|
|
a.update({"estado": "cerrada", "fecha_cierre": "2026-08-15", "modificado": "2026-08-15T10:00:00.000000Z"})
|
|
self.start9._guardar()
|
|
usb = self.abrir_usb()
|
|
usb.generar_recurrentes(dt.date(2026, 11, 17)) # el USB no sabe que se cerró
|
|
r = self.en_los_dos_sentidos(leer(self.cu), leer(self.c9))
|
|
self.assertEqual(self._gastos(r), ["2026-07-01", "2026-08-01"])
|
|
|
|
|
|
class TestRentaFija(DosCopias):
|
|
"""Un bono con cupón anual el 30 de junio, comprado en 2024."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
a = self.start9.guardar_activo({"tipo": "RentaFija", "nombre": "Bono", "renta_fija": {
|
|
"clase": "bono", "nominal": 10000, "interes": 4.0, "pagos": 1, "vencimiento": "2029-06-30"}})
|
|
self.aid = a["id"]
|
|
self.start9.guardar_movimiento({"activo": self.aid, "tipo": "Compra", "fecha": "2024-01-15",
|
|
"importe": 10000, "cantidad": 10000})
|
|
|
|
def _cupones(self, datos):
|
|
return sorted((m["fecha"], m["id"].startswith("rf-")) for m in datos.vivos("movimientos")
|
|
if m["activo"] == self.aid and m["tipo"] == "Dividendo")
|
|
|
|
def test_los_cupones_llevan_la_marca_de_la_app(self):
|
|
cupones = [m for m in self.start9.datos.vivos("movimientos") if m["id"].startswith("rf-")]
|
|
self.assertTrue(cupones)
|
|
self.assertTrue(all(D.nivel_automatica(m) for m in cupones))
|
|
|
|
def test_el_cupon_apuntado_a_mano_en_una_copia_no_se_duplica(self):
|
|
self.copiar_al_usb() # el USB tiene los automáticos
|
|
auto = next(m for m in self.start9.datos.vivos("movimientos") if m["id"].endswith("2025-06-30-cupon"))
|
|
self.start9.guardar_movimiento({"activo": self.aid, "tipo": "Dividendo", "fecha": "2025-07-02",
|
|
"importe": 330, "retencion": 70, "notas": "del extracto"})
|
|
# En el Start9 el automático se retira solo, al ver el de verdad
|
|
self.assertNotIn(("2025-06-30", True), self._cupones(self.start9.datos))
|
|
start9 = leer(self.c9)
|
|
# Y aunque el Start9 no lo hubiera tenido nunca, el del USB no se cuela
|
|
start9.movimientos = [m for m in start9.movimientos if m["id"] != auto["id"]]
|
|
r = self.en_los_dos_sentidos(leer(self.cu), start9)
|
|
self.assertIn(("2025-07-02", False), self._cupones(r))
|
|
self.assertNotIn(("2025-06-30", True), self._cupones(r))
|
|
|
|
def test_vendido_en_una_copia_no_llegan_cupones_posteriores_de_la_otra(self):
|
|
self.copiar_al_usb()
|
|
self.start9.guardar_movimiento({"activo": self.aid, "tipo": "Reembolso", "fecha": "2025-03-01",
|
|
"importe": 10050, "cantidad": 10000})
|
|
self.assertEqual([f for f, auto in self._cupones(self.start9.datos) if auto], ["2024-06-30"])
|
|
r = self.en_los_dos_sentidos(leer(self.cu), leer(self.c9))
|
|
self.assertEqual([f for f, auto in self._cupones(r) if auto], ["2024-06-30"])
|
|
|
|
def test_el_cupon_corregido_a_mano_gana_al_generado_despues(self):
|
|
auto = next(m for m in self.start9.datos.vivos("movimientos") if m["id"].endswith("2025-06-30-cupon"))
|
|
self.start9.guardar_movimiento({**auto, "importe": 333.0, "estimado": False})
|
|
start9 = leer(self.c9)
|
|
usb = leer(self.c9) # el USB lo generó más tarde, sin tocar
|
|
m = next(x for x in usb.movimientos if x["id"] == auto["id"])
|
|
m.update({**auto, "modificado": "2099-01-01T00:00:00.000000Z",
|
|
"automatica": {"nivel": D.AUTO_ESTIMADA, "modificado": "2099-01-01T00:00:00.000000Z"}})
|
|
r = self.en_los_dos_sentidos(usb, start9)
|
|
self.assertEqual(next(x for x in r.movimientos if x["id"] == auto["id"])["importe"], 333.0)
|
|
|
|
|
|
class TestPrecioManual(DosCopias):
|
|
def datos_iniciales(self):
|
|
return Datos([{"id": "F1", "nombre": "Fondo", "tipo": "Fondo", "estado": "abierta"}], [], [], [])
|
|
|
|
def test_lo_descargado_no_pisa_lo_puesto_a_mano(self):
|
|
self.start9.precio_manual("F1", "2026-10-20", 12.5)
|
|
self.start9.aplicar_precios([{"activo": "F1", "fecha": "2026-10-20", "precio": 11.0, "fuente": "Yahoo"}])
|
|
p = next(x for x in self.start9.datos.vivos("precios") if x["fecha"] == "2026-10-20")
|
|
self.assertEqual((p["precio"], p["fuente"]), (12.5, "manual"))
|
|
|
|
def test_al_sincronizar_manda_el_manual_aunque_el_descargado_sea_posterior(self):
|
|
self.copiar_al_usb()
|
|
self.start9.precio_manual("F1", "2026-10-20", 12.5)
|
|
usb = self.abrir_usb()
|
|
usb.aplicar_precios([{"activo": "F1", "fecha": "2026-10-20", "precio": 11.0, "fuente": "Yahoo"}])
|
|
r = self.en_los_dos_sentidos(leer(self.cu), leer(self.c9))
|
|
self.assertEqual([(x["precio"], x["fuente"]) for x in r.vivos("precios")], [(12.5, "manual")])
|
|
|
|
|
|
def monedero(mid, **kw):
|
|
return {"id": mid, "nombre": f"Monedero {mid}", "clave": f"xpub-{mid}", **kw}
|
|
|
|
|
|
class TestAjustes(unittest.TestCase):
|
|
"""bitcoin.json y extractos.json, tal como se guardan y se mezclan."""
|
|
|
|
def setUp(self):
|
|
t = tempfile.TemporaryDirectory()
|
|
self.addCleanup(t.cleanup)
|
|
self.c9, self.cu = Path(t.name) / "start9" / "datos", Path(t.name) / "usb" / "datos"
|
|
for c in (self.c9, self.cu):
|
|
c.mkdir(parents=True)
|
|
|
|
def _btc(self, carpeta):
|
|
return json.loads(btc.almacen.leer_texto(btc.ruta_config(carpeta)))
|
|
|
|
def test_la_comprobacion_diaria_del_usb_no_tira_el_monedero_nuevo_del_start9(self):
|
|
base = btc.cargar(self.c9)
|
|
base["monederos"] = [monedero("w1")]
|
|
btc.guardar(self.c9, base)
|
|
btc.guardar(self.cu, self._btc(self.c9)) # las dos copias igual
|
|
# En el Start9 añade un monedero y toma una decisión del historial
|
|
s9 = btc.cargar(self.c9)
|
|
s9["monederos"].append(monedero("w2"))
|
|
s9["resueltas"] = {"tx9:0": {"accion": "compra"}}
|
|
btc.guardar(self.c9, s9)
|
|
# Después, el USB se abre: la comprobación diaria guarda saldos y otra decisión suya
|
|
usb = btc.cargar(self.cu)
|
|
usb["monederos"][0].update({"saldo": 0.5, "comprobado": "2099-01-01T00:00:00+00:00"})
|
|
usb["ultima"] = {"fecha": "2099-01-01T00:00:00+00:00"}
|
|
btc.guardar(self.cu, usb)
|
|
usb = btc.cargar(self.cu)
|
|
usb["resueltas"] = {"txU:1": {"accion": "regalo"}}
|
|
btc.guardar(self.cu, usb)
|
|
for local, remoto in ((self._btc(self.cu), self._btc(self.c9)), (self._btc(self.c9), self._btc(self.cu))):
|
|
r = elegir_ajustes({"bitcoin.json": local}, {"bitcoin.json": remoto})["bitcoin.json"]
|
|
self.assertEqual([m["id"] for m in r["monederos"]], ["w1", "w2"])
|
|
self.assertEqual(sorted(r["resueltas"]), ["tx9:0", "txU:1"])
|
|
self.assertEqual(r["monederos"][0]["saldo"], 0.5) # lo leído por el USB, más reciente
|
|
self.assertEqual(r["ultima"]["fecha"], "2099-01-01T00:00:00+00:00")
|
|
|
|
def test_guardar_solo_saldos_no_cuenta_como_cambio_del_usuario(self):
|
|
c = btc.cargar(self.c9)
|
|
c["monederos"] = [monedero("w1")]
|
|
btc.guardar(self.c9, c)
|
|
antes = self._btc(self.c9)["modificado"]
|
|
c = btc.cargar(self.c9)
|
|
c["monederos"][0].update({"saldo": 1.0, "comprobado": "x", "vistas": ["tx1"], "revision": {}})
|
|
c["ultima"] = {"fecha": "x"}
|
|
btc.guardar(self.c9, c)
|
|
self.assertEqual(self._btc(self.c9)["modificado"], antes)
|
|
|
|
def test_un_monedero_borrado_en_una_copia_no_vuelve(self):
|
|
c = btc.cargar(self.c9)
|
|
c["monederos"] = [monedero("w1"), monedero("w2")]
|
|
btc.guardar(self.c9, c)
|
|
btc.guardar(self.cu, self._btc(self.c9))
|
|
c = btc.cargar(self.c9)
|
|
c["monederos"] = [m for m in c["monederos"] if m["id"] != "w2"]
|
|
btc.guardar(self.c9, c)
|
|
u = btc.cargar(self.cu)
|
|
u["monederos"][1]["saldo"] = 3.0 # el USB solo leyó saldos después
|
|
btc.guardar(self.cu, u)
|
|
for a, b in ((self._btc(self.cu), self._btc(self.c9)), (self._btc(self.c9), self._btc(self.cu))):
|
|
r = elegir_ajustes({"bitcoin.json": a}, {"bitcoin.json": b})["bitcoin.json"]
|
|
self.assertEqual([m["id"] for m in r["monederos"]], ["w1"])
|
|
|
|
def test_cambios_distintos_en_cada_copia_se_quedan_los_dos(self):
|
|
anterior = marcas.marcar(None, {"fiscal": {"territorio": "comun"}, "trading": {"metodo": "fifo"}},
|
|
"extractos.json", "2026-10-01T00:00:00")
|
|
s9 = marcas.marcar(anterior, {**anterior, "fiscal": {"territorio": "navarra"}}, "extractos.json", "2026-10-02T00:00:00")
|
|
usb = marcas.marcar(anterior, {**anterior, "trading": {"metodo": "medio"}, "primer_uso": {"copia": "2026-10-03"}},
|
|
"extractos.json", "2026-10-03T00:00:00")
|
|
for a, b in ((usb, s9), (s9, usb)):
|
|
r = elegir_ajustes({"extractos.json": a}, {"extractos.json": b})["extractos.json"]
|
|
self.assertEqual((r["fiscal"], r["trading"], r["primer_uso"]),
|
|
({"territorio": "navarra"}, {"metodo": "medio"}, {"copia": "2026-10-03"}))
|
|
|
|
def test_con_una_copia_de_antes_de_las_marcas(self):
|
|
viejo = {"fiscal": {"territorio": "comun"}, "trading": {"metodo": "fifo"}, "modificado": "2026-10-05T00:00:00"}
|
|
nuevo = marcas.marcar({"fiscal": {"territorio": "comun"}, "modificado": "2026-10-01T00:00:00"},
|
|
{"fiscal": {"territorio": "navarra"}}, "extractos.json", "2026-10-09T00:00:00")
|
|
r = elegir_ajustes({"extractos.json": viejo}, {"extractos.json": nuevo})["extractos.json"]
|
|
self.assertEqual(r["fiscal"], {"territorio": "navarra"}) # lo cambiado después, del nuevo
|
|
self.assertEqual(r["trading"], {"metodo": "fifo"}) # lo que el nuevo no tenía, del viejo
|
|
|
|
|
|
class TestVersion(unittest.TestCase):
|
|
def test_un_caudal_anterior_no_puede_sincronizar(self):
|
|
paquete = empaquetar(Datos(), {})
|
|
self.assertEqual(paquete["version"], 3)
|
|
with self.assertRaises(ErrorEnlace):
|
|
desempaquetar({**paquete, "version": 2})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|