ObsiGate/tests/test_xlsx_structure.py

"""Structural mutations of an .xlsx workbook (#153 A14): service + endpoint.

Covers sheet add/rename/delete/duplicate and row/col insert/delete, the
atomicity of the batch (one locked rewrite) and the shared P0 guards
(lossy 409 gate, backup, path safety).
"""

from __future__ import annotations

import zipfile
from pathlib import Path

import pytest

openpyxl = pytest.importorskip("openpyxl")

VAULT = "TestVault"


@pytest.fixture
def book(test_vault_dir: str) -> str:
    path = Path(test_vault_dir) / "struct.xlsx"
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "Data"
    ws.append(["Nom", "Valeur"])
    ws.append(["a", 1])
    ws.append(["b", 2])
    wb.create_sheet("Vide")
    wb.save(path)
    return str(path)


def _put(client, path="struct.xlsx", actions=None, **extra):
    return client.put(
        f"/api/file/{VAULT}/xlsx/structure",
        params={"path": path},
        json={"actions": actions, **extra},
    )


def _wb(path):
    wb = openpyxl.load_workbook(path)
    try:
        return wb
    finally:
        pass


class TestSheetOps:
    def test_add_rename_delete_sheet(self, client, book):
        resp = _put(client, actions=[
            {"op": "sheet_add", "name": "Extra", "at": 0},
            {"op": "sheet_rename", "from": "Vide", "to": "Renommée"},
        ])
        assert resp.status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb.sheetnames[0] == "Extra"  # inserted at position 0
        assert "Renommée" in wb.sheetnames and "Vide" not in wb.sheetnames
        wb.close()

        resp = _put(client, actions=[{"op": "sheet_delete", "name": "Extra"}])
        assert resp.status_code == 200
        wb = openpyxl.load_workbook(book)
        assert "Extra" not in wb.sheetnames
        wb.close()

    def test_delete_last_sheet_refused(self, client, test_vault_dir):
        (Path(test_vault_dir) / "solo.xlsx").write_bytes(book_bytes("Solo"))
        resp = _put(client, path="solo.xlsx", actions=[
            {"op": "sheet_delete", "name": "Solo"},
        ])
        assert resp.status_code == 400

    def test_duplicate_copies_values(self, client, book):
        resp = _put(client, actions=[
            {"op": "sheet_duplicate", "name": "Data", "as": "Data copie"},
        ])
        assert resp.status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb["Data copie"]["A1"].value == "Nom"
        assert wb["Data copie"]["B3"].value == 2
        wb.close()


def book_bytes(sheet_name: str) -> bytes:
    import io

    wb = openpyxl.Workbook()
    wb.active.title = sheet_name
    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


class TestRowColOps:
    def test_row_insert_shifts_and_delete_removes(self, client, book):
        resp = _put(client, actions=[
            {"op": "row_insert", "sheet": "Data", "at": 2, "count": 1},
        ])
        assert resp.status_code == 200
        wb = openpyxl.load_workbook(book)
        ws = wb["Data"]
        assert ws["A2"].value is None  # the new blank row
        assert ws["A3"].value == "a"   # shifted down
        wb.close()

        resp = _put(client, actions=[
            {"op": "row_delete", "sheet": "Data", "at": 2, "count": 1},
        ])
        assert resp.status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb["Data"]["A2"].value == "a"
        wb.close()

    def test_col_insert_and_delete(self, client, book):
        assert _put(client, actions=[
            {"op": "col_insert", "sheet": "Data", "at": 2},
        ]).status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb["Data"]["B1"].value is None
        assert wb["Data"]["C1"].value == "Valeur"
        wb.close()

        assert _put(client, actions=[
            {"op": "col_delete", "sheet": "Data", "at": 2},
        ]).status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb["Data"]["B1"].value == "Valeur"
        wb.close()

    def test_insert_past_the_last_row_and_column_takes_effect(self, client, book):
        # openpyxl's sparse model silently ignores inserts past the used
        # extent: the service materialises the missing rows/columns first so
        # « below the last row » / « right of the last column » behave.
        wb = openpyxl.load_workbook(book)
        last_row, last_col = wb["Data"].max_row, wb["Data"].max_column
        wb.close()
        assert _put(client, actions=[
            {"op": "col_insert", "sheet": "Data", "at": last_col + 1},
        ]).status_code == 200
        assert _put(client, actions=[
            {"op": "row_insert", "sheet": "Data", "at": last_row + 1},
        ]).status_code == 200
        wb = openpyxl.load_workbook(book)
        ws = wb["Data"]
        assert ws.max_column == last_col + 1, "one blank column appended"
        assert ws.max_row == last_row + 1, "one blank row appended"
        wb.close()

    def test_insert_far_past_the_extent_extends_the_sheet(self, client, book):
        wb = openpyxl.load_workbook(book)
        last_row = wb["Data"].max_row
        wb.close()
        assert _put(client, actions=[
            {"op": "row_insert", "sheet": "Data", "at": last_row + 3},
        ]).status_code == 200
        wb = openpyxl.load_workbook(book)
        assert wb["Data"].max_row >= last_row + 3
        wb.close()


class TestGuards:
    def test_unknown_sheet_is_400(self, client, book):
        resp = _put(client, actions=[{"op": "row_insert", "sheet": "Nope", "at": 1}])
        assert resp.status_code == 400

    def test_unknown_op_is_400(self, client, book):
        resp = _put(client, actions=[{"op": "sheet_explode", "name": "X"}])
        assert resp.status_code == 400

    def test_bad_position_is_400(self, client, book):
        resp = _put(client, actions=[
            {"op": "row_insert", "sheet": "Data", "at": "deux"},
        ])
        assert resp.status_code == 400

    def test_empty_actions_is_400(self, client, book):
        assert _put(client, actions=[]).status_code == 400

    def test_lossy_workbook_refused_without_force(self, client, test_vault_dir):
        """Same 409 gate as the cell edits (A1)."""
        path = Path(test_vault_dir) / "lossy-struct.xlsx"
        wb = openpyxl.Workbook()
        wb.active.title = "S"
        wb["S"]["A1"] = "=A2"  # no cached value -> add one via the raw XML
        wb.save(path)
        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"))
        items[sheet] = items[sheet].decode("utf-8").replace(
            "<f>A2</f>", "<f>A2</f><v>7</v>"
        ).encode("utf-8")
        with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
            for name, blob in items.items():
                zf.writestr(name, blob)

        resp = _put(client, path="lossy-struct.xlsx", actions=[
            {"op": "sheet_add", "name": "X"},
        ])
        assert resp.status_code == 409
        assert resp.json()["code"] == "xlsx_lossy_content"

        resp = _put(client, path="lossy-struct.xlsx", actions=[
            {"op": "sheet_add", "name": "X"},
        ], force=True)
        assert resp.status_code == 200

    def test_atomicity_one_bad_action_writes_nothing(self, client, book):
        """A batch with a valid action followed by a bad one writes nothing."""
        resp = _put(client, actions=[
            {"op": "sheet_add", "name": "Temp"},
            {"op": "sheet_delete", "name": "Inexistante"},
        ])
        assert resp.status_code == 400
        wb = openpyxl.load_workbook(book)
        assert "Temp" not in wb.sheetnames
        wb.close()