astrololo/services/data/tests/test_rejestr_integracja.py

"""Rejestr plików a RESZTA warstwy danych — punkty styku (DAN-27).

DLACZEGO OSOBNY PLIK. test_files.py sprawdza sam rejestr w izolacji i przechodził
na zielono, podczas gdy na produkcji leżało wyszukiwanie (500) i lista baz (502).
Rejestr wszedł w miejsce starego mechanizmu włączania baz, więc groźne jest nie
to, co robi w środku, tylko czy MÓWI TYM SAMYM JĘZYKIEM, co jego odbiorcy.

Oba tamte błędy były jednolinijkowe i oba niewidoczne dla testów jednostkowych:
  * NameError, bo przepisując `_enabled_files` usunąłem lokalny import `bases`,
  * KeyError, bo rejestr oddawał `in_use`, a endpoint /bases czytał `enabled`.
"""
import pathlib

import pytest


@pytest.fixture()
def excel_dir(tmp_path, monkeypatch):
    monkeypatch.setenv("EXCEL_DIR", str(tmp_path))
    monkeypatch.setenv("CACHE_DIR", str(tmp_path / ".cache"))
    monkeypatch.delenv("DISABLED_BASES", raising=False)
    monkeypatch.delenv("FILES_STATE", raising=False)
    return tmp_path


def _xlsx(path, rows=(("Ma Ari", "odważny"), ("Ve Tau", "zgodny"))):
    import openpyxl

    wb = openpyxl.Workbook()
    ws = wb.active
    ws.append(["significator", "effect"])
    for r in rows:
        ws.append(list(r))
    wb.save(path)
    return path


def _provider(excel_dir):
    from app.config import Settings
    from app.providers.excel_provider import ExcelDataProvider

    return ExcelDataProvider(Settings())


def test_search_does_not_explode_on_the_registry(excel_dir):
    """Regresja: `_enabled_files` wołało bases.disabled_entries() bez importu,
    więc KAŻDE wyszukiwanie kończyło się NameError → 500 z warstwy danych."""
    from app.models import SearchQuery

    _xlsx(excel_dir / "baza.xlsx")
    p = _provider(excel_dir)
    p.warmup()
    out = p.search(SearchQuery(key="significator", value="Ma Ari", exact=False, limit=10))
    assert out.total >= 1, "zastana baza nie weszła do wyszukiwania"


def test_search_still_works_with_disabled_bases_set(excel_dir, monkeypatch):
    """DISABLED_BASES zostaje jako awaryjne wyłączenie i ma odsiewać DODATKOWO —
    to właśnie ta gałąź kodu wywalała się na braku importu."""
    from app.models import SearchQuery

    _xlsx(excel_dir / "a.xlsx")
    _xlsx(excel_dir / "b.xlsx")
    monkeypatch.setenv("DISABLED_BASES", "b.xlsx")
    p = _provider(excel_dir)
    p.warmup()
    out = p.search(SearchQuery(key="significator", value="Ma Ari", exact=False, limit=10))
    assert out.total >= 1


def test_registry_speaks_the_language_the_bases_endpoint_reads(excel_dir):
    """Regresja: endpoint /bases liczy `b["enabled"]`, rejestr oddawał `in_use`.
    KeyError → 500 z danych → 502 z logiki → „Warstwa logiczna niedostępna"
    na ekranie Ustawienia."""
    _xlsx(excel_dir / "baza.xlsx")
    items = _provider(excel_dir).list_bases()
    assert items, "lista baz jest pusta"
    for row in items:
        for key in ("name", "path", "enabled", "in_use", "size_mb", "modified"):
            assert key in row, f"brak pola `{key}` — odbiorca dostanie KeyError"
        assert row["enabled"] == row["in_use"], "dwa pola, jedna prawda"


def test_bases_endpoint_answers_end_to_end(excel_dir):
    """Przez TRASĘ, nie przez dostawcę: to ona wywalała się na produkcji."""
    _xlsx(excel_dir / "baza.xlsx")
    import importlib

    from app import main as data_main

    importlib.reload(data_main)
    body = data_main.bases()
    assert body["total"] == 1
    assert body["enabled"] == 1, "zastana baza powinna być włączona po adopcji"
    assert body["bases"][0]["name"] == "baza.xlsx"


def test_switching_a_base_off_is_visible_in_both_places(excel_dir):
    """Odstawienie bazy ma zniknąć i z wyszukiwania, i z licznika na Ustawieniach."""
    from app import files
    from app.models import SearchQuery

    _xlsx(excel_dir / "baza.xlsx")
    p = _provider(excel_dir)
    p.warmup()
    assert p.search(SearchQuery(key="significator", value="Ma Ari", limit=10)).total >= 1

    files.set_status(excel_dir, "baza.xlsx", files.READY)
    p2 = _provider(excel_dir)
    p2.warmup()
    assert p2.search(SearchQuery(key="significator", value="Ma Ari", limit=10)).total == 0
    assert [b["enabled"] for b in p2.list_bases()] == [False]