556 lines
32 KiB
Python
556 lines
32 KiB
Python
"""
|
||
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()
|