136 lines
6.1 KiB
Python
136 lines
6.1 KiB
Python
"""
|
|
Pruebas de la versión del formato de los datos y de las migraciones.
|
|
|
|
La app aún está en el formato 1 y no tiene migraciones: aquí se simula una
|
|
versión 2 (y una 3) cambiando VERSION y MIGRACIONES solo durante cada prueba.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import datos as D # noqa: E402
|
|
from cartera.datos import Datos, ErrorDatos, escribir, leer, version_de # noqa: E402
|
|
from cartera.operaciones import Cartera # noqa: E402
|
|
|
|
|
|
def _acentos_fuera(tablas):
|
|
"""Migración de ejemplo: el tipo «Acción» pasó a llamarse «Accion»."""
|
|
for a in tablas["activos"]:
|
|
if a.get("tipo") == "Acción":
|
|
a["tipo"] = "Accion"
|
|
|
|
|
|
def _moneda_explicita(tablas):
|
|
"""Otra de ejemplo: los movimientos llevan siempre moneda."""
|
|
for m in tablas["movimientos"]:
|
|
m.setdefault("moneda", "EUR")
|
|
|
|
|
|
@contextmanager
|
|
def formato(version, migraciones):
|
|
with mock.patch.object(D, "VERSION", version), \
|
|
mock.patch.dict(D.MIGRACIONES, migraciones, clear=True):
|
|
yield
|
|
|
|
|
|
def jsonl(ruta: Path, registros) -> None:
|
|
ruta.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in registros), encoding="utf-8")
|
|
|
|
|
|
class TestMigraciones(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmp.cleanup)
|
|
self.carpeta = Path(self._tmp.name) / "datos"
|
|
self.carpeta.mkdir()
|
|
|
|
def datos_v1(self):
|
|
"""Datos tal como los dejaría la app en formato 1 (sin version.json)."""
|
|
jsonl(self.carpeta / "activos.jsonl", [{"id": "A1", "nombre": "Acme", "tipo": "Acción"}])
|
|
jsonl(self.carpeta / "movimientos.jsonl", [{"id": "m1", "fecha": "2025-01-02", "activo": "A1",
|
|
"tipo": "Compra", "cantidad": 3, "precio": 10}])
|
|
|
|
def test_la_app_declara_todas_sus_migraciones(self):
|
|
self.assertEqual(sorted(D.MIGRACIONES), list(range(2, D.VERSION + 1)))
|
|
|
|
def test_sin_version_es_la_1_y_al_escribir_se_apunta(self):
|
|
self.assertEqual(version_de(self.carpeta), 1)
|
|
escribir(self.carpeta, Datos())
|
|
self.assertEqual(json.loads((self.carpeta / "version.json").read_text()), {"version": D.VERSION})
|
|
self.assertFalse(any(p.name.startswith("copia_v") for p in self.carpeta.iterdir()))
|
|
|
|
def test_leer_migra_en_memoria_sin_tocar_el_disco(self):
|
|
self.datos_v1()
|
|
antes = {p.name: p.read_bytes() for p in self.carpeta.iterdir()}
|
|
with self.assertRaises(ErrorDatos):
|
|
leer(self.carpeta) # en formato 1 «Acción» no existe
|
|
with formato(2, {2: _acentos_fuera}):
|
|
self.assertEqual(leer(self.carpeta).activos[0]["tipo"], "Accion")
|
|
self.assertEqual({p.name: p.read_bytes() for p in self.carpeta.iterdir()}, antes)
|
|
|
|
def test_al_abrir_se_guarda_migrado_con_copia_del_original(self):
|
|
self.datos_v1()
|
|
original = (self.carpeta / "activos.jsonl").read_bytes()
|
|
with formato(3, {2: _acentos_fuera, 3: _moneda_explicita}):
|
|
Cartera(self.carpeta).cerrar() # de la 1 a la 3 de una vez
|
|
self.assertEqual(version_de(self.carpeta), 3)
|
|
self.assertEqual((self.carpeta / "copia_v1" / "activos.jsonl").read_bytes(), original)
|
|
self.assertFalse((self.carpeta / "copia_v1" / "version.json").exists())
|
|
d = leer(self.carpeta)
|
|
self.assertEqual((d.activos[0]["tipo"], d.movimientos[0]["moneda"]), ("Accion", "EUR"))
|
|
|
|
c = Cartera(self.carpeta) # ya al día: no vuelve a copiar
|
|
c.guardar_movimiento({"activo": "A1", "tipo": "Venta", "fecha": "2025-02-01",
|
|
"cantidad": 1, "precio": 12})
|
|
c.cerrar()
|
|
self.assertEqual((self.carpeta / "copia_v1" / "activos.jsonl").read_bytes(), original)
|
|
self.assertEqual([p.name for p in self.carpeta.iterdir() if p.name.startswith("copia_v")],
|
|
["copia_v1"])
|
|
|
|
def test_una_escritura_cortada_se_remigra_sin_daño(self):
|
|
self.datos_v1()
|
|
with formato(3, {2: _acentos_fuera, 3: _moneda_explicita}):
|
|
migrados = leer(self.carpeta)
|
|
# Simula un corte: los .jsonl ya migrados pero version.json sin escribir
|
|
for nombre in D.FICHEROS:
|
|
jsonl(self.carpeta / f"{nombre}.jsonl", getattr(migrados, nombre))
|
|
self.assertEqual(version_de(self.carpeta), 1)
|
|
self.assertEqual(D.huella_datos(leer(self.carpeta)), D.huella_datos(migrados))
|
|
|
|
def test_datos_de_una_app_mas_nueva_no_se_abren_ni_se_pisan(self):
|
|
self.datos_v1()
|
|
(self.carpeta / "version.json").write_text(json.dumps({"version": D.VERSION + 1}))
|
|
antes = (self.carpeta / "activos.jsonl").read_bytes()
|
|
for intento in (lambda: leer(self.carpeta), lambda: Cartera(self.carpeta),
|
|
lambda: escribir(self.carpeta, Datos())):
|
|
with self.subTest(intento=intento), self.assertRaisesRegex(ErrorDatos, "más nueva"):
|
|
intento()
|
|
self.assertEqual((self.carpeta / "activos.jsonl").read_bytes(), antes)
|
|
|
|
def test_version_ilegible(self):
|
|
for malo in ('{"version": 0}', '{"version": "2"}', '[2]', 'no es json', '{"version": true}'):
|
|
with self.subTest(malo=malo):
|
|
(self.carpeta / "version.json").write_text(malo)
|
|
with self.assertRaisesRegex(ErrorDatos, "version.json"):
|
|
leer(self.carpeta)
|
|
|
|
def test_los_errores_tras_migrar_señalan_la_linea_original(self):
|
|
jsonl(self.carpeta / "activos.jsonl", [{"id": "A1", "nombre": "Bien", "tipo": "Fondo"},
|
|
{"id": "A2", "nombre": "Mal", "tipo": "Inventado"}])
|
|
with formato(2, {2: _acentos_fuera}), self.assertRaisesRegex(ErrorDatos, r"activos\.jsonl:2"):
|
|
leer(self.carpeta)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|