esetheyers
cf7cc9a8c7
Diseño para móvil, programa de escritorio para Linux (un solo fichero) y versiones firmadas: manifiesto de huellas, firma PGP y la clave del autor (VERIFICAR.md).
337 lines
15 KiB
Python
337 lines
15 KiB
Python
"""
|
|
Pruebas de las fuentes de precios, sin tocar internet.
|
|
|
|
Cada prueba usa una respuesta REAL guardada en `pruebas/respuestas/`, recortada
|
|
a lo que el programa lee. Así, el día que haya que cambiar algo de `precios.py`,
|
|
estas pruebas dicen si se ha roto la lectura de alguna fuente; y cuando una web
|
|
cambie de verdad, lo dirá «Comprobar las fuentes» (que sí sale a internet).
|
|
|
|
Todos los identificadores que salen aquí son públicos y genéricos —un fondo
|
|
indexado, el ETF de S&P 500 más común, bitcoin y el oro—: no dicen nada de la
|
|
cartera de nadie.
|
|
|
|
Aquí solo se prueba el núcleo. Las fuentes de gestoras concretas, si la
|
|
instalación las trae, se apartan mientras duran estas pruebas y tienen las
|
|
suyas en otro fichero.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import fuentes, precios # noqa: E402
|
|
|
|
RESPUESTAS = Path(__file__).resolve().parent / "respuestas"
|
|
|
|
# Qué fichero contesta a cada dirección
|
|
RUTAS = (
|
|
("morningstar.com/api", "morningstar.json"),
|
|
("morningstar.es/api", "morningstar.json"),
|
|
("morningstar.co.uk/api", "morningstar.json"),
|
|
("cnmv.es", "cnmv.html"),
|
|
("api/v3/simple/price", "coingecko.json"),
|
|
("currencies/xau.json", "fawaz_xau.json"),
|
|
("currencies/usd.json", "fawaz_usd.json"),
|
|
("instrument/search/main", "ls_busqueda.json"),
|
|
("chart/dataForInstrument", "ls_grafico.json"),
|
|
("ls-tc.de/de/", "ls_ficha.html"),
|
|
)
|
|
|
|
|
|
class Respuesta:
|
|
def __init__(self, texto: str, status_code: int = 200):
|
|
self.text = texto
|
|
self.status_code = status_code
|
|
|
|
def json(self):
|
|
return json.loads(self.text)
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise OSError(f"HTTP {self.status_code}")
|
|
|
|
|
|
class SesionFalsa:
|
|
"""Contesta con los ficheros guardados. `caidas` son trozos de dirección que
|
|
fallan, para simular una fuente rota."""
|
|
|
|
rutas = RUTAS
|
|
respuestas = (RESPUESTAS,) # carpetas donde buscar cada fichero, por orden
|
|
|
|
def __init__(self, cambios: dict[str, str] | None = None, caidas: tuple[str, ...] = ()):
|
|
self.cambios = cambios or {} # trozo de dirección → texto de respuesta
|
|
self.caidas = caidas
|
|
self.pedidas: list[str] = []
|
|
|
|
def get(self, url, params=None, headers=None, timeout=None):
|
|
self.pedidas.append(url)
|
|
for trozo in self.caidas:
|
|
if trozo in url:
|
|
raise OSError("la red no va")
|
|
for trozo, texto in self.cambios.items():
|
|
if trozo in url:
|
|
return Respuesta(texto)
|
|
for trozo, fichero in self.rutas:
|
|
if trozo in url:
|
|
carpeta = next(c for c in self.respuestas if (c / fichero).exists())
|
|
return Respuesta((carpeta / fichero).read_text(encoding="utf-8"))
|
|
raise AssertionError(f"la prueba no esperaba una petición a {url}")
|
|
|
|
|
|
REGISTROS = ("FONDOS_POR_ISIN", "FONDOS_POR_NOMBRE", "SONDAS_EXTRA")
|
|
|
|
|
|
class Base(unittest.TestCase):
|
|
con_gestoras = False # las pruebas del núcleo corren como la versión sin fuentes de gestoras
|
|
|
|
def setUp(self):
|
|
precios.limpiar_cache()
|
|
self.espera = precios.ESPERA_MORNINGSTAR
|
|
precios.ESPERA_MORNINGSTAR = 0 # sin cortesía con Morningstar en las pruebas
|
|
self.registros = {n: list(getattr(precios, n)) for n in REGISTROS}
|
|
if not self.con_gestoras:
|
|
for n in REGISTROS:
|
|
getattr(precios, n).clear()
|
|
|
|
def tearDown(self):
|
|
precios.ESPERA_MORNINGSTAR = self.espera
|
|
for n in REGISTROS:
|
|
getattr(precios, n)[:] = self.registros[n]
|
|
precios.limpiar_cache()
|
|
|
|
|
|
class TestFondos(Base):
|
|
def test_morningstar_lee_precio_y_fecha(self):
|
|
p = precios.morningstar("IE00B03HCZ61", SesionFalsa())
|
|
self.assertEqual(p.precio, 200.8208)
|
|
self.assertEqual(p.fecha, dt.date(2026, 9, 21))
|
|
self.assertEqual(p.fuente, "morningstar")
|
|
|
|
def test_morningstar_no_confunde_el_precio_con_miles(self):
|
|
"""El JSON trae 200.8208: el punto es decimal, no separador de miles."""
|
|
p = precios.morningstar("IE00B03HCZ61", SesionFalsa())
|
|
self.assertLess(p.precio, 1000)
|
|
|
|
def test_morningstar_sin_resultados_no_es_un_fallo(self):
|
|
sesion = SesionFalsa({"morningstar": (RESPUESTAS / "morningstar_sin_resultados.json").read_text()})
|
|
avisos: list[str] = []
|
|
self.assertIsNone(precios.morningstar("ES0000000000", sesion, avisos))
|
|
self.assertEqual(avisos, []) # el ISIN no está: la fuente funciona
|
|
|
|
def test_morningstar_caido_prueba_los_tres_hosts_y_avisa(self):
|
|
sesion = SesionFalsa(caidas=("morningstar",))
|
|
avisos: list[str] = []
|
|
self.assertIsNone(precios.morningstar("IE00B03HCZ61", sesion, avisos))
|
|
self.assertEqual(len(sesion.pedidas), 3)
|
|
self.assertIn("Morningstar", avisos[0])
|
|
|
|
def test_cnmv_avisa_de_que_ya_no_publica_el_valor(self):
|
|
avisos: list[str] = []
|
|
self.assertIsNone(precios.cnmv("IE00B03HCZ61", SesionFalsa(), avisos))
|
|
self.assertIn("ya no publica", avisos[0])
|
|
|
|
|
|
|
|
class TestOtrasFuentes(Base):
|
|
def test_coingecko(self):
|
|
p = precios.coingecko("bitcoin", SesionFalsa())
|
|
self.assertEqual(p.precio, 75368)
|
|
self.assertEqual(p.fuente, "coingecko")
|
|
|
|
def test_coingecko_moneda_desconocida(self):
|
|
sesion = SesionFalsa({"simple/price": "{}"})
|
|
self.assertIsNone(precios.coingecko("no-existe", sesion))
|
|
|
|
def test_metal_oro_por_onza(self):
|
|
p = precios.metal("xau", SesionFalsa())
|
|
self.assertAlmostEqual(p.precio, 3779.94311611, places=5)
|
|
self.assertEqual(p.fecha, dt.date(2026, 9, 22))
|
|
|
|
def test_ls_exchange_valora_al_precio_de_compra_como_las_apps_de_los_brokers(self):
|
|
p = precios.ls_exchange("IE00B5BMR087", SesionFalsa())
|
|
self.assertEqual(p.precio, 730.66) # el de compra (bid), no el medio (731,03)
|
|
self.assertEqual(p.fuente, "ls-exchange")
|
|
|
|
def test_ls_exchange_sin_la_ficha_usa_el_precio_medio_y_lo_dice(self):
|
|
avisos = []
|
|
p = precios.ls_exchange("IE00B5BMR087", SesionFalsa(caidas=("ls-tc.de/de/",)), avisos)
|
|
self.assertEqual(p.precio, 731.03)
|
|
self.assertIn("uso el medio", avisos[0])
|
|
|
|
def test_ls_exchange_no_se_fia_de_un_precio_de_compra_disparatado(self):
|
|
ficha = (RESPUESTAS / "ls_ficha.html").read_text().replace("730,6600", "73,0660")
|
|
p = precios.ls_exchange("IE00B5BMR087", SesionFalsa({"ls-tc.de/de/": ficha}))
|
|
self.assertEqual(p.precio, 731.03)
|
|
|
|
def test_ls_exchange_con_el_mercado_cerrado_usa_el_cierre_anterior(self):
|
|
sesion = SesionFalsa({"dataForInstrument": (RESPUESTAS / "ls_grafico_cerrado.json").read_text()})
|
|
p = precios.ls_exchange("IE00B5BMR087", sesion)
|
|
self.assertEqual(p.precio, 730.66) # el último precio de compra; la fecha, la del cierre
|
|
self.assertLess(p.fecha, dt.date.today())
|
|
self.assertLess(p.fecha.weekday(), 5) # nunca un sábado o un domingo
|
|
|
|
def test_ls_exchange_cerrado_y_sin_ficha_usa_el_cierre_anterior(self):
|
|
sesion = SesionFalsa({"dataForInstrument": (RESPUESTAS / "ls_grafico_cerrado.json").read_text()},
|
|
caidas=("ls-tc.de/de/",))
|
|
self.assertEqual(precios.ls_exchange("IE00B5BMR087", sesion).precio, 730.1)
|
|
|
|
def test_ls_exchange_isin_desconocido(self):
|
|
sesion = SesionFalsa({"search/main": "[]"})
|
|
self.assertIsNone(precios.ls_exchange("XX0000000000", sesion))
|
|
|
|
def test_cambio_de_dolares_a_euros(self):
|
|
valor = precios.tipo_cambio(dt.date(2026, 9, 22), "usd", "eur", SesionFalsa())
|
|
self.assertAlmostEqual(valor, 0.8533)
|
|
|
|
def test_a_euros_no_toca_los_euros(self):
|
|
self.assertEqual(precios.a_euros(100.0, "EUR", SesionFalsa()), 100.0)
|
|
|
|
|
|
def grafico_yahoo(cierres, moneda="EUR") -> str:
|
|
"""Respuesta inventada de la API de gráficos de Yahoo: [(fecha, cierre|None)]."""
|
|
marcas = [int(dt.datetime(f.year, f.month, f.day, 7, tzinfo=dt.timezone.utc).timestamp()) for f, _ in cierres]
|
|
return json.dumps({"chart": {"result": [{"meta": {"currency": moneda}, "timestamp": marcas,
|
|
"indicators": {"quote": [{"close": [c for _, c in cierres]}]}}],
|
|
"error": None}})
|
|
|
|
|
|
class TestYahooFondos(Base):
|
|
def test_ultimo_cierre_publicado_y_su_fecha(self):
|
|
texto = grafico_yahoo([(dt.date(2026, 9, 21), 15.1), (dt.date(2026, 9, 22), 15.2),
|
|
(dt.date(2026, 9, 23), None), (dt.date(2026, 9, 24), None)])
|
|
p = precios.yahoo_fondo("0P0000TEST", SesionFalsa({"finance.yahoo.com/v8": texto}))
|
|
self.assertEqual((p.precio, p.fecha, p.fuente, p.codigo), (15.2, dt.date(2026, 9, 22), "yahoo", "0P0000TEST.F"))
|
|
|
|
def test_en_dolares_se_pasa_a_euros(self):
|
|
texto = grafico_yahoo([(dt.date(2026, 9, 22), 100.0)], moneda="USD")
|
|
p = precios.yahoo_fondo("0P0000TEST.F", SesionFalsa({"finance.yahoo.com/v8": texto}))
|
|
self.assertAlmostEqual(p.precio, 100.0 * precios.tipo_cambio(dt.date(2026, 9, 22), "usd", "eur",
|
|
SesionFalsa()))
|
|
|
|
def test_una_accion_por_su_ticker_sin_bibliotecas_de_mas(self):
|
|
sesion = SesionFalsa({"finance.yahoo.com/v8": grafico_yahoo([(dt.date(2026, 9, 21), 100.0), (dt.date(2026, 9, 22), None)], "USD")})
|
|
p = precios.yahoo("spy", sesion)
|
|
self.assertEqual((p.fecha, p.fuente), (dt.date(2026, 9, 21), "yahoo"))
|
|
self.assertAlmostEqual(p.precio, precios.a_euros(100.0, "USD", sesion))
|
|
self.assertTrue(any("/SPY" in u for u in sesion.pedidas))
|
|
self.assertNotIn("yfinance", sys.modules)
|
|
|
|
def test_simbolo_desconocido_o_sin_cierres(self):
|
|
avisos = []
|
|
vacio = json.dumps({"chart": {"result": None, "error": {"code": "Not Found"}}})
|
|
self.assertIsNone(precios.yahoo_fondo("0P0000NADA", SesionFalsa({"finance.yahoo.com/v8": vacio}), avisos))
|
|
self.assertIn("Yahoo no tiene 0P0000NADA.F", avisos)
|
|
sin_cierres = grafico_yahoo([(dt.date(2026, 9, 22), None)])
|
|
self.assertIsNone(precios.yahoo_fondo("0P0000TEST", SesionFalsa({"finance.yahoo.com/v8": sin_cierres})))
|
|
|
|
|
|
class TestQueFuenteLeTocaACadaActivo(Base):
|
|
def elegir(self, activo, sesion=None):
|
|
return precios.de_activo(activo, sesion or SesionFalsa())
|
|
|
|
def test_fondo_con_isin_va_a_morningstar(self):
|
|
p = self.elegir({"id": "F1", "tipo": "Fondo", "codigo": "IE00B03HCZ61", "nombre": "Un fondo"})
|
|
self.assertEqual(p.fuente, "morningstar")
|
|
self.assertEqual(p.activo, "F1")
|
|
|
|
def test_plan_con_codigo_dgs_no_pregunta_a_yahoo(self):
|
|
sesion = SesionFalsa()
|
|
self.elegir({"id": "P1", "tipo": "PlanPensiones", "codigo": "N0000", "nombre": "Plan De Ejemplo Ppi"},
|
|
sesion)
|
|
self.assertFalse(any("yahoo" in u for u in sesion.pedidas))
|
|
|
|
def test_cripto_va_a_coingecko(self):
|
|
self.assertEqual(self.elegir({"id": "C1", "tipo": "Cripto", "codigo": "bitcoin"}).fuente, "coingecko")
|
|
|
|
def test_metal_va_a_fawaz(self):
|
|
self.assertEqual(self.elegir({"id": "M1", "tipo": "Metal", "codigo": "xau"}).fuente, "fawaz")
|
|
|
|
def test_etf_con_isin_va_a_ls_exchange(self):
|
|
"""En euros, en un mercado alemán que cotiza de día y de tarde."""
|
|
self.assertEqual(self.elegir({"id": "E1", "tipo": "ETF", "codigo": "IE00B5BMR087"}).fuente, "ls-exchange")
|
|
|
|
def test_accion_con_ticker_va_a_yahoo(self):
|
|
llamadas = []
|
|
original = precios.yahoo
|
|
precios.yahoo = lambda t, s, a=None: llamadas.append(t) or precios.Precio("", t, 1.0, dt.date.today(), "yahoo")
|
|
try:
|
|
p = self.elegir({"id": "A1", "tipo": "Accion", "codigo": "SPY"})
|
|
finally:
|
|
precios.yahoo = original
|
|
self.assertEqual(llamadas, ["SPY"])
|
|
self.assertEqual(p.fuente, "yahoo")
|
|
|
|
def test_un_activo_de_precio_manual_no_consulta_nada(self):
|
|
sesion = SesionFalsa()
|
|
self.assertIsNone(self.elegir({"id": "CC1", "tipo": "Cuenta", "codigo": ""}, sesion))
|
|
self.assertEqual(sesion.pedidas, [])
|
|
|
|
class TestComprobarLasFuentes(Base):
|
|
def test_todas_bien_menos_la_cnmv(self):
|
|
informe = precios.comprobar(SesionFalsa({"finance.yahoo": "", "query": ""}))
|
|
por_clave = {f["fuente"]: f for f in informe["fuentes"]}
|
|
for clave in ("morningstar", "coingecko", "fawaz", "ls-exchange", "cambio"):
|
|
self.assertTrue(por_clave[clave]["ok"], f"{clave}: {por_clave[clave]['detalle']}")
|
|
self.assertFalse(por_clave["cnmv"]["ok"])
|
|
self.assertTrue(por_clave["cnmv"]["esperado"]) # rota a sabiendas desde 2026
|
|
self.assertNotIn("CNMV", informe["rotas"])
|
|
|
|
def test_dice_cual_esta_rota(self):
|
|
informe = precios.comprobar(SesionFalsa(caidas=("morningstar", "finance.yahoo")))
|
|
self.assertIn("Morningstar", informe["rotas"])
|
|
self.assertNotIn("CoinGecko", informe["rotas"])
|
|
|
|
def test_no_consulta_ningun_activo_de_la_cartera(self):
|
|
"""Las sondas son identificadores públicos fijos, escritos en el código."""
|
|
sesion = SesionFalsa({"finance.yahoo": "", "query": ""})
|
|
precios.comprobar(sesion)
|
|
for url in sesion.pedidas:
|
|
self.assertNotIn("ES0113728", url) # ningún ISIN que no sea el de la sonda
|
|
|
|
|
|
class TestLoQueUsaLaAplicacion(Base):
|
|
def test_fuentes_precio_devuelve_el_registro_listo_para_guardar(self):
|
|
r, mensaje = fuentes.precio({"id": "C1", "tipo": "Cripto", "codigo": "bitcoin"}, SesionFalsa())
|
|
self.assertEqual(r, {"activo": "C1", "fecha": r["fecha"], "precio": 75368.0, "fuente": "coingecko"})
|
|
self.assertEqual(mensaje, "")
|
|
|
|
def test_fuentes_precio_explica_por_que_no_hay_precio(self):
|
|
sesion = SesionFalsa(caidas=("coingecko",))
|
|
r, mensaje = fuentes.precio({"id": "C1", "tipo": "Cripto", "codigo": "bitcoin"}, sesion)
|
|
self.assertIsNone(r)
|
|
self.assertIn("CoinGecko", mensaje)
|
|
|
|
def test_una_fuente_rota_no_tumba_la_actualizacion_entera(self):
|
|
activos = [{"id": "C1", "tipo": "Cripto", "codigo": "bitcoin"},
|
|
{"id": "M1", "tipo": "Metal", "codigo": "xau"}]
|
|
sesion = SesionFalsa(caidas=("coingecko",))
|
|
informe = [dict(zip(("activo", "ok"), (a["id"], fuentes.precio(a, sesion)[0] is not None)))
|
|
for a in activos]
|
|
self.assertEqual(informe, [{"activo": "C1", "ok": False}, {"activo": "M1", "ok": True}])
|
|
|
|
def test_la_ruta_de_comprobar_esta_conectada(self):
|
|
import tempfile
|
|
|
|
from cartera import servidor
|
|
from cartera.operaciones import Cartera
|
|
|
|
original = fuentes.comprobar
|
|
fuentes.comprobar = lambda: {"fuentes": [], "rotas": [], "fecha": "2026-09-22T00:00:00"}
|
|
try:
|
|
with tempfile.TemporaryDirectory() as carpeta:
|
|
rutas = servidor.App(Cartera(Path(carpeta))).rutas_post()
|
|
self.assertEqual(rutas["/api/fuentes/comprobar"]({})["rotas"], [])
|
|
finally:
|
|
fuentes.comprobar = original
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|