astrololo/services/astrodemo/tests/test_astrodemo.py

"""astrodemo — wersja demonstracyjna (PRE-28).

DWIE WŁASNOŚCI, KTÓRYCH PILNUJE TEN PLIK

1. Demo ma DWIE funkcje i ani jednej więcej. Sprawdzamy to na trasach aplikacji,
   nie w szablonie: trasa, o której nikt nie pamiętał, jest tu realnym ryzykiem,
   bo ta usługa rozmawia z warstwą danych produkcji.

2. „Dodać plik do bazy" znaczy, że plik ZACZYNA BYĆ UŻYWANY. W pełnej aplikacji
   wgranie i włączenie to dwie decyzje (DAN-27); tutaj muszą być jedną, inaczej
   po wgraniu nic się nie zmienia i demo wygląda na zepsute.
"""
import base64

import pytest

from app import security


@pytest.fixture()
def env(monkeypatch):
    monkeypatch.setenv("ASTRODEMO_USERS", "demo:tajne-demo,klientA:hasloA,klientB:hasloB")
    monkeypatch.delenv("ASTRODEMO_USER", raising=False)
    monkeypatch.delenv("ASTRODEMO_PASSWORD", raising=False)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
    # Bez klucza podpisu usługa celowo nie wstaje (fail-closed, LOG-34).
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    monkeypatch.setenv("COOKIES_INSECURE", "1")      # TestClient jedzie po http


def _auth(user="demo", password=""):
    """Nagłówek z WAŻNĄ SESJĄ — odpowiednik bycia zalogowanym.

    Po przejściu z Basic na sesje (LOG-34) „zalogowany" znaczy „ma podpisane
    ciasteczko", nie „ma nagłówek z hasłem". Hasło przyjmowane tylko po to, żeby
    nie przepisywać wszystkich wywołań — sprawdzanie poświadczeń ma własne testy."""
    from app import security, session

    return {"Cookie": f"{session.COOKIE}={security.issue_session(user)}"}


REPORT = {
    "engine": "test",
    "objects": [{
        "object": "Sun", "sign": "Taurus", "house": 11, "direction": "D",
        "planet_token": "Su", "planet_total": 3,
        "facets": [{"label": "w znaku Taurus", "token": "Su Tau", "count": 1, "score": 5.0,
                    "samples": [{"significator": "Su Tau", "expanded": "Sun in Taurus",
                                 "effect": "wytrwały, praktyczny"}]}],
    }],
}


def _client(monkeypatch, calls=None, upload=None):
    """Podmieniamy METODY KLASY, nie obiekt modułowy: klient jest budowany per
    żądanie, żeby pula wynikała z zalogowanego konta. Dzięki temu podglądamy
    też `self.tenant`, czyli to, do której puli faktycznie poszło zapytanie."""
    from starlette.testclient import TestClient

    from app.clients.logic_client import LogicClient
    from app.main import app

    calls = calls if calls is not None else []
    domyslny = {"path": "nowa.xlsx", "name": "nowa.xlsx", "accepted": True}

    monkeypatch.setattr(LogicClient, "report",
                        lambda self, **kw: (calls.append(("report", self.tenant, kw)), REPORT)[1])
    monkeypatch.setattr(LogicClient, "files_upload",
                        lambda self, *a, **kw: (calls.append(("upload", self.tenant, a)),
                                                upload or domyslny)[1])
    monkeypatch.setattr(LogicClient, "files_status",
                        lambda self, *a, **kw: (calls.append(("status", self.tenant, a)), {})[1])
    return TestClient(app), calls


FORM = {"date": "1984-04-30", "time": "11:20", "tz_offset": "2",
        "lat": "50.06", "lon": "19.94"}


# ── powierzchnia: dokładnie dwie funkcje ────────────────────────────────

def test_the_service_exposes_nothing_beyond_its_two_functions():
    """Trasa, o której nikt nie pamiętał, jest tu realnym ryzykiem — ta usługa
    rozmawia z warstwą danych PRODUKCJI."""
    from app.main import app

    trasy = {(m, r.path) for r in app.routes for m in (getattr(r, "methods", None) or set())
             if not r.path.startswith("/static") and m in {"GET", "POST"}}
    assert trasy == {("GET", "/"), ("POST", "/"), ("POST", "/plik"), ("GET", "/health"),
                     ("GET", "/logowanie"), ("POST", "/logowanie"), ("POST", "/wyloguj")}


def test_the_client_can_only_ask_for_what_the_demo_needs():
    """Klient warstwy logicznej jest okrojony CELOWO: metody, której nie ma, nikt
    przypadkiem nie zawoła i nie podpowie ona, że po drugiej stronie jest więcej."""
    from app.clients.logic_client import LogicClient

    publiczne = {n for n in dir(LogicClient) if not n.startswith("_")}
    assert publiczne == {"report", "files_upload", "files_status", "health"}


def test_api_docs_are_off():
    """/docs wypisałoby komplet tras — demo ma nie zdradzać własnej powierzchni."""
    from app.main import app

    assert app.docs_url is None and app.openapi_url is None and app.redoc_url is None


# ── logowanie ────────────────────────────────────────────────────────────

def test_without_credentials_nothing_is_served(env, monkeypatch):
    c, _ = _client(monkeypatch)
    for method, path in (("get", "/"), ("post", "/"), ("post", "/plik")):
        r = getattr(c, method)(path, follow_redirects=False)
        assert r.status_code == 303, path
        assert "/logowanie" in r.headers["location"], path
    assert c.get("/health").status_code == 200, "sonda musi działać bez hasła"


def test_demo_account_is_separate_from_the_main_application(env, monkeypatch):
    """Konto astrodemo MUSI być własne: demo pracuje na tej samej warstwie
    danych co produkcja, więc odcięcie go nie może wymagać zmiany haseł
    w głównej aplikacji."""
    monkeypatch.setenv("APP_USER", "astrololo")
    monkeypatch.setenv("APP_PASSWORD", "haslo-glownej-aplikacji")
    c, _ = _client(monkeypatch)
    assert security.verify("astrololo", "haslo-glownej-aplikacji") is None
    assert security.verify("demo", "tajne-demo") == "demo"


def test_a_password_with_polish_letters_is_refused_not_crashed(env, monkeypatch):
    monkeypatch.setenv("ASTRODEMO_USERS", "demo:zażółć-gęślą")
    c, _ = _client(monkeypatch)



# ── funkcja 1: dodanie pliku ─────────────────────────────────────────────

def test_adding_a_file_uploads_AND_switches_it_on(env, monkeypatch):
    """Sedno wymagania: jedna czynność, nie dwie."""
    c, calls = _client(monkeypatch)
    r = c.post("/plik", headers=_auth(), follow_redirects=False,
               files={"upload": ("baza.xlsx", b"zawartosc", "application/vnd.ms-excel")})
    assert r.status_code == 303
    rodzaje = [k[0] for k in calls]
    assert rodzaje == ["upload", "status"], f"oczekiwano wgrania i włączenia, było: {rodzaje}"
    assert calls[1][2][1] == "active", "plik nie został włączony do użytku"


def test_a_rejected_file_is_not_switched_on_and_the_reason_stays_hidden(env, monkeypatch):
    """Walidacja zostaje, ale jej REGUŁY zna wyłącznie administrator (DAN-27) —
    demo nie ma prawa ich zdradzić, nawet mimochodem."""
    from urllib.parse import unquote_plus

    c, calls = _client(monkeypatch, upload={"path": "zla.xlsx", "name": "zla.xlsx",
                                            "accepted": False})
    r = c.post("/plik", headers=_auth(), follow_redirects=False,
               files={"upload": ("zla.xlsx", b"x", "text/plain")})
    assert r.status_code == 303
    assert [k[0] for k in calls] == ["upload"], "odrzucony plik został włączony do użytku"
    komunikat = unquote_plus(r.headers["location"])
    for slowo in ("walidacj", "reguł", "kolumn", "rozszerzeni", "rozmiar"):
        assert slowo not in komunikat, f"komunikat zdradza mechanizm: „{slowo}”"


def test_an_empty_file_is_refused_before_anything_is_sent(env, monkeypatch):
    c, calls = _client(monkeypatch)
    r = c.post("/plik", headers=_auth(), follow_redirects=False,
               files={"upload": ("pusty.xlsx", b"", "application/vnd.ms-excel")})
    assert r.status_code == 303 and calls == []


# ── funkcja 2: interpretacja urodzeniowa ────────────────────────────────

def test_asking_for_an_interpretation_returns_the_matches(env, monkeypatch):
    c, calls = _client(monkeypatch)
    r = c.post("/", headers=_auth(), data=FORM)
    assert r.status_code == 200
    assert "Sun in Taurus" in r.text and "wytrwały, praktyczny" in r.text
    assert [k[0] for k in calls] == ["report"]


def test_local_time_is_converted_to_utc_before_asking(env, monkeypatch):
    """11:20 przy offsecie +2 to 09:20 UTC. Pomyłka tutaj przesuwa CAŁY horoskop,
    a wynik nadal wygląda wiarygodnie — dlatego sprawdzamy wprost."""
    c, calls = _client(monkeypatch)
    c.post("/", headers=_auth(), data=FORM)
    assert calls[0][2]["when_utc_iso"].startswith("1984-04-30T09:20")


def test_a_malformed_date_is_explained_not_thrown(env, monkeypatch):
    c, calls = _client(monkeypatch)
    r = c.post("/", headers=_auth(), data={**FORM, "date": "30 kwietnia"})
    assert r.status_code == 200 and "RRRR-MM-DD" in r.text
    assert calls == [], "poszło zapytanie mimo błędnych danych"


def test_a_broken_logic_layer_is_reported_not_crashed(env, monkeypatch):
    import httpx

    from app.clients.logic_client import LogicClient

    c, _ = _client(monkeypatch)
    monkeypatch.setattr(LogicClient, "report", lambda self, **kw: (_ for _ in ()).throw(
        httpx.ConnectError("brak połączenia")))
    r = c.post("/", headers=_auth(), data=FORM)
    # Komunikat jest CELOWO nieinformacyjny: treść wyjątku niosła nazwę usługi,
    # jej port i nazwy tras, czyli mapę tego, co stoi za tym ekranem.
    assert r.status_code == 200
    assert "Nie udało się teraz wykonać tej operacji" in r.text
    assert "Warstwa logiczna" not in r.text and "8001" not in r.text


def test_the_page_never_offers_anything_it_cannot_do(env, monkeypatch):
    """Demo ma wyglądać na KOMPLETNE w swojej postaci — żadnych śladów funkcji,
    których nie ma (ta sama zasada, co przy uprawnieniach w PRE-27)."""
    c, _ = _client(monkeypatch)
    html = c.get("/", headers=_auth()).text
    for slowo in ("Kosmogram", "Synastria", "Kalendarz", "Skompiluj", "Ustawienia",
                  "Konta", "PDF", "Excel", "prompt", "model"):
        assert slowo not in html, f"strona wspomina o „{slowo}”, czego demo nie umie"


# ── pule kont: izolacja (PRE-29) ─────────────────────────────────────────
# Demo jest udostępniane szeroko i różnym osobom, więc pula musi wynikać
# z tego, KTO SIĘ ZALOGOWAŁ — i z niczego innego. Wyciek byłby cichy: klient
# zobaczyłby cudze bazy jako własne.

def test_the_pool_is_taken_from_the_login(env, monkeypatch):
    c, calls = _client(monkeypatch)
    c.post("/", headers=_auth("klientA", "hasloA"), data=FORM)
    c.post("/", headers=_auth("klientB", "hasloB"), data=FORM)
    assert [k[1] for k in calls] == ["klientA", "klientB"]


def test_uploads_go_to_the_pool_of_the_account_that_sent_them(env, monkeypatch):
    c, calls = _client(monkeypatch)
    c.post("/plik", headers=_auth("klientA", "hasloA"), follow_redirects=False,
           files={"upload": ("baza.xlsx", b"x", "application/vnd.ms-excel")})
    assert {k[1] for k in calls} == {"klientA"}, "wgranie trafiło do cudzej puli"


def test_the_pool_cannot_be_chosen_from_the_form(env, monkeypatch):
    """Gdyby pula przychodziła z żądania, wystarczyłoby podstawić cudzy login."""
    c, calls = _client(monkeypatch)
    c.post("/", headers=_auth("klientA", "hasloA"),
           data={**FORM, "tenant": "klientB", "user": "klientB", "login": "klientB"})
    assert [k[1] for k in calls] == ["klientA"], "pula dała się podmienić z formularza"


def test_every_account_has_its_own_pool_by_construction(env, monkeypatch):
    """Każdy login = inna pula. Dwa konta nie mogą dzielić zbioru baz."""
    c, calls = _client(monkeypatch)
    for login, haslo in (("demo", "tajne-demo"), ("klientA", "hasloA"), ("klientB", "hasloB")):
        c.post("/", headers=_auth(login, haslo), data=FORM)
    pule = [k[1] for k in calls]
    assert len(set(pule)) == len(pule) == 3


def test_an_unknown_account_gets_nothing(env, monkeypatch):
    """Bez ważnej sesji żądanie nie schodzi niżej — nie wystarczy, że skończy
    się przekierowaniem: nie może po drodze dotknąć warstwy logicznej."""
    c, calls = _client(monkeypatch)
    r = c.post("/", data=FORM, follow_redirects=False)
    assert r.status_code == 303 and "/logowanie" in r.headers["location"]
    assert calls == [], "poszło zapytanie mimo braku sesji"


def test_logging_in_and_out_of_the_demo(env, monkeypatch):
    """Pełny obieg: demo bywa używane na cudzych komputerach, więc wyjście musi
    naprawdę kończyć sesję, a nie tylko tak wyglądać."""
    from app import session

    c, _ = _client(monkeypatch)
    r = c.post("/logowanie", follow_redirects=False,
               data={"login": "klientA", "haslo": "hasloA", "dokad": "/"})
    assert r.status_code == 303 and session.COOKIE in r.cookies

    wyjscie = c.post("/wyloguj", headers=_auth("klientA"), follow_redirects=False)
    assert wyjscie.status_code == 303
    ustawione = wyjscie.headers.get("set-cookie", "")
    assert session.COOKIE in ustawione and "Max-Age=0" in ustawione


def test_removing_an_account_kills_its_open_session(env, monkeypatch):
    """Demo nie ma gdzie trwale zapisywać stanu, więc zdalne unieważnienie robi
    się przez ASTRODEMO_USERS — usunięcie albo zmiana hasła ubija sesję odciskiem."""
    c, _ = _client(monkeypatch)
    ciasteczko = _auth("klientA")
    assert c.get("/", headers=ciasteczko, follow_redirects=False).status_code == 200

    monkeypatch.setenv("ASTRODEMO_USERS", "demo:tajne-demo,klientB:hasloB")   # bez klientA
    assert c.get("/", headers=ciasteczko, follow_redirects=False).status_code == 303


def test_the_demo_login_cannot_be_turned_into_a_redirector(env, monkeypatch):
    c, _ = _client(monkeypatch)
    for zly in ("https://obcy.pl", "//obcy.pl", "http://obcy.pl/x"):
        r = c.post("/logowanie", follow_redirects=False,
                   data={"login": "demo", "haslo": "tajne-demo", "dokad": zly})
        assert r.headers["location"] == "/", f"przepuszczono: {zly}"