"""Rejestr plików baz: stany, walidacja, archiwizacja (DAN-27). Testujemy tu RDZEŃ — bez HTTP i bez uprawnień, bo uprawnienia rozstrzyga prezentacja (patrz services/presentation/tests/test_pliki.py). Tutaj chodzi o to, żeby żadna operacja nie gubiła pliku i żeby bramka „do użytku tylko po walidacji" faktycznie trzymała. """ import pathlib import pytest from app import files @pytest.fixture() def root(tmp_path, monkeypatch): monkeypatch.delenv("FILES_STATE", raising=False) return tmp_path def _xlsx(path, rows=3, header=("id", "opis")): import openpyxl wb = openpyxl.Workbook() ws = wb.active ws.append(list(header)) for i in range(rows): ws.append([i, f"treść {i}"]) wb.save(path) return path # ── rejestr i stany ────────────────────────────────────────────────────── def test_new_file_is_visible_but_not_in_use(root): _xlsx(root / "baza.xlsx") entry = files.registry(root)[0] assert entry["status"] == files.READY assert entry["in_use"] is False, "nowy plik nie może sam wejść do wyszukiwania" def test_only_active_files_reach_the_search(root): _xlsx(root / "a.xlsx") _xlsx(root / "b.xlsx") assert files.usable_paths(root) == [] files.set_status(root, "a.xlsx", files.ACTIVE) assert [pathlib.Path(p).name for p in files.usable_paths(root)] == ["a.xlsx"] def test_state_survives_a_restart(root): """Stan jest KLIKANY, więc musi być trwały — inaczej restart poda po cichu przywracałby bazy wyłączone świadomie.""" _xlsx(root / "a.xlsx") files.set_status(root, "a.xlsx", files.ACTIVE) assert files.state_path(root).exists() assert files.registry(root)[0]["in_use"] is True # ── archiwizacja ───────────────────────────────────────────────────────── def test_archiving_freezes_the_file_but_never_removes_it(root): """To jest najdalej idąca operacja osoby wgrywającej dane: plik ZOSTAJE.""" p = _xlsx(root / "stara.xlsx") files.set_status(root, "stara.xlsx", files.ACTIVE) files.set_status(root, "stara.xlsx", files.ARCHIVED, by="dane") entry = files.registry(root)[0] assert p.exists(), "plik zniknął z dysku — archiwizacja ma go zachować" assert entry["status"] == files.ARCHIVED assert entry["in_use"] is False assert entry["archived_at"], "brak znacznika czasu archiwizacji" def test_archived_file_cannot_slip_back_into_use_by_itself(root): _xlsx(root / "stara.xlsx") files.set_status(root, "stara.xlsx", files.ARCHIVED) assert files.usable_paths(root) == [] # ── walidacja: bramka do użytku ────────────────────────────────────────── def test_upload_keeps_a_file_that_fails_validation(root): """Rzecz najważniejsza: wgranego pliku NIE TRACIMY, choćby nie przeszedł.""" files.set_rules(root, {"extensions": [".xlsx"]}) out = files.store_upload(root, "notatka.txt", "to nie jest baza".encode("utf-8"), by="dane") assert out["accepted"] is False assert (root / out["path"]).exists(), "plik odrzucony zniknął z dysku" admin_view = files.registry(root, for_admin=True)[0] assert admin_view["status"] == files.QUARANTINE assert admin_view["rejected_for"], "administrator ma widzieć powód" def test_a_held_file_is_invisible_without_admin_rights(root): """Gdyby plik wstrzymany był widoczny z powodem odrzucenia, każdy wgrywający poznałby reguły walidacji — a te są narzędziem administratora.""" files.store_upload(root, "notatka.txt", "nie baza".encode("utf-8")) assert files.registry(root, for_admin=False) == [] assert len(files.registry(root, for_admin=True)) == 1 def test_a_held_file_cannot_be_switched_into_use(root): files.store_upload(root, "notatka.txt", "nie baza".encode("utf-8")) rel = files.registry(root, for_admin=True)[0]["path"] with pytest.raises(ValueError): files.set_status(root, rel, files.ACTIVE) def test_activation_revalidates_and_holds_a_file_that_stopped_qualifying(root): """Reguły mogą się zmienić PO wgraniu — bramka sprawdza w chwili włączania, a nie tylko przy wgrywaniu.""" _xlsx(root / "mala.xlsx", rows=2) files.set_status(root, "mala.xlsx", files.ACTIVE) files.set_rules(root, {"min_rows": 500}) files.set_status(root, "mala.xlsx", files.READY) with pytest.raises(ValueError): files.set_status(root, "mala.xlsx", files.ACTIVE) assert (root / "mala.xlsx").exists() @pytest.mark.parametrize("rule,value,bad", [ ("extensions", [".xlsx"], "plik.csv"), ("max_size_mb", 0.000001, "plik.xlsx"), ]) def test_rules_reject_what_they_are_meant_to(root, rule, value, bad): files.set_rules(root, {rule: value}) out = files.store_upload(root, bad, b"x" * 2048) assert out["accepted"] is False def test_required_columns_are_checked_inside_the_workbook(root): files.set_rules(root, {"required_columns": ["id", "znaczenie"]}) _xlsx(root / "tmp.xlsx", header=("id", "opis")) why = files.validate(root / "tmp.xlsx", root) assert why and "znaczenie" in why[0] def test_duplicate_content_is_rejected_by_hash_not_by_name(root): files.set_rules(root, {"reject_duplicate_content": True}) data = _xlsx(root / "wzor.xlsx").read_bytes() first = files.store_upload(root, "pierwsza.xlsx", data) assert first["accepted"] is True second = files.store_upload(root, "inna-nazwa.xlsx", data) assert second["accepted"] is False def test_upload_never_overwrites_someone_elses_base(root): files.store_upload(root, "baza.xlsx", _xlsx(root / "w.xlsx").read_bytes()) (root / "w.xlsx").unlink() files.set_rules(root, {"reject_duplicate_content": False}) out = files.store_upload(root, "baza.xlsx", "inna treść".encode("utf-8")) assert out["name"] != "baza.xlsx" assert (root / "baza.xlsx").exists() and (root / out["path"]).exists() # ── kasowanie ──────────────────────────────────────────────────────────── def test_delete_removes_the_file_and_its_entry(root): _xlsx(root / "a.xlsx") files.set_status(root, "a.xlsx", files.ACTIVE) files.delete(root, "a.xlsx") assert not (root / "a.xlsx").exists() assert files.registry(root, for_admin=True) == [] assert files.usable_paths(root) == []