bruno
31d4616baf
L'édition d'un .xlsx pouvait détruire une partie du classeur, le concurrencer en silence, ou diffuser une injection de formule. - BUG-085 : inspect_workbook() détecte ce qu'un round-trip openpyxl perd (valeurs calculées en cache, slicers, contrôles, connexions, custom XML, signature, commentaires enrichis, macros) → xlsx_lossy_features exposé en lecture, bandeau FR/EN, et 409 xlsx_lossy_content sans `force` (confirmation explicite puis reprise). Périmètre réel revalidé : graphiques, images et TCD survivent au round-trip. - BUG-086 : écriture atomique (fichier .tmp + os.replace) : un plantage ne peut plus tronquer le classeur, le backup reste intact. - BUG-087 : verrou par fichier autour du read-modify-write (timeout 15 s, 409 conflict) ; endpoint xlsx/save devenu synchrone pour que l'attente s'exécute dans le threadpool. - BUG-088 : une saisie en '=' ou '@' est stockée en texte, sauf opt-in `allow_formula` ou le bouton f(x) de la visionneuse. Le handler ServiceError expose désormais code + details, que api() propage. - BUG-084 : la suppression d'une vault purge enfin l'index inversé (documents fantômes qui continuaient de matcher) et is_stale() devient is_ready(), le nom étant trompeur (la staleness n'existe plus). Tests : 1390 pytest, 10 JSDOM (xlsx-viewer.test.mjs, branché au CI), 3 E2E Playwright, suite E2E complète verte, ruff/mypy 0. 🤖 Generated with Codebuff Co-Authored-By: Codebuff <[email protected]>
362 lines
14 KiB
Python
362 lines
14 KiB
Python
"""Display / edit / download for .xlsx files (viewer + PUT xlsx/save).
|
|
|
|
Covers #152 (affichage / édition) and #153 P0 : A1 alerte de fidélité avant
|
|
écriture, A2 écriture atomique, A3 verrou par fichier, A4 neutralisation de
|
|
l'injection de formule.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
openpyxl = pytest.importorskip("openpyxl")
|
|
|
|
VAULT = "TestVault"
|
|
|
|
|
|
@pytest.fixture
|
|
def xlsx_file(test_vault_dir: str) -> str:
|
|
from openpyxl import Workbook
|
|
|
|
path = Path(test_vault_dir) / "budget.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Budget"
|
|
ws["A1"] = "Poste"
|
|
ws["B1"] = 100
|
|
ws["A2"] = "Total"
|
|
ws["B2"] = "=B1*2"
|
|
notes = wb.create_sheet("Notes")
|
|
notes["A1"] = "hello"
|
|
wb.save(path)
|
|
return str(path)
|
|
|
|
|
|
def _add_lossy_parts(path: Path, parts: dict[str, bytes]) -> None:
|
|
"""Re-pack *path* with extra OPC parts openpyxl cannot write back."""
|
|
with zipfile.ZipFile(path) as zf:
|
|
items = {n: zf.read(n) for n in zf.namelist()}
|
|
items.update(parts)
|
|
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for name, blob in items.items():
|
|
zf.writestr(name, blob)
|
|
|
|
|
|
@pytest.fixture
|
|
def lossy_xlsx(test_vault_dir: str) -> str:
|
|
"""Workbook with a slicer + a formula carrying its cached result."""
|
|
from openpyxl import Workbook
|
|
|
|
path = Path(test_vault_dir) / "risky.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Data"
|
|
ws["A1"] = 3
|
|
ws["A2"] = "=A1*3"
|
|
wb.save(path)
|
|
# <f>…</f><v>…</v> : openpyxl keeps the formula, drops the cached result.
|
|
with zipfile.ZipFile(path) as zf:
|
|
items = {n: zf.read(n) for n in zf.namelist()}
|
|
sheet = next(n for n in items if n.startswith("xl/worksheets/sheet"))
|
|
xml = items[sheet].decode("utf-8").replace(
|
|
"<f>A1*3</f>", "<f>A1*3</f><v>9</v>"
|
|
)
|
|
items[sheet] = xml.encode("utf-8")
|
|
items["xl/slicers/slicer1.xml"] = b"<slicer/>"
|
|
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for name, blob in items.items():
|
|
zf.writestr(name, blob)
|
|
return str(path)
|
|
|
|
|
|
# ── Display ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxDisplay:
|
|
def test_renders_all_sheets(self, client, xlsx_file):
|
|
resp = client.get(f"/api/file/{VAULT}", params={"path": "budget.xlsx"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["is_xlsx"] is True
|
|
assert data["unsupported"] is False
|
|
assert [s["name"] for s in data["xlsx_sheets"]] == ["Budget", "Notes"]
|
|
|
|
first = data["xlsx_sheets"][0]["html"]
|
|
assert "Poste" in first
|
|
assert 'data-cell="B1"' in first
|
|
assert "=B1*2" in first # formula kept as text (data_only=False)
|
|
assert 'data-cell="A1"' in data["xlsx_sheets"][1]["html"]
|
|
|
|
def test_corrupt_xlsx_returns_500(self, client, test_vault_dir):
|
|
bad = Path(test_vault_dir) / "corrupt.xlsx"
|
|
bad.write_bytes(b"this is not a zip archive")
|
|
resp = client.get(f"/api/file/{VAULT}", params={"path": "corrupt.xlsx"})
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ── Index parity (tree visibility) ────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxIndexing:
|
|
def test_xlsx_in_supported_extensions(self):
|
|
from backend.indexer import SUPPORTED_EXTENSIONS
|
|
|
|
assert ".xlsx" in SUPPORTED_EXTENSIONS
|
|
|
|
def test_xlsx_indexed_metadata_only(self, test_vault_dir, xlsx_file):
|
|
from backend.indexer import _index_single_file_sync
|
|
|
|
info = _index_single_file_sync(VAULT, test_vault_dir, xlsx_file)
|
|
assert info is not None
|
|
assert info["extension"] == ".xlsx"
|
|
assert info["content"] == "" # binary: never read into TF-IDF
|
|
assert info["title"] # filename-derived title
|
|
|
|
|
|
# ── Edit ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxSave:
|
|
def _save(self, client, body, path="budget.xlsx"):
|
|
return client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": path},
|
|
json=body,
|
|
)
|
|
|
|
def test_save_updates_cell_with_number_coercion(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"B1": "250"}})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
wb = load_workbook(xlsx_file)
|
|
assert wb["Budget"]["B1"].value == 250 # int, not "250"
|
|
|
|
def test_save_leaves_other_sheets_and_formulas(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": "Titre", "B1": 42}})
|
|
assert resp.status_code == 200
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
wb = load_workbook(xlsx_file)
|
|
assert wb["Budget"]["B2"].value == "=B1*2"
|
|
assert wb["Notes"]["A1"].value == "hello"
|
|
|
|
def test_save_empty_string_clears_cell(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": ""}})
|
|
assert resp.status_code == 200
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
assert load_workbook(xlsx_file)["Budget"]["A1"].value is None
|
|
|
|
def test_save_creates_backup(self, client, xlsx_file):
|
|
from backend.services.backups import get_backup_dir
|
|
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": "backup-me"}})
|
|
assert resp.status_code == 200
|
|
backup_dir = Path(get_backup_dir(VAULT, "budget.xlsx"))
|
|
assert backup_dir.is_dir()
|
|
assert list(backup_dir.glob("*.bak"))
|
|
|
|
def test_unknown_sheet_400(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Nope", "cells": {"A1": "x"}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_invalid_cell_ref_400(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A-1": "x"}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_wrong_extension_400(self, client, test_vault_dir):
|
|
(Path(test_vault_dir) / "note.md").write_text("# hi\n", encoding="utf-8")
|
|
resp = self._save(client, {"sheet": "Sheet", "cells": {"A1": "x"}}, path="note.md")
|
|
assert resp.status_code == 400
|
|
|
|
def test_missing_file_404(self, client):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": "x"}}, path="absent.xlsx")
|
|
assert resp.status_code == 404
|
|
|
|
def test_too_many_cells_400(self, client, xlsx_file):
|
|
cells = {f"A{i}": i for i in range(1, 502)}
|
|
resp = self._save(client, {"sheet": "Budget", "cells": cells})
|
|
assert resp.status_code == 400
|
|
|
|
def test_nested_value_400(self, client, xlsx_file):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": {"nested": 1}}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_missing_sheet_field_400(self, client, xlsx_file):
|
|
resp = self._save(client, {"cells": {"A1": "x"}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_non_boolean_flag_400(self, client, xlsx_file):
|
|
for flag in ("force", "allow_formula"):
|
|
resp = self._save(client, {"sheet": "Budget", "cells": {"A1": "x"}, flag: "yes"})
|
|
assert resp.status_code == 400, flag
|
|
|
|
|
|
# ── #153 A1 — lossy-write guard ──────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxLossyGuard:
|
|
def _save(self, client, body, path):
|
|
return client.put(
|
|
f"/api/file/{VAULT}/xlsx/save", params={"path": path}, json=body,
|
|
)
|
|
|
|
def test_read_reports_lossy_features(self, client, lossy_xlsx):
|
|
data = client.get(f"/api/file/{VAULT}", params={"path": "risky.xlsx"}).json()
|
|
assert data["is_xlsx"] is True
|
|
assert "slicers" in data["xlsx_lossy_features"]
|
|
assert "cached_values" in data["xlsx_lossy_features"]
|
|
|
|
def test_read_reports_nothing_for_a_plain_workbook(self, client, xlsx_file):
|
|
data = client.get(f"/api/file/{VAULT}", params={"path": "budget.xlsx"}).json()
|
|
# B2 holds "=B1*2" but openpyxl wrote no cached <v> for it.
|
|
assert data["xlsx_lossy_features"] == []
|
|
|
|
def test_save_refuses_without_force(self, client, lossy_xlsx):
|
|
resp = self._save(client, {"sheet": "Data", "cells": {"B1": "hello"}}, "risky.xlsx")
|
|
assert resp.status_code == 409
|
|
body = resp.json()
|
|
assert body["code"] == "xlsx_lossy_content"
|
|
assert "slicers" in body["details"]["features"]
|
|
|
|
def test_refused_save_leaves_the_file_untouched(self, client, lossy_xlsx):
|
|
before = Path(lossy_xlsx).read_bytes()
|
|
self._save(client, {"sheet": "Data", "cells": {"B1": "hello"}}, "risky.xlsx")
|
|
assert Path(lossy_xlsx).read_bytes() == before
|
|
|
|
def test_save_with_force_succeeds(self, client, lossy_xlsx):
|
|
resp = self._save(
|
|
client,
|
|
{"sheet": "Data", "cells": {"B1": "hello"}, "force": True},
|
|
"risky.xlsx",
|
|
)
|
|
assert resp.status_code == 200
|
|
assert openpyxl.load_workbook(lossy_xlsx)["Data"]["B1"].value == "hello"
|
|
|
|
def test_inspect_flags_every_known_family(self, lossy_xlsx):
|
|
from backend.xlsx_reader import LOSSY_PARTS, inspect_workbook
|
|
|
|
for key, prefixes in LOSSY_PARTS.items():
|
|
_add_lossy_parts(
|
|
Path(lossy_xlsx),
|
|
{f"{prefixes[0]}probe.xml": b"<x/>" for _ in [0]},
|
|
)
|
|
assert key in inspect_workbook(Path(lossy_xlsx)), key
|
|
|
|
def test_inspect_is_quiet_on_a_corrupt_archive(self, test_vault_dir):
|
|
from backend.xlsx_reader import inspect_workbook
|
|
|
|
bad = Path(test_vault_dir) / "broken.xlsx"
|
|
bad.write_bytes(b"not a zip at all")
|
|
assert inspect_workbook(bad) == []
|
|
|
|
|
|
# ── #153 A2 — atomic write ───────────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxAtomicWrite:
|
|
def test_failed_save_keeps_the_original(self, client, xlsx_file, monkeypatch):
|
|
from openpyxl.workbook.workbook import Workbook
|
|
|
|
before = Path(xlsx_file).read_bytes()
|
|
|
|
def boom(self, *args, **kwargs):
|
|
raise OSError("disk full")
|
|
|
|
monkeypatch.setattr(Workbook, "save", boom)
|
|
# TestClient re-raises the server exception (in production: 500).
|
|
with pytest.raises(OSError):
|
|
client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": "budget.xlsx"},
|
|
json={"sheet": "Budget", "cells": {"A1": "perdu"}},
|
|
)
|
|
# The workbook on disk is byte-identical : the write never reached it.
|
|
assert Path(xlsx_file).read_bytes() == before
|
|
# No temporary file left behind in the vault.
|
|
assert list(Path(xlsx_file).parent.glob("*.tmp")) == []
|
|
|
|
def test_no_tmp_left_after_a_successful_save(self, client, xlsx_file):
|
|
client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": "budget.xlsx"},
|
|
json={"sheet": "Budget", "cells": {"A1": "ok"}},
|
|
)
|
|
assert list(Path(xlsx_file).parent.glob("*.tmp")) == []
|
|
|
|
|
|
# ── #153 A3 — per-file write lock ────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxWriteLock:
|
|
def test_concurrent_write_returns_409(self, client, xlsx_file):
|
|
from backend.services import mutations
|
|
|
|
key = str(Path(xlsx_file).resolve())
|
|
with mutations._xlsx_write_lock(key):
|
|
resp = client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": "budget.xlsx"},
|
|
json={"sheet": "Budget", "cells": {"A1": "concurrent"}},
|
|
)
|
|
assert resp.status_code == 409
|
|
assert resp.json()["code"] == "conflict"
|
|
# The blocked call wrote nothing.
|
|
assert openpyxl.load_workbook(xlsx_file)["Budget"]["A1"].value == "Poste"
|
|
|
|
def test_lock_is_released_after_a_normal_save(self, client, xlsx_file):
|
|
from backend.services import mutations
|
|
|
|
key = str(Path(xlsx_file).resolve())
|
|
client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": "budget.xlsx"},
|
|
json={"sheet": "Budget", "cells": {"A1": "premier"}},
|
|
)
|
|
# The lock must be free again once the request returned.
|
|
with mutations._xlsx_write_lock(key):
|
|
pass
|
|
assert openpyxl.load_workbook(xlsx_file)["Budget"]["A1"].value == "premier"
|
|
|
|
|
|
# ── #153 A4 — formula injection ──────────────────────────────────────────
|
|
|
|
|
|
class TestXlsxFormulaGuard:
|
|
def _save(self, client, cells, **extra):
|
|
return client.put(
|
|
f"/api/file/{VAULT}/xlsx/save",
|
|
params={"path": "budget.xlsx"},
|
|
json={"sheet": "Budget", "cells": cells, **extra},
|
|
)
|
|
|
|
def test_formula_like_value_is_stored_as_text(self, client, xlsx_file):
|
|
resp = self._save(client, {"A3": "=cmd|'/c calc'!A1"})
|
|
assert resp.status_code == 200
|
|
cell = openpyxl.load_workbook(xlsx_file)["Budget"]["A3"]
|
|
assert cell.value == "=cmd|'/c calc'!A1"
|
|
assert cell.data_type == "s" # pas de <f> dans l'archive
|
|
|
|
def test_at_prefix_is_stored_as_text(self, client, xlsx_file):
|
|
self._save(client, {"A4": "@SUM(A1:A2)"})
|
|
assert openpyxl.load_workbook(xlsx_file)["Budget"]["A4"].data_type == "s"
|
|
|
|
def test_allow_formula_keeps_a_real_formula(self, client, xlsx_file):
|
|
resp = self._save(client, {"A3": "=B1+5"}, allow_formula=True)
|
|
assert resp.status_code == 200
|
|
assert openpyxl.load_workbook(xlsx_file)["Budget"]["A3"].data_type == "f"
|
|
|
|
def test_numbers_are_unaffected(self, client, xlsx_file):
|
|
self._save(client, {"B3": "42", "B4": "-3.5"})
|
|
ws = openpyxl.load_workbook(xlsx_file)["Budget"]
|
|
assert ws["B3"].value == 42 and isinstance(ws["B3"].value, int)
|
|
assert ws["B4"].value == -3.5
|