esetheyers
7e18576f86
Taxes for the whole of Spain (common territory, Navarre, Álava, Bizkaia and Gipuzkoa); flats, garages and land valued on their own; Treasury bills, bonds and deposits priced on BME; broker accounts; a guided start; a page per investment with its chart and the tax of selling today; net worth at the top; undo when deleting; Woodpecker CI for the tests and the Linux build.
201 lines
9.8 KiB
Python
201 lines
9.8 KiB
Python
"""
|
|
Pruebas de las acciones sobre varias operaciones a la vez y de los
|
|
justificantes adjuntos (guardado, lectura, descarga por el servidor).
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import adjuntos # noqa: E402
|
|
from cartera.datos import Datos, ErrorDatos, escribir, leer # noqa: E402
|
|
from cartera.operaciones import Cartera, ErrorOperacion # noqa: E402
|
|
|
|
PDF = b"%PDF-1.4 justificante de prueba\n"
|
|
|
|
|
|
class Base(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
self.carpeta = Path(self._tmp.name) / "datos"
|
|
escribir(self.carpeta, Datos([{"id": "F1", "nombre": "Fondo uno", "tipo": "Fondo"},
|
|
{"id": "F2", "nombre": "Fondo dos", "tipo": "Fondo"},
|
|
{"id": "D1", "nombre": "Hipoteca", "tipo": "Deuda", "fecha_inicio": "2025-01-15",
|
|
"deuda": {"capital": 1000.0, "tin": 2.0, "cuotas": 12,
|
|
"primera_cuota": "2025-02-01"}}], [], [], []))
|
|
self.c = Cartera(self.carpeta)
|
|
self.addCleanup(self.c.cerrar)
|
|
|
|
def op(self, tipo, fecha, cantidad, precio=10, activo="F1", **kw):
|
|
return self.c.guardar_movimiento({"activo": activo, "tipo": tipo, "fecha": fecha,
|
|
"cantidad": cantidad, "precio": precio, **kw})["id"]
|
|
|
|
def en_disco(self):
|
|
return {m["id"]: m for m in leer(self.carpeta).vivos("movimientos")}
|
|
|
|
|
|
class TestVariasALaVez(Base):
|
|
def test_borrar_varias(self):
|
|
a, b, c = self.op("Compra", "2025-01-01", 1), self.op("Compra", "2025-02-01", 1), self.op("Compra", "2025-03-01", 1)
|
|
r = self.c.borrar_movimientos([a, c, a])
|
|
self.assertEqual((r["borradas"], sorted(r["ids"])), (2, sorted([a, c]))) # repetidas cuentan una vez
|
|
self.assertEqual(set(self.en_disco()), {b})
|
|
|
|
def test_si_una_no_existe_no_se_toca_ninguna(self):
|
|
a = self.op("Compra", "2025-01-01", 1)
|
|
for mal in ([a, "no-existe"], [], None, "texto"):
|
|
with self.subTest(mal=mal), self.assertRaises(ErrorOperacion):
|
|
self.c.borrar_movimientos(mal)
|
|
self.assertIn(a, self.en_disco())
|
|
|
|
def test_confirmar_solo_las_estimadas_y_avisa_de_pendientes(self):
|
|
est = self.op("Compra", "2025-01-01", 1, estimado=True)
|
|
normal = self.op("Compra", "2025-02-01", 1)
|
|
pend = self.op("Compra", "2025-03-01", 1, estimado=True)
|
|
self.c._movimiento(pend)["pendiente"] = True
|
|
r = self.c.confirmar_movimientos([est, normal, pend])
|
|
self.assertEqual(r, {"confirmadas": 1, "pendientes": 1})
|
|
self.assertFalse(self.en_disco()[est]["estimado"])
|
|
|
|
def test_mover_de_activo_en_orden(self):
|
|
"""Una venta se mueve después de la compra del mismo día aunque venga antes en la lista."""
|
|
compra = self.op("Compra", "2025-01-01", 5)
|
|
venta = self.op("Venta", "2025-01-01", 3)
|
|
r = self.c.mover_movimientos([venta, compra], "F2")
|
|
self.assertEqual(r, {"movidas": 2, "activo": "F2"})
|
|
self.assertEqual({m["activo"] for m in self.en_disco().values()}, {"F2"})
|
|
|
|
def test_mover_es_todo_o_nada(self):
|
|
self.op("Compra", "2025-01-01", 5) # esta se queda en F1
|
|
compra = self.op("Compra", "2025-01-02", 1)
|
|
venta = self.op("Venta", "2025-02-01", 3) # en F2 solo habría 1: falla la segunda
|
|
antes = json.dumps(self.en_disco(), sort_keys=True)
|
|
with self.assertRaisesRegex(ErrorOperacion, "No se ha movido ninguna"):
|
|
self.c.mover_movimientos([compra, venta], "F2")
|
|
self.assertEqual(json.dumps(self.en_disco(), sort_keys=True), antes)
|
|
self.assertEqual({m["activo"] for m in self.c.datos.vivos("movimientos") if m["id"] in (compra, venta)},
|
|
{"F1"}) # también en memoria
|
|
|
|
def test_mover_a_un_prestamo_recibido_no_se_puede(self):
|
|
a = self.op("Compra", "2025-01-01", 1)
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.c.mover_movimientos([a], "D1")
|
|
|
|
|
|
class TestAdjuntos(Base):
|
|
def test_adjuntar_leer_y_quitar(self):
|
|
mid = self.op("Compra", "2025-01-01", 1)
|
|
m = self.c.adjuntar(mid, "../../carpeta/extracto banco.pdf", PDF)
|
|
ref = m["adjuntos"][0]
|
|
self.assertEqual(ref, {"id": hashlib.sha256(PDF).hexdigest(), "nombre": "extracto banco.pdf",
|
|
"tipo": "application/pdf", "bytes": len(PDF)})
|
|
self.assertEqual(self.en_disco()[mid]["adjuntos"], [ref])
|
|
self.assertEqual(self.c.leer_adjunto(mid, ref["id"]), (ref, PDF))
|
|
self.c.quitar_adjunto(mid, ref["id"])
|
|
self.assertEqual(self.en_disco()[mid]["adjuntos"], [])
|
|
self.assertTrue(adjuntos.tiene(self.carpeta, ref["id"])) # el fichero se queda
|
|
|
|
def test_editar_la_operacion_conserva_los_adjuntos(self):
|
|
mid = self.op("Compra", "2025-01-01", 1)
|
|
self.c.adjuntar(mid, "a.pdf", PDF)
|
|
self.c.guardar_movimiento({"id": mid, "activo": "F1", "tipo": "Compra", "fecha": "2025-01-02",
|
|
"cantidad": 2, "precio": 10})
|
|
self.assertEqual(len(self.en_disco()[mid]["adjuntos"]), 1)
|
|
|
|
def test_el_mismo_fichero_ocupa_una_vez(self):
|
|
a, b = self.op("Compra", "2025-01-01", 1), self.op("Compra", "2025-02-01", 1)
|
|
self.c.adjuntar(a, "uno.pdf", PDF)
|
|
self.c.adjuntar(b, "otro-nombre.pdf", PDF)
|
|
self.assertEqual(len(list((self.carpeta / "adjuntos").iterdir())), 1)
|
|
with self.assertRaisesRegex(ErrorOperacion, "ya está adjunto"):
|
|
self.c.adjuntar(a, "repetido.pdf", PDF)
|
|
|
|
def test_limites_y_tipos(self):
|
|
mid = self.op("Compra", "2025-01-01", 1)
|
|
for nombre, contenido, error in (("x.html", b"<script>", "no se admite"), ("x", b"a", "no se admite"),
|
|
("x.pdf", b"", "vacío"),
|
|
("x.pdf", b"a" * (adjuntos.MAX_BYTES + 1), "MB")):
|
|
with self.subTest(nombre=nombre), self.assertRaisesRegex(ErrorOperacion, error):
|
|
self.c.adjuntar(mid, nombre, contenido)
|
|
for i in range(adjuntos.MAX_POR_OPERACION):
|
|
self.c.adjuntar(mid, f"{i}.txt", f"fichero {i}".encode())
|
|
with self.assertRaisesRegex(ErrorOperacion, "Como mucho"):
|
|
self.c.adjuntar(mid, "otro.txt", b"uno mas")
|
|
|
|
def test_un_fichero_danado_se_detecta(self):
|
|
mid = self.op("Compra", "2025-01-01", 1)
|
|
h = self.c.adjuntar(mid, "a.pdf", PDF)["adjuntos"][0]["id"]
|
|
(self.carpeta / "adjuntos" / h).write_bytes(b"otra cosa")
|
|
with self.assertRaisesRegex(ErrorOperacion, "dañado"):
|
|
self.c.leer_adjunto(mid, h)
|
|
|
|
def test_solo_se_lee_lo_que_la_operacion_tiene_adjunto(self):
|
|
a, b = self.op("Compra", "2025-01-01", 1), self.op("Compra", "2025-02-01", 1)
|
|
h = self.c.adjuntar(a, "a.pdf", PDF)["adjuntos"][0]["id"]
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.c.leer_adjunto(b, h)
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.c.leer_adjunto(a, "../../activos.jsonl")
|
|
|
|
def test_datos_con_adjuntos_mal_formados_no_se_leen(self):
|
|
escribir(self.carpeta.parent / "otra", Datos([{"id": "F1", "nombre": "F", "tipo": "Fondo"}], [], [], []))
|
|
for mal in ([{"id": "x", "nombre": "a", "bytes": 1}], "no-lista", [{"id": "a" * 64, "nombre": "a"}]):
|
|
with self.subTest(mal=mal), self.assertRaises(ErrorDatos):
|
|
escribir(self.carpeta.parent / "otra", Datos(
|
|
[{"id": "F1", "nombre": "F", "tipo": "Fondo"}],
|
|
[{"id": "m1", "activo": "F1", "tipo": "Compra", "fecha": "2025-01-01", "adjuntos": mal}], [], []))
|
|
|
|
|
|
class TestServidor(Base):
|
|
def setUp(self):
|
|
super().setUp()
|
|
from cartera import servidor
|
|
self.app = servidor.App(self.c)
|
|
|
|
def test_descarga_siempre_como_fichero(self):
|
|
from cartera import servidor
|
|
mid = self.op("Compra", "2025-01-01", 1)
|
|
rutas = self.app.rutas_post()
|
|
m = rutas["/api/movimiento/adjuntar"]({"id": mid, "nombre": "página.html.pdf",
|
|
"contenido": base64.b64encode(PDF).decode()})
|
|
h = m["adjuntos"][0]["id"]
|
|
with self.assertRaisesRegex(ErrorOperacion, "no ha llegado bien"):
|
|
rutas["/api/movimiento/adjuntar"]({"id": mid, "nombre": "b.pdf", "contenido": "no es base64!"})
|
|
self.assertEqual(self.app.estado()["adjuntos_faltan"], [])
|
|
|
|
httpd = servidor.ThreadingHTTPServer(("127.0.0.1", 0), servidor.crear_manejador(self.app))
|
|
self.addCleanup(httpd.server_close)
|
|
self.app.puerto = httpd.server_address[1] # lo que hace arrancar(): el Host se comprueba
|
|
hilo = threading.Thread(target=httpd.serve_forever, daemon=True)
|
|
hilo.start()
|
|
self.addCleanup(httpd.shutdown)
|
|
puerto = httpd.server_address[1]
|
|
pedir = lambda token: urllib.request.urlopen(urllib.request.Request(
|
|
f"http://127.0.0.1:{puerto}/api/adjunto?m={mid}&a={h}",
|
|
headers={"X-Token": token}), timeout=10)
|
|
with pedir(self.app.token) as r:
|
|
self.assertEqual(r.read(), PDF)
|
|
self.assertEqual(r.headers["Content-Type"], "application/octet-stream")
|
|
self.assertIn("attachment", r.headers["Content-Disposition"])
|
|
self.assertIn("p%C3%A1gina.html.pdf", r.headers["Content-Disposition"])
|
|
with self.assertRaises(urllib.error.HTTPError) as e:
|
|
pedir("token-malo")
|
|
self.assertEqual(e.exception.code, 403)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|