""" 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()