""" 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()