"""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_a_file_dropped_on_the_share_is_adopted_as_active(root): """Baza położona na udziale poza aplikacją (np. przez NFS) ma działać — tak było przed DAN-27 i tak ma zostać. Plik WGRANY EKRANEM to inna sprawa: ten wymaga świadomego włączenia (patrz test niżej).""" _xlsx(root / "baza.xlsx") entry = files.registry(root)[0] assert entry["status"] == files.ACTIVE assert entry["in_use"] is True def test_only_active_files_reach_the_search(root): _xlsx(root / "a.xlsx") _xlsx(root / "b.xlsx") files.registry(root) # przyjęcie zastanych files.set_status(root, "b.xlsx", files.READY) # świadome odstawienie assert [pathlib.Path(p).name for p in files.usable_paths(root)] == ["a.xlsx"] files.set_status(root, "b.xlsx", files.ACTIVE) assert len(files.usable_paths(root)) == 2 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) == [] # ── przejście na rejestr nie może wyłączyć wyszukiwania ───────────────── def test_bases_already_on_the_share_stay_in_use_after_the_switch(root): """Dotąd bazy działały domyślnie (wyłączało się je przez DISABLED_BASES). Po przejściu na rejestr pusty stan oznaczałby, że program nagle niczego nie znajduje — cicha zmiana zachowania gorsza od awarii, bo wygląda jak pusta baza.""" _xlsx(root / "main_base.xlsx") _xlsx(root / "zodiac_pl.xlsx") assert len(files.usable_paths(root)) == 2, "zastane bazy wypadły z wyszukiwania" assert all(e["in_use"] for e in files.registry(root)) def test_adoption_happens_once_and_respects_later_decisions(root): """Po przyjęciu stan jest zapisany, więc świadome odstawienie bazy ZOSTAJE — kolejny odczyt nie może jej wskrzesić.""" _xlsx(root / "a.xlsx") _xlsx(root / "b.xlsx") files.registry(root) # przyjęcie files.set_status(root, "a.xlsx", files.READY) # świadome odstawienie assert [pathlib.Path(p).name for p in files.usable_paths(root)] == ["b.xlsx"] files.set_status(root, "b.xlsx", files.READY) # odstawiamy wszystko assert files.usable_paths(root) == [], "pusty wybór został wskrzeszony" def test_uploaded_files_still_need_an_explicit_switch_on(root): """Przyjęcie dotyczy TYLKO baz zastanych. Plik wgrany ekranem ktoś musi świadomie włączyć — inaczej nowa baza wchodziłaby do wyników sama.""" _xlsx(root / "zastana.xlsx") files.registry(root) out = files.store_upload(root, "nowa.xlsx", _xlsx(root / "tmp.xlsx").read_bytes()) assert out["accepted"] is True names = [pathlib.Path(p).name for p in files.usable_paths(root)] assert "nowa.xlsx" not in names, "wgrana baza weszła do wyników bez decyzji" def test_adoption_survives_a_read_only_share(root, monkeypatch): """Na udziale tylko do odczytu stanu nie da się zapisać — zachowanie ma zostać to samo, tylko przyjęcie powtórzy się przy każdym uruchomieniu.""" _xlsx(root / "a.xlsx") def boom(*a, **kw): raise OSError("read-only file system") monkeypatch.setattr(files, "_write_state", boom) assert len(files.usable_paths(root)) == 1