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]


# ── pule kont: izolacja (PRE-29) ─────────────────────────────────────────
# Najgroźniejsza rzecz w tej zmianie. Wyciek między pulami jest CICHY: klient
# zobaczyłby cudze opisy jako własny wynik i nikt by się nie zorientował.

def _pool(root, name, plik, tresc):
    from app import files

    pula = files.tenant_root(root, name)
    pula.mkdir(parents=True, exist_ok=True)
    _xlsx(pula / plik, rows=[(tresc, f"opis {tresc}")])
    files.registry(pula)                                  # przyjęcie zastanych
    return pula


def test_each_account_sees_only_its_own_files(excel_dir):
    from app import files

    _pool(excel_dir, "klientA", "a.xlsx", "Ma Ari")
    _pool(excel_dir, "klientB", "b.xlsx", "Ve Tau")

    a = [e["name"] for e in files.registry(files.tenant_root(excel_dir, "klientA"))]
    b = [e["name"] for e in files.registry(files.tenant_root(excel_dir, "klientB"))]
    assert a == ["a.xlsx"] and b == ["b.xlsx"]


def test_search_never_reaches_another_account_pool(excel_dir):
    from app.models import SearchQuery

    _pool(excel_dir, "klientA", "a.xlsx", "Ma Ari")
    _pool(excel_dir, "klientB", "b.xlsx", "Ve Tau")
    p = _provider(excel_dir)
    p.warmup()

    swoje = p.search(SearchQuery(key="significator", value="Ma Ari", limit=10, tenant="klientA"))
    cudze = p.search(SearchQuery(key="significator", value="Ve Tau", limit=10, tenant="klientA"))
    assert swoje.total >= 1, "konto nie widzi własnej bazy"
    assert cudze.total == 0, "konto sięgnęło do puli innego konta"


def test_the_query_cache_cannot_serve_one_account_the_answer_of_another(excel_dir):
    """Bez puli w kluczu cache wynik policzony dla jednego konta trafiłby do
    drugiego — cicha wymiana treści baz, niewidoczna w logach."""
    from app.models import SearchQuery

    _pool(excel_dir, "klientA", "a.xlsx", "Ma Ari")
    _pool(excel_dir, "klientB", "b.xlsx", "Ma Ari")       # TA SAMA szukana wartość
    p = _provider(excel_dir)
    p.warmup()

    a = p.search(SearchQuery(key="significator", value="Ma Ari", limit=10, tenant="klientA"))
    b = p.search(SearchQuery(key="significator", value="Ma Ari", limit=10, tenant="klientB"))
    assert a.total >= 1 and b.total >= 1
    assert b.cache != "hit", "odpowiedź dla jednego konta wyszła z cache drugiego"


def test_a_crafted_pool_name_cannot_escape_the_share(excel_dir):
    """Nazwa puli przychodzi z zewnątrz (login), więc „../..” albo ukośnik
    wyprowadziłyby zapytanie wprost do cudzych baz."""
    from app import files

    for zlosliwa in ("../..", "../klientB", "/etc", "a/../../b", "..", "."):
        wynik = files.tenant_root(excel_dir, zlosliwa)
        assert excel_dir in wynik.parents or wynik == excel_dir, f"ucieczka: {zlosliwa} → {wynik}"


def test_no_pool_means_the_whole_share_so_production_is_untouched(excel_dir):
    """Produkcja nie wie o pulach i ma działać dokładnie jak dotąd."""
    from app import files
    from app.models import SearchQuery

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