astrololo/services/presentation/tests/test_pliki.py

"""Trzy poziomy dostępu do plików baz (DAN-27 × PRE-27).

  user        („files")       — widzi listę i decyduje, z czego program korzysta,
  data_input  (+„files_input")— dokłada wgrywanie i ARCHIWIZACJĘ,
  administrator               — kasowanie, przywracanie i REGUŁY WALIDACJI.

Osobno pilnujemy własności negatywnej: poza administratorem NIKT nie ma skąd
wiedzieć, że walidacja w ogóle istnieje. To łatwo zepsuć dobrą intencją —
„pokażmy człowiekowi, czemu plik nie przeszedł" — więc testy są wprost o tym.
"""
import pytest

from app import accounts as store
from app import features


@pytest.fixture()
def env(tmp_path, monkeypatch):
    monkeypatch.setenv("ACCOUNTS_FILE", str(tmp_path / "accounts.json"))
    monkeypatch.setenv("APP_USER", "szef")
    monkeypatch.setenv("APP_PASSWORD", "tajne-szefa")
    monkeypatch.delenv("APP_USERS", raising=False)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
    return tmp_path


def _auth(user, password):
    import base64

    return {"Authorization": "Basic " + base64.b64encode(
        f"{user}:{password}".encode()).decode()}


REGISTRY = [
    {"name": "glowna.xlsx", "path": "glowna.xlsx", "size_mb": 1.0, "modified": "2026-01-01",
     "status": "active", "in_use": True, "archived_at": "", "uploaded_at": "",
     "uploaded_by": "", "sha256": "abc"},
    {"name": "stara.xlsx", "path": "stara.xlsx", "size_mb": 2.0, "modified": "2025-01-01",
     "status": "archived", "in_use": False, "archived_at": "2026-02-02T10:00:00+00:00",
     "uploaded_at": "", "uploaded_by": "", "sha256": "def"},
]
HELD = {"name": "podejrzana.xlsx", "path": "podejrzana.xlsx", "size_mb": 0.1,
        "modified": "2026-03-03", "status": "quarantine", "in_use": False,
        "archived_at": "", "uploaded_at": "2026-03-03T09:00:00+00:00",
        "uploaded_by": "dane", "sha256": "ghi",
        "rejected_for": ["brak wymaganych kolumn: znaczenie"]}
RULES = {"extensions": [".xlsx"], "max_size_mb": 50, "min_rows": 1,
         "required_columns": ["id", "znaczenie"], "reject_duplicate_content": True}


def _client(monkeypatch, calls=None):
    """Klient z zastubowaną warstwą logiczną. `calls` zbiera to, co poszło w dół."""
    from starlette.testclient import TestClient

    from app.main import app, logic

    calls = calls if calls is not None else []

    def files_list(for_admin=False):
        rows = list(REGISTRY) + ([HELD] if for_admin else [])
        return {"files": rows, "rules": RULES if for_admin else {}, "root": "/x"}

    monkeypatch.setattr(logic, "files_list", files_list)
    for name in ("files_status", "files_upload", "files_delete", "files_rules"):
        monkeypatch.setattr(logic, name,
                            lambda *a, _n=name, **kw: (calls.append((_n, a, kw)),
                                                       {"accepted": True, "name": "x.xlsx"})[1])
    return TestClient(app), calls


# ── poziom „user" ────────────────────────────────────────────────────────

def test_user_sees_the_list_and_can_switch_bases_on_and_off(env, monkeypatch):
    store.create("ula", "x", ["files"])
    c, calls = _client(monkeypatch)
    html = c.get("/files", headers=_auth("ula", "x")).text
    assert "glowna.xlsx" in html and "stara.xlsx" in html

    r = c.post("/files/use", headers=_auth("ula", "x"), follow_redirects=False,
               data={"path": "glowna.xlsx", "use": "0"})
    assert r.status_code == 303
    assert calls and calls[0][0] == "files_status"


def test_user_gets_no_upload_no_archive_no_delete(env, monkeypatch):
    store.create("ula", "x", ["files"])
    c, _ = _client(monkeypatch)
    html = c.get("/files", headers=_auth("ula", "x")).text
    for absent in ("/files/upload", "/files/archive", "/files/delete",
                   "/files/restore", "/files/rules"):
        assert absent not in html, f"strona zdradza {absent}"
    for path in ("/files/upload", "/files/archive", "/files/delete",
                 "/files/restore", "/files/rules"):
        r = c.post(path, headers=_auth("ula", "x"), data={"path": "glowna.xlsx"})
        assert r.status_code == 404, f"{path} → {r.status_code}"


# ── poziom „data_input" ──────────────────────────────────────────────────

def test_data_input_can_upload_and_archive(env, monkeypatch):
    store.create("dane", "x", ["files", "files_input"])
    c, calls = _client(monkeypatch)
    html = c.get("/files", headers=_auth("dane", "x")).text
    assert "/files/upload" in html and "/files/archive" in html

    r = c.post("/files/upload", headers=_auth("dane", "x"), follow_redirects=False,
               files={"upload": ("nowa.xlsx", b"zawartosc", "application/vnd.ms-excel")})
    assert r.status_code == 303 and calls[-1][0] == "files_upload"

    r = c.post("/files/archive", headers=_auth("dane", "x"), follow_redirects=False,
               data={"path": "glowna.xlsx"})
    assert r.status_code == 303
    assert calls[-1][1][1] == "archived", "archiwizacja ma ustawiać stan `archived`"


def test_data_input_cannot_delete_restore_or_set_rules(env, monkeypatch):
    """Wymóg wprost: osoba wgrywająca dane może CO NAJWYŻEJ zarchiwizować."""
    store.create("dane", "x", ["files", "files_input"])
    c, _ = _client(monkeypatch)
    html = c.get("/files", headers=_auth("dane", "x")).text
    for absent in ("/files/delete", "/files/restore", "/files/rules"):
        assert absent not in html, f"strona zdradza {absent}"
    for path in ("/files/delete", "/files/restore", "/files/rules"):
        assert c.post(path, headers=_auth("dane", "x"),
                      data={"path": "glowna.xlsx"}).status_code == 404


# ── własność negatywna: walidacja jest tajemnicą administratora ─────────

@pytest.mark.parametrize("perms", [["files"], ["files", "files_input"]])
def test_nobody_below_admin_learns_that_validation_exists(env, monkeypatch, perms):
    store.create("ktos", "x", perms)
    c, _ = _client(monkeypatch)
    html = c.get("/files", headers=_auth("ktos", "x")).text
    assert "podejrzana.xlsx" not in html, "plik wstrzymany nie ma prawa się pokazać"
    for leak in ("walidacj", "Walidacj", "reguł", "Reguł", "znaczenie",
                 "brak wymaganych kolumn", "rozszerzeni", "Maksymalny rozmiar"):
        assert leak not in html, f"strona zdradza mechanizm: „{leak}”"


def test_refusal_to_activate_gives_no_reason_below_admin(env, monkeypatch):
    """Powód odmowy zdradzałby regułę. Komunikat ma być bez treści."""
    import httpx

    from app.main import app, logic
    from starlette.testclient import TestClient

    store.create("ula", "x", ["files"])

    def boom(*a, **kw):
        raise httpx.HTTPStatusError(
            "422", request=httpx.Request("POST", "http://x"),
            response=httpx.Response(422, json={"detail": "brak wymaganych kolumn: znaczenie"}))

    monkeypatch.setattr(logic, "files_list", lambda for_admin=False: {"files": [], "rules": {}})
    monkeypatch.setattr(logic, "files_status", boom)
    r = TestClient(app).post("/files/use", headers=_auth("ula", "x"),
                             follow_redirects=False, data={"path": "x.xlsx", "use": "1"})
    assert r.status_code == 303
    assert "znaczenie" not in r.headers["location"]
    assert "kolumn" not in r.headers["location"]


# ── administrator ────────────────────────────────────────────────────────

def test_admin_sees_held_files_with_the_reason_and_the_rules(env, monkeypatch):
    c, _ = _client(monkeypatch)
    html = c.get("/files", headers=_auth("szef", "tajne-szefa")).text
    assert "podejrzana.xlsx" in html
    assert "brak wymaganych kolumn: znaczenie" in html
    assert "Reguły przyjmowania baz" in html
    for op in ("/files/delete", "/files/restore", "/files/rules",
               "/files/upload", "/files/archive"):
        assert op in html, f"administratorowi brakuje {op}"


def test_admin_can_delete_restore_and_change_the_rules(env, monkeypatch):
    c, calls = _client(monkeypatch)
    admin = _auth("szef", "tajne-szefa")
    c.post("/files/delete", headers=admin, follow_redirects=False, data={"path": "a.xlsx"})
    c.post("/files/restore", headers=admin, follow_redirects=False, data={"path": "a.xlsx"})
    c.post("/files/rules", headers=admin, follow_redirects=False,
           data={"extensions": ".xlsx, .xlsm", "max_size_mb": "10", "min_rows": "5",
                 "required_columns": "id, znaczenie", "reject_duplicate_content": "1"})
    names = [c[0] for c in calls]
    assert names == ["files_delete", "files_status", "files_rules"]
    sent = calls[-1][1][0]
    assert sent["extensions"] == [".xlsx", ".xlsm"]
    assert sent["required_columns"] == ["id", "znaczenie"]
    assert sent["reject_duplicate_content"] is True


def test_archived_file_cannot_be_switched_on_from_the_page(env, monkeypatch):
    """Archiwum jest ZAMROŻONE: przełącznik przy nim w ogóle się nie pojawia,
    a przywrócić może wyłącznie administrator."""
    store.create("dane", "x", ["files", "files_input"])
    c, _ = _client(monkeypatch)
    html = c.get("/files", headers=_auth("dane", "x")).text
    row = html[html.index("stara.xlsx") - 700:html.index("stara.xlsx")]
    assert "/files/use" not in row, "zarchiwizowana baza ma przełącznik użycia"


def test_files_tab_is_hidden_without_the_permission(env, monkeypatch):
    store.create("bez", "x", ["chart"])
    c, _ = _client(monkeypatch)
    html = c.get("/", headers=_auth("bez", "x")).text
    assert 'href="/files"' not in html and "Pliki" not in html
    assert c.get("/files", headers=_auth("bez", "x")).status_code == 404


def test_route_map_covers_every_file_route():
    for route in ("/files", "/files/use", "/files/upload", "/files/archive",
                  "/files/restore", "/files/delete", "/files/rules"):
        method = "GET" if route == "/files" else "POST"
        assert (method, route) in features.ROUTES, route