""" Pruebas de la fase 2 (aplicación). Sin datos reales y sin internet. python3 -m unittest discover -s pruebas -v """ from __future__ import annotations import datetime as dt import json import os import sys import tempfile import threading import unittest import urllib.error import urllib.request from pathlib import Path PROYECTO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROYECTO)) from cartera import servidor # noqa: E402 from cartera.calculo import calcular # noqa: E402 from cartera.datos import Datos, escribir, fusionar, huella_datos, leer # noqa: E402 from cartera.operaciones import Cartera, CarteraOcupada, ErrorOperacion # noqa: E402 from cartera.vista import resumen # noqa: E402 def activo(aid, tipo="Fondo", **kw): return {"id": aid, "nombre": f"Activo {aid}", "tipo": tipo, "codigo": "", "unidad": "u", "fecha_inicio": "2024-01-01", "visible": True, "estado": "abierta", "orden": 1, **kw} class ConCarpeta(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.carpeta = Path(self._tmp.name) escribir(self.carpeta, Datos( [activo("F1"), activo("F2"), activo("C1", "Cripto"), activo("P1", "PlanPensiones"), activo("PR1", "Prestamo", fecha_inicio="2025-01-01", prestamo={"importe_final": 1100.0, "vencimiento": "2026-01-01"})], [], [{"fecha": "2026-09-10", "activo": "P1", "precio": 10.0, "fuente": "x"}, {"fecha": "2026-09-14", "activo": "F1", "precio": 12.0, "fuente": "x"}])) self.c = Cartera(self.carpeta) def tearDown(self): self.c.cerrar() self._tmp.cleanup() class TestOperaciones(ConCarpeta): def test_calcula_el_dato_que_falta(self): m = self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "importe": "1000", "precio": "10"}) self.assertEqual((m["cantidad"], m["precio"], m["importe"]), (100.0, 10.0, 1000.0)) m = self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-02", "cantidad": "50", "importe": "600,5"}) self.assertAlmostEqual(m["precio"], 12.01) def test_no_vende_lo_que_no_tiene(self): self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 10, "precio": 1}) with self.assertRaises(ErrorOperacion): self.c.guardar_movimiento({"activo": "F1", "tipo": "Venta", "fecha": "2025-06-01", "cantidad": 11, "precio": 1}) with self.assertRaises(ErrorOperacion): # vender antes de comprar self.c.guardar_movimiento({"activo": "F1", "tipo": "Venta", "fecha": "2024-06-01", "cantidad": 1, "precio": 1}) def test_rechaza_basura(self): for mal in ({"activo": "F1", "tipo": "Compra", "fecha": "2025-13-01", "cantidad": 1, "precio": 1}, {"activo": "NO", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 1, "precio": 1}, {"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": -1, "precio": 1}, {"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": "nan", "precio": 1}, {"activo": "F1", "tipo": "Regalo", "fecha": "2025-01-01", "cantidad": 1, "precio": 1}, {"activo": "F1", "tipo": "Compra", "fecha": "2999-01-01", "cantidad": 1, "precio": 1}): with self.subTest(mal=mal), self.assertRaises(ErrorOperacion): self.c.guardar_movimiento(mal) self.assertEqual(leer(self.carpeta).movimientos, []) def test_gasto_y_mineria(self): self.c.guardar_movimiento({"activo": "C1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 0.01, "precio": 50000}) self.c.guardar_movimiento({"activo": "C1", "tipo": "Gasto", "fecha": "2025-02-01", "importe": 100}) self.c.guardar_movimiento({"activo": "C1", "tipo": "Mineria", "fecha": "2025-03-01", "cantidad": 0.002}) self.c.precio_manual("C1", "2025-12-31", 60000) r = calcular(self.c.datos, dt.date(2025, 12, 31)) m, h = r["motor"]["activos"]["C1"], r["hoja"]["activos"]["C1"] self.assertAlmostEqual(m["cantidad"], 0.012) self.assertAlmostEqual(m["ganancia"], 0.012 * 60000 - 600) # el gasto cuenta como invertido self.assertAlmostEqual(h["coste"], 600) self.assertAlmostEqual(r["motor"]["global"]["roi"], (720 - 600) / 600) def test_cerrar_prestamo_y_reabrir(self): self.c.guardar_movimiento({"activo": "PR1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 1, "importe": 1000}) with self.assertRaises(ErrorOperacion): self.c.cerrar_activo("PR1", "2024-12-31", 1100) # antes de la compra: quedaría con cosas posteriores self.c.cerrar_activo("PR1", "2025-12-15", 1090) r = resumen(self.c.datos, dt.date(2026, 9, 1)) cerrado = next(f for f in r["cerrados"] if f["id"] == "PR1") self.assertAlmostEqual(cerrado["ganancia"], 90) self.assertEqual(cerrado["dias"], 348) self.assertGreater(cerrado["tir"], 0.09) self.assertNotIn("PR1", [f["id"] for f in r["abiertos"]]) self.assertAlmostEqual(r["global"]["ganancia_realizada"], 90) with self.assertRaises(ErrorOperacion): self.c.cerrar_activo("PR1", "2025-12-20", 0) self.c.reabrir_activo("PR1") self.assertIn("PR1", [f["id"] for f in resumen(self.c.datos, dt.date(2026, 9, 1))["abiertos"]]) def test_traspaso(self): self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 100, "precio": 10}) sal, ent = self.c.traspaso({"origen": "F1", "destino": "F2", "fecha": "2025-06-01", "cantidad_salida": 100, "importe": 1200, "cantidad_entrada": 60, "coste_fiscal": 1000}) self.assertEqual((sal["tipo"], ent["tipo"], ent["coste_traspaso"]), ("Traspaso_Salida", "Traspaso_Entrada", 1000)) r = calcular(self.c.datos, dt.date(2025, 6, 2)) self.assertAlmostEqual(r["hoja"]["activos"]["F2"]["coste"], 1000) with self.assertRaises(ErrorOperacion): self.c.traspaso({"origen": "F1", "destino": "F2", "fecha": "2025-06-02", "cantidad_salida": 1, "importe": 12, "cantidad_entrada": 1}) def test_traspaso_de_cripto_entre_un_exchange_y_un_nodo(self): # «BTC en Cambium» (C1) → «BTC en Phoenix» (C2): no es venta, el coste viaja con él self.c.guardar_activo({"nombre": "BTC en Phoenix", "tipo": "Cripto", "codigo": "bitcoin", "unidad": "BTC", "fecha_inicio": "2024-01-01"}) self.c.datos.activos[[a["id"] for a in self.c.datos.activos].index("C1")]["codigo"] = "bitcoin" self.c.guardar_movimiento({"activo": "C1", "tipo": "Compra", "fecha": "2024-03-01", "cantidad": 0.02, "importe": 1200}) c2 = next(a["id"] for a in self.c.datos.vivos("activos") if a["nombre"] == "BTC en Phoenix") # sin importe ni coste: precio del día (estimado) y coste medio del origen; llega menos (comisión) sal, ent = self.c.traspaso({"origen": "C1", "destino": c2, "fecha": "2024-03-10", "cantidad_salida": 0.01, "cantidad_entrada": 0.0099}) self.assertEqual((sal["cantidad"], ent["cantidad"]), (0.01, 0.0099)) self.assertAlmostEqual(ent["coste_traspaso"], 600) self.assertTrue(sal["estimado"] and ent["estimado"]) self.assertGreater(sal["importe"], 0) # con cantidad de llegada vacía llega lo mismo que sale, y con importe dado no es estimado _, ent = self.c.traspaso({"origen": "C1", "destino": c2, "fecha": "2024-03-11", "cantidad_salida": 0.005, "importe": 300}) self.assertEqual((ent["cantidad"], ent.get("estimado")), (0.005, None)) with self.assertRaisesRegex(ErrorOperacion, "No puede llegar más"): self.c.traspaso({"origen": "C1", "destino": c2, "fecha": "2024-03-12", "cantidad_salida": 0.001, "cantidad_entrada": 0.002}) with self.assertRaisesRegex(ErrorOperacion, "misma moneda"): self.c.traspaso({"origen": "C1", "destino": "F1", "fecha": "2024-03-12", "cantidad_salida": 0.001, "importe": 50, "cantidad_entrada": 1}) with self.assertRaisesRegex(ErrorOperacion, "tantas"): self.c.traspaso({"origen": "C1", "destino": c2, "fecha": "2024-03-12", "cantidad_salida": 1}) def test_cerrar_fondo_traspasado_no_duplica_ganancia(self): self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 100, "precio": 10}) self.c.guardar_movimiento({"activo": "F1", "tipo": "Reembolso", "fecha": "2025-03-01", "cantidad": 20, "precio": 11}) self.c.traspaso({"origen": "F1", "destino": "F2", "fecha": "2025-06-01", "cantidad_salida": 80, "importe": 960, "cantidad_entrada": 48, "coste_fiscal": 800}) self.c.precio_manual("F2", "2025-12-31", 20) hoy = dt.date(2025, 12, 31) antes = resumen(self.c.datos, hoy)["global"] self.c.cerrar_activo("F1", "2025-06-01", 0) r = resumen(self.c.datos, hoy) f1 = next(f for f in r["cerrados"] if f["id"] == "F1") self.assertAlmostEqual(f1["ganancia"], 20 * 11 - 200) # solo el reembolso realiza self.assertAlmostEqual(f1["traspasado"], 960) self.assertAlmostEqual(r["global"]["ganancia_latente"], antes["ganancia_latente"]) self.assertAlmostEqual(r["global"]["ganancia_realizada"], 20) self.assertAlmostEqual(r["global"]["ganancia_latente"], 48 * 20 - 800) def test_alta_con_operacion_todo_o_nada(self): nuevo = {"nombre": "Iberdrola", "tipo": "Accion", "codigo": "IBE.MC", "unidad": "acciones"} r = self.c.alta_con_operacion(nuevo, {"tipo": "Compra", "fecha": "2026-09-16", "cantidad": 10, "precio": 12.5}, {"precio": 13.0, "fecha": "2026-09-16", "fuente": "yahoo"}) self.assertEqual(r["activo"]["id"], "A1") self.assertEqual(r["movimiento"]["activo"], "A1") self.assertEqual(calcular(self.c.datos, dt.date(2026, 9, 17))["motor"]["activos"]["A1"]["valor"], 130) with self.assertRaises(ErrorOperacion): # operación inválida → el activo no queda dado de alta self.c.alta_con_operacion(nuevo, {"tipo": "Compra", "fecha": "2026-09-16"}) self.assertEqual([a["id"] for a in leer(self.carpeta).vivos("activos") if a["tipo"] == "Accion"], ["A1"]) def test_dividendos_cuentan_en_rentabilidad_no_en_coste(self): self.c.guardar_activo({"nombre": "Acción Ejemplo", "tipo": "Accion", "codigo": "ES0000000037", "alias": "ejemplo"}) self.c.guardar_movimiento({"activo": "A1", "tipo": "Compra", "fecha": "2025-09-17", "cantidad": 10, "precio": 80}) self.c.guardar_movimiento({"activo": "A1", "tipo": "Dividendo", "fecha": "2026-03-01", "importe": 30, "retencion": 7}) with self.assertRaises(ErrorOperacion): self.c.guardar_movimiento({"activo": "A1", "tipo": "Dividendo", "fecha": "2026-04-01"}) self.c.precio_manual("A1", "2026-09-17", 80) hoy = dt.date(2026, 9, 17) r = calcular(self.c.datos, hoy) m, h = r["motor"]["activos"]["A1"], r["hoja"]["activos"]["A1"] self.assertEqual((m["cantidad"], h["coste"], h["ganancia"]), (10, 800, 0)) # el dividendo no toca el coste self.assertAlmostEqual(m["ganancia"], 30) # pero sí la ganancia total self.assertGreater(m["irr"], 0.03) f = next(x for x in resumen(self.c.datos, hoy)["abiertos"] if x["id"] == "A1") self.assertEqual((f["dividendos"], f["dividendos_año"]), (30, 30)) self.assertEqual(self.c.datos.vivos("movimientos")[-1]["retencion"], 7) self.c.guardar_movimiento({"activo": "A1", "tipo": "Venta", "fecha": "2026-09-17", "cantidad": 10, "precio": 80}) self.c.cerrar_activo("A1", "2026-09-17", 0) cerrada = next(x for x in resumen(self.c.datos, hoy)["cerrados"] if x["id"] == "A1") self.assertAlmostEqual(cerrada["ganancia"], 30) # realizada = dividendos def test_recompensa_es_ganancia(self): self.c.guardar_activo({"nombre": "Acción Ejemplo", "tipo": "Accion", "codigo": "ES0000000037"}) self.c.guardar_movimiento({"activo": "A1", "tipo": "Compra", "fecha": "2026-01-02", "cantidad": 10, "precio": 80}) m = self.c.guardar_movimiento({"activo": "A1", "tipo": "Recompensa", "fecha": "2026-02-01", "cantidad": 0.5, "importe": 999, "valor_recibido": 41}) self.assertEqual((m["importe"], m["precio"], m["valor_recibido"]), (None, None, 41)) with self.assertRaises(ErrorOperacion): self.c.guardar_movimiento({"activo": "A1", "tipo": "Recompensa", "fecha": "2026-02-01", "valor_recibido": 5}) self.c.precio_manual("A1", "2026-09-17", 80) hoy = dt.date(2026, 9, 17) r = calcular(self.c.datos, hoy) mo, ho = r["motor"]["activos"]["A1"], r["hoja"]["activos"]["A1"] self.assertEqual((mo["cantidad"], ho["coste"]), (10.5, 800)) self.assertAlmostEqual(mo["ganancia"], 40) # 0,5 acciones gratis × 80 € self.assertAlmostEqual(ho["ganancia"], 40) f = next(x for x in resumen(self.c.datos, hoy)["abiertos"] if x["id"] == "A1") self.assertEqual((f["recompensas"], f["recompensas_valor"]), (0.5, 41)) def test_cuentas_son_liquidez_y_no_tocan_la_rentabilidad(self): self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-09-17", "cantidad": 100, "precio": 10}) self.c.precio_manual("F1", "2026-09-17", 12) hoy = dt.date(2026, 9, 17) antes = resumen(self.c.datos, hoy)["global"] cuenta = self.c.guardar_activo({"nombre": "Bankinter", "tipo": "Cuenta", "unidad": "€"}) self.assertEqual(cuenta["id"], "CC1") self.c.precio_manual(cuenta["id"], "2026-09-16", 5000) self.c.guardar_movimiento({"activo": cuenta["id"], "tipo": "Dividendo", "fecha": "2026-07-01", "importe": 12.5}) r = resumen(self.c.datos, hoy) self.assertEqual([(c["id"], c["saldo"], c["intereses_año"]) for c in r["liquidez"]], [("CC1", 5000, 12.5)]) self.assertNotIn("CC1", [f["id"] for f in r["abiertos"]]) for clave in ("valor", "coste", "ganancia_latente", "tir", "roi", "ytd"): self.assertEqual(r["global"][clave], antes[clave], clave) # la liquidez no altera la rentabilidad self.assertEqual((r["global"]["liquidez"], r["global"]["patrimonio"]), (5000, antes["valor"] + 5000)) self.assertEqual(r["global"]["dividendos_año"], 12.5) def test_seguimiento_por_valor_sin_participaciones(self): plan = self.c.guardar_activo({"nombre": "Plan por valor", "tipo": "PlanPensiones", "unidad": "€", "valoracion": "total", "fecha_inicio": "2024-01-05"}) self.c.guardar_movimiento({"activo": plan["id"], "tipo": "Compra", "fecha": "2024-01-05", "importe": 1000}) self.c.guardar_recurrente({"activo": plan["id"], "tipo": "Compra", "importe": 50, "dia": 5, "desde": "2024-02-05", "hasta": "2024-12-31"}) self.assertEqual(self.c.generar_recurrentes(dt.date(2026, 9, 17))["creadas"], 11) self.c.precio_manual(plan["id"], "2025-12-31", 1600) self.c.precio_manual(plan["id"], "2026-09-16", 1800) hoy = dt.date(2026, 9, 17) f = next(x for x in resumen(self.c.datos, hoy)["abiertos"] if x["id"] == plan["id"]) self.assertEqual((f["invertido"], f["valor"], f["cantidad"]), (1550, 1800, None)) self.assertAlmostEqual(f["ganancia"], 250) self.assertAlmostEqual(f["rentabilidad"], 250 / 1550) self.assertAlmostEqual(f["ytd"], 1800 / 1600 - 1) # sin aportaciones este año self.assertGreater(f["tir"], 0.05) with self.assertRaises(ErrorOperacion): # sin importe no vale self.c.guardar_movimiento({"activo": plan["id"], "tipo": "Compra", "fecha": "2026-09-01", "cantidad": 3}) for tipo in ("Reembolso", "Venta"): # salidas: solo el dinero, sin participaciones m = self.c.guardar_movimiento({"activo": plan["id"], "tipo": tipo, "fecha": "2026-09-10", "importe": 100}) self.assertEqual((m["cantidad"], m["importe"]), (None, 100)) def test_cerrada_por_valor_no_sigue_valiendo(self): # Al cerrar algo llevado por valor no hay participaciones que vender: su último valor # apuntado se quedaba contando y bajaba el «Peso» de las abiertas e inflaba la rentabilidad self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2024-01-02", "cantidad": 100, "precio": 10}) plan = self.c.guardar_activo({"nombre": "Plan por valor", "tipo": "PlanPensiones", "unidad": "€", "valoracion": "total", "fecha_inicio": "2024-01-05"}) self.c.guardar_movimiento({"activo": plan["id"], "tipo": "Compra", "fecha": "2024-01-05", "importe": 1000}) self.c.precio_manual(plan["id"], "2026-06-30", 1100) self.c.guardar_movimiento({"activo": plan["id"], "tipo": "Reembolso", "fecha": "2026-07-01", "importe": 1100}) self.c.cerrar_activo(plan["id"], "2026-07-01", 0) hoy = dt.date(2026, 9, 17) r = resumen(self.c.datos, hoy) self.assertNotIn(plan["id"], [f["id"] for f in r["abiertos"]]) self.assertAlmostEqual(sum(f["peso"] for f in r["abiertos"]), 1.0) self.assertEqual(r["global"]["valor"], 1200) self.assertEqual(calcular(self.c.datos, hoy)["hoja"]["activos"][plan["id"]]["valor"], 0) self.assertAlmostEqual(r["global"]["roi"], (1200 + 1100 - 2000) / 2000) # ganó 100 y F1 200 # La gráfica de evolución: el plan cuenta hasta el cierre y después ya no evolucion = {p["fecha"]: p["valor"] for p in r["evolucion"]} self.assertEqual(evolucion["2026-06-30"], 2100) self.assertEqual(evolucion["2026-07-31"], 1000) self.assertEqual(r["evolucion"][-1]["valor"], r["global"]["valor"]) def test_borrado_persiste_como_marca(self): m = self.c.guardar_movimiento({"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": 1, "precio": 1}) self.c.borrar_movimiento(m["id"]) en_disco = leer(self.carpeta) self.assertEqual(len(en_disco.movimientos), 1) self.assertTrue(en_disco.movimientos[0]["borrado"]) self.assertEqual(en_disco.vivos("movimientos"), []) def test_cerrojo(self): with self.assertRaises(CarteraOcupada): Cartera(self.carpeta) def test_en_windows_el_cerrojo_usa_lo_que_ofrece_windows(self): """Sin fcntl se reserva el primer byte del fichero con msvcrt. Aquí se simula: en un Windows de verdad no se ha ejecutado.""" from unittest import mock from cartera import operaciones class FalsoMsvcrt: LK_NBLCK = 2 tomados: set = set() @classmethod def locking(cls, fd, modo, bytes_): clave = os.fstat(fd).st_ino if clave in cls.tomados: raise OSError(13, "Permission denied") cls.tomados.add(clave) cls.pedido = (modo, bytes_) self.c.cerrar() with mock.patch.object(operaciones, "fcntl", None), mock.patch.object(operaciones, "msvcrt", FalsoMsvcrt): una = Cartera(self.carpeta) self.assertEqual(FalsoMsvcrt.pedido, (FalsoMsvcrt.LK_NBLCK, 1)) with self.assertRaises(CarteraOcupada): Cartera(self.carpeta) una.cerrar() with mock.patch.object(operaciones, "fcntl", None), mock.patch.object(operaciones, "msvcrt", None): sin_nada = Cartera(self.carpeta) # un sistema sin ninguna de las dos: abre, sin protección self.assertIsNone(sin_nada._cerrojo) sin_nada.cerrar() self.c = Cartera(self.carpeta) class TestRecurrentes(ConCarpeta): def test_gasto_en_dolares_idempotente_y_pendiente_sin_red(self): self.c.guardar_recurrente({"activo": "C1", "tipo": "Gasto", "importe": 50, "moneda": "USD", "dia": 1, "desde": "2026-07-01"}) r = self.c.generar_recurrentes(dt.date(2026, 9, 17), cambio=lambda f, mon: None) self.assertEqual((r["creadas"], r["completadas"], r["pendientes"]), (3, 0, 3)) r = self.c.generar_recurrentes(dt.date(2026, 9, 17), cambio=lambda f, mon: 0.9) self.assertEqual((r["creadas"], r["completadas"], r["pendientes"]), (0, 3, 0)) gastos = [m for m in self.c.datos.vivos("movimientos") if m["tipo"] == "Gasto"] self.assertEqual(sorted(m["id"] for m in gastos)[0][-10:], "2026-07-01") self.assertTrue(all(abs(m["importe"] - 45.0) < 1e-9 for m in gastos)) # Borrar una generada no la resucita self.c.borrar_movimiento(gastos[0]["id"]) self.assertEqual(self.c.generar_recurrentes(dt.date(2026, 9, 17), cambio=lambda f, mon: 0.9)["creadas"], 0) def test_aportacion_estimada_y_afinado(self): self.c.guardar_recurrente({"activo": "P1", "tipo": "Compra", "importe": 150, "moneda": "EUR", "dia": 12, "desde": "2026-09-01"}) self.c.generar_recurrentes(dt.date(2026, 9, 17)) m = next(m for m in self.c.datos.vivos("movimientos") if m["activo"] == "P1") self.assertTrue(m["estimado"]) self.assertAlmostEqual(m["cantidad"], 15.0) # VL del día 10 self.c.aplicar_precios([{"activo": "P1", "fecha": "2026-09-12", "precio": 12.5, "fuente": "x"}]) self.c.generar_recurrentes(dt.date(2026, 9, 17)) self.assertAlmostEqual(m["cantidad"], 12.0) # ya con el VL exacto self.c.confirmar_movimiento(m["id"]) self.assertFalse(m["estimado"]) def test_sin_valor_liquidativo_reciente_queda_pendiente(self): self.c.guardar_recurrente({"activo": "F1", "tipo": "Compra", "importe": 100, "dia": 14, "desde": "2026-07-14"}) r = self.c.generar_recurrentes(dt.date(2026, 9, 17)) self.assertEqual((r["creadas"], r["pendientes"]), (3, 2)) # julio y agosto sin VL cercano avisos = resumen(self.c.datos, dt.date(2026, 9, 17))["avisos"] self.assertEqual(sum(a["tipo"] == "pendiente" for a in avisos), 2) class TestRecurrentesEnDosCopias(ConCarpeta): """La misma automática generada en el Start9 y, días después, en el USB: al fusionar no se duplica y no pisa lo que ya era mejor en la otra copia.""" def setUp(self): super().setUp() self.c.guardar_recurrente({"activo": "C1", "tipo": "Gasto", "importe": 50, "moneda": "USD", "dia": 1, "desde": "2026-09-01"}) self.c.guardar_recurrente({"activo": "P1", "tipo": "Compra", "importe": 150, "dia": 12, "desde": "2026-09-01"}) # El USB es una copia de antes de que vencieran self._tmp_usb = tempfile.TemporaryDirectory() self.carpeta_usb = Path(self._tmp_usb.name) escribir(self.carpeta_usb, leer(self.carpeta)) self.hoy = dt.date(2026, 9, 17) def tearDown(self): super().tearDown() self._tmp_usb.cleanup() def _usb(self, **kw): usb = Cartera(self.carpeta_usb) try: usb.generar_recurrentes(self.hoy, **kw) return leer(self.carpeta_usb) finally: usb.cerrar() def _gasto(self, datos): return [m for m in datos.vivos("movimientos") if m["tipo"] == "Gasto"] def test_la_corregida_a_mano_gana_a_la_generada_despues(self): self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) g = self._gasto(self.c.datos)[0] self.c.guardar_movimiento({"id": g["id"], "activo": "C1", "tipo": "Gasto", "fecha": g["fecha"], "importe": 46.3, "notas": "cargo real del banco"}) start9 = leer(self.carpeta) usb = self._usb(cambio=lambda f, mon: 0.95) # más tarde, con otro cambio for a, b in ((usb, start9), (start9, usb)): r, _ = fusionar(a, b) gastos = self._gasto(r) self.assertEqual(len(gastos), 1) # no se duplica self.assertAlmostEqual(gastos[0]["importe"], 46.3) self.assertEqual(gastos[0]["notas"], "cargo real del banco") def test_la_completa_gana_a_la_pendiente_y_la_exacta_a_la_estimada(self): self.c.aplicar_precios([{"activo": "P1", "fecha": "2026-09-12", "precio": 12.5, "fuente": "x"}]) self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) start9 = leer(self.carpeta) usb = self._usb(cambio=lambda f, mon: None) # sin red y sin el VL del día 12 self.assertTrue(self._gasto(usb)[0]["pendiente"]) for a, b in ((usb, start9), (start9, usb)): r, _ = fusionar(a, b) self.assertAlmostEqual(self._gasto(r)[0]["importe"], 45.0) compra = next(m for m in r.vivos("movimientos") if m["activo"] == "P1") self.assertAlmostEqual(compra["cantidad"], 12.0) # VL exacto del día 12, no el estimado def test_una_pendiente_de_una_version_anterior_no_gana_a_la_completa(self): self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) start9 = leer(self.carpeta) usb = self._usb(cambio=lambda f, mon: None) for m in usb.movimientos: # como la dejaba la versión anterior m.pop("automatica", None) m["modificado"] = "2099-01-01T00:00:00.000000Z" r, _ = fusionar(usb, start9) self.assertAlmostEqual(self._gasto(r)[0]["importe"], 45.0) def _regla_gasto(self): return next(r for r in self.c.datos.vivos("recurrentes") if r["tipo"] == "Gasto") def _fechas_gasto(self, datos): return sorted(m["fecha"] for m in self._gasto(datos)) def test_finalizada_en_el_start9_no_vuelve_con_lo_que_genero_el_usb(self): self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) self.c.guardar_recurrente({**self._regla_gasto(), "hasta": "2026-09-30"}) start9 = leer(self.carpeta) self.hoy = dt.date(2026, 11, 17) # el USB se abre en noviembre sin saberlo usb = self._usb(cambio=lambda f, mon: 0.9) self.assertEqual(self._fechas_gasto(usb), ["2026-09-01", "2026-10-01", "2026-11-01"]) for a, b in ((usb, start9), (start9, usb)): r, _ = fusionar(a, b) self.assertEqual(self._fechas_gasto(r), ["2026-09-01"]) self.assertEqual(next(x for x in r.recurrentes if x["tipo"] == "Gasto")["hasta"], "2026-09-30") r1, _ = fusionar(usb, start9) r2, _ = fusionar(start9, usb) self.assertEqual(huella_datos(r1), huella_datos(r2)) # las dos copias llegan a lo mismo def test_si_la_tocaste_a_mano_se_queda_aunque_sobre(self): self.hoy = dt.date(2026, 11, 17) usb = Cartera(self.carpeta_usb) try: usb.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) oct_ = next(m for m in usb.datos.vivos("movimientos") if m["fecha"] == "2026-10-01") usb.guardar_movimiento({"id": oct_["id"], "activo": "C1", "tipo": "Gasto", "fecha": oct_["fecha"], "importe": 44.0}) finally: usb.cerrar() self.c.guardar_recurrente({**self._regla_gasto(), "hasta": "2026-09-30"}) r, _ = fusionar(leer(self.carpeta_usb), leer(self.carpeta)) self.assertEqual(self._fechas_gasto(r), ["2026-09-01", "2026-10-01"]) def test_acortarla_retira_las_que_sobran_en_la_misma_copia(self): self.hoy = dt.date(2026, 11, 17) self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) self.c.guardar_recurrente({**self._regla_gasto(), "hasta": "2026-09-30"}) self.assertEqual(self._fechas_gasto(self.c.datos), ["2026-09-01"]) def _pausada_o_borrada(self, accion): hoy_real = dt.date.today() self.hoy = hoy_real self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) regla = self._regla_gasto() if accion == "pausar": self.c.guardar_recurrente({**regla, "pausado": True}) else: self.c.borrar_recurrente(regla["id"]) start9 = leer(self.carpeta) antes = self._fechas_gasto(start9) self.hoy = hoy_real + dt.timedelta(days=95) # el USB se abre meses después usb = self._usb(cambio=lambda f, mon: 0.9) self.assertGreater(len(self._fechas_gasto(usb)), len(antes)) for a, b in ((usb, start9), (start9, usb)): r, _ = fusionar(a, b) self.assertEqual(self._fechas_gasto(r), antes) # las de antes se quedan; las nuevas, no def test_pausada_en_el_start9(self): self._pausada_o_borrada("pausar") def test_borrada_en_el_start9(self): self._pausada_o_borrada("borrar") def test_un_borrado_gana_a_la_generada_despues(self): self.c.generar_recurrentes(self.hoy, cambio=lambda f, mon: 0.9) self.c.borrar_movimiento(self._gasto(self.c.datos)[0]["id"]) start9 = leer(self.carpeta) usb = self._usb(cambio=lambda f, mon: 0.95) for a, b in ((usb, start9), (start9, usb)): r, _ = fusionar(a, b) self.assertEqual(self._gasto(r), []) class TestServidor(unittest.TestCase): @classmethod def setUpClass(cls): cls._tmp = tempfile.TemporaryDirectory() escribir(Path(cls._tmp.name), Datos([activo("F1")], [], [])) cls.app = servidor.arrancar(Path(cls._tmp.name), 0) cls.hilo = threading.Thread(target=cls.app.servidor.serve_forever, daemon=True) cls.hilo.start() cls.base = f"http://127.0.0.1:{cls.app.puerto}" @classmethod def tearDownClass(cls): cls.app.servidor.shutdown() cls.app.servidor.server_close() cls.app.cartera.cerrar() cls._tmp.cleanup() def pedir(self, ruta, cuerpo=None, token=True, host=None): cab = {"Content-Type": "application/json"} if token: cab["X-Token"] = self.app.token if host: cab["Host"] = host datos = json.dumps(cuerpo).encode() if cuerpo is not None else None req = urllib.request.Request(self.base + ruta, data=datos, headers=cab) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status, r.read() except urllib.error.HTTPError as e: return e.code, e.read() def test_seguridad(self): self.assertEqual(self.pedir("/api/estado", token=False)[0], 403) self.assertEqual(self.pedir("/", token=False)[0], 403) self.assertEqual(self.pedir(f"/?t={self.app.token}", host="malicioso.example")[0], 403) self.assertEqual(self.pedir("/api/movimiento", {"x": 1}, token=False)[0], 403) self.assertEqual(self.pedir("/web/../datos.py")[0], 404) def test_pagina_y_estado(self): codigo, html = self.pedir(f"/?t={self.app.token}", token=False) self.assertEqual(codigo, 200) self.assertIn(self.app.token.encode(), html) codigo, cuerpo = self.pedir("/api/estado") self.assertEqual(codigo, 200) self.assertEqual(json.loads(cuerpo)["activos"][0]["id"], "F1") def test_alta_y_error_legible(self): codigo, cuerpo = self.pedir("/api/movimiento", {"activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "cantidad": "10", "precio": "2,5"}) self.assertEqual(codigo, 200) self.assertEqual(json.loads(cuerpo)["resultado"]["importe"], 25.0) codigo, cuerpo = self.pedir("/api/movimiento", {"activo": "F1", "tipo": "Compra", "fecha": "ayer"}) self.assertEqual(codigo, 400) self.assertIn("fecha", json.loads(cuerpo)["error"]) if __name__ == "__main__": unittest.main()