ecw-charge-validation/tests/test_spreadsheet_write.py

"""
write_xlsx(): the write side of core/spreadsheet.py.

    python tests\\test_spreadsheet_write.py

Round-trips through the existing `read_rows()` reader rather than asserting
on raw XML — what matters is that a file this writes is a file this project's
own reader (and real Excel) can open, not the exact bytes.

Also write_xlsx_keep_old(), the one writer both dated outputs go through
(review item D10): a fresh path, an existing file kept aside, and an existing
file that is OPEN — held here by a plain `open(path, "rb")`, which on Windows
refuses the rename with WinError 32 exactly as Excel does — whose run must
land under a `__HHMMSS` name instead of being lost. Every case works in its
own temp folder.
"""

import io
import re
import sys
import pathlib
import tempfile
import contextlib
from datetime import datetime

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "src"))

from ecw_bot.core import spreadsheet                 # noqa: E402

TMP = pathlib.Path(tempfile.mkdtemp())


def case_write_then_read_round_trips():
    rows = [
        ["CLAIMS#", "PAYER", "Group Name"],
        ["9107000", "Medicaid", "Medicaid"],
        ["9106000", "DentaQuest", "Dentaquest"],
    ]
    path = TMP / "roundtrip.xlsx"
    spreadsheet.write_xlsx(path, rows)
    back = spreadsheet.read_rows(path)
    assert back == rows, back


def case_special_characters_are_escaped():
    rows = [["PATIENT", "NOTE"],
            ["Smith & Sons", "A < B > C"],
            ["O'Brien", '"quoted"']]
    path = TMP / "escaped.xlsx"
    spreadsheet.write_xlsx(path, rows)
    back = spreadsheet.read_rows(path)
    assert back == rows, back


def case_wide_row_uses_correct_column_letters():
    """27 columns exercises the single-letter -> AA rollover."""
    header = [f"col{i}" for i in range(27)]
    data = [str(i) for i in range(27)]
    path = TMP / "wide.xlsx"
    spreadsheet.write_xlsx(path, [header, data])
    back = spreadsheet.read_rows(path)
    assert back[1][26] == "26", back[1]        # column AA (index 26)
    assert back[1] == data, back[1]


def case_control_characters_are_stripped_not_written():
    rows = [["A"], ["value\x00with\x01control\x1fchars"]]
    path = TMP / "control_chars.xlsx"
    spreadsheet.write_xlsx(path, rows)
    back = spreadsheet.read_rows(path)
    assert back == [["A"], ["valuewithcontrolchars"]], back


def case_creates_parent_directory():
    path = TMP / "nested" / "new_dir" / "file.xlsx"
    assert not path.parent.exists()
    spreadsheet.write_xlsx(path, [["A"], ["1"]])
    assert path.exists()


def case_non_string_values_are_stringified():
    rows = [["N"], [5], [3.5]]
    path = TMP / "numbers.xlsx"
    spreadsheet.write_xlsx(path, rows)
    back = spreadsheet.read_rows(path)
    assert back == [["N"], ["5"], ["3.5"]], back


# ── write_xlsx_keep_old ──────────────────────────────────────────────────────

OLD = [["CLAIMS#", "Group Name"], ["9107000", "hand-filled by the operator"]]
NEW = [["CLAIMS#", "Group Name"], ["9107000", "Medicaid"], ["9106000", ""]]


def _folder_with_old_file(name="Medcor Ready_10-05-2026.xlsx"):
    """A fresh temp folder holding a same-day file with OLD rows in it."""
    folder = pathlib.Path(tempfile.mkdtemp(dir=TMP))
    out = folder / name
    spreadsheet.write_xlsx(out, OLD)
    return folder, out


def _keep_old(out, rows=NEW):
    """Run the helper, returning (path it returned, what it printed)."""
    printed = io.StringIO()
    with contextlib.redirect_stdout(printed):
        got = spreadsheet.write_xlsx_keep_old(out, rows, "[TAG]")
    return got, printed.getvalue()


@contextlib.contextmanager
def _held_open(path):
    """
    Hold `path` open the way Excel does for the duration. A Python handle
    blocks the RENAME (WinError 32) but not a write over it — so the cases
    below also prove the old file was never written over. The rename is
    probed first: a platform that does not lock an open file cannot
    exercise this path, and should fail loudly rather than pass vacuously.
    """
    with open(path, "rb") as handle:
        probe = path.with_name("probe-rename.xlsx")
        try:
            path.rename(probe)
        except OSError:
            pass
        else:
            probe.rename(path)
            raise AssertionError("an open file is not locked against rename "
                                 "here — cannot exercise the locked path")
        yield handle


class _FixedClock:
    """Stands in for spreadsheet.datetime: now() is always 12:34:56."""
    @staticmethod
    def now():
        return datetime(2026, 10, 5, 12, 34, 56)


def case_keep_old_fresh_path_is_written_as_is():
    folder = pathlib.Path(tempfile.mkdtemp(dir=TMP))
    out = folder / "Charge Summary_10-05-2026.xlsx"
    got, printed = _keep_old(out)
    assert got == out, got
    assert spreadsheet.read_rows(out) == NEW
    assert [p.name for p in folder.iterdir()] == [out.name], \
        list(folder.iterdir())
    assert printed == "", printed


def case_keep_old_existing_file_is_kept_aside_and_new_rows_take_its_name():
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    got, printed = _keep_old(out)
    assert got == out, got
    assert spreadsheet.read_rows(out) == NEW
    kept = list(folder.glob("Medcor Ready_10-05-2026__*.xlsx"))
    assert len(kept) == 1, kept
    assert re.fullmatch(r"Medcor Ready_10-05-2026__\d{6}\.xlsx",
                        kept[0].name), kept[0].name
    assert kept[0].read_bytes() == old_bytes, "the kept file was changed"
    assert printed.strip() == (f"[TAG] {out.name} already existed — kept as "
                               f"{kept[0].name}"), printed


def case_keep_old_open_file_is_left_alone_and_new_rows_get_a_free_name():
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    with _held_open(out):
        got, printed = _keep_old(out)     # must not raise
        assert out.exists(), "the open file was moved"
    assert got != out, got
    assert got.parent == folder, got
    assert re.fullmatch(r"Medcor Ready_10-05-2026__\d{6}\.xlsx",
                        got.name), got.name
    assert spreadsheet.read_rows(got) == NEW, "this run's rows were lost"
    assert out.read_bytes() == old_bytes, "the open file was changed"
    assert sorted(p.name for p in folder.iterdir()) == \
        sorted([out.name, got.name]), list(folder.iterdir())
    assert printed.startswith(f"[TAG] {out.name} is open or locked ("), \
        printed
    assert "WinError 32" in printed, printed
    assert printed.rstrip().endswith(f"— saved this run as {got.name} "
                                     f"instead"), printed


def case_keep_old_names_skip_ones_already_taken_in_the_same_second():
    folder, out = _folder_with_old_file()
    taken = folder / "Medcor Ready_10-05-2026__123456.xlsx"
    taken.write_bytes(b"an earlier run, same second")
    orig = spreadsheet.datetime
    spreadsheet.datetime = _FixedClock
    try:
        got, printed = _keep_old(out)
        assert got == out, got
        assert "kept as Medcor Ready_10-05-2026__123456_2.xlsx" in printed, \
            printed
        with _held_open(out):
            got, printed = _keep_old(out)
        assert got.name == "Medcor Ready_10-05-2026__123456_3.xlsx", got
    finally:
        spreadsheet.datetime = orig
    assert taken.read_bytes() == b"an earlier run, same second"


def case_keep_old_failed_write_after_the_rename_puts_the_old_file_back():
    """
    The rename succeeds but writing the new file in its place fails (disk
    full, a sync client grabbing the name): the old file goes back under its
    own name — the failed write's half-file removed first — and the rows go
    to a free `__HHMMSS` name instead.
    """
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    real_write = spreadsheet.write_xlsx

    def _fails_on_the_usual_name(path, rows):
        if pathlib.Path(path) == out:
            pathlib.Path(path).write_bytes(b"PK half a zip")
            raise OSError(28, "No space left on device")
        real_write(path, rows)

    spreadsheet.write_xlsx = _fails_on_the_usual_name
    try:
        got, printed = _keep_old(out)
    finally:
        spreadsheet.write_xlsx = real_write
    assert out.read_bytes() == old_bytes, "the old file was not put back"
    assert got != out and spreadsheet.read_rows(got) == NEW, got
    assert sorted(p.name for p in folder.iterdir()) == \
        sorted([out.name, got.name]), list(folder.iterdir())
    assert "No space left on device" in printed, printed


def case_keep_old_failed_move_back_says_the_usual_name_may_be_incomplete():
    """
    Fix round 1 (2026-10-05): the write to the usual name fails AND its
    half-file cannot be removed (a virus scanner holding it — here a real
    open handle, so the unlink is refused with WinError 32). The old file
    cannot go back, so the log must say the usual name may be incomplete
    and where the operator's earlier sheet really is.
    """
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    real_write = spreadsheet.write_xlsx
    handles = []

    def _fails_and_stays_held(path, rows):
        if pathlib.Path(path) == out:
            pathlib.Path(path).write_bytes(b"PK half a zip")
            handles.append(open(path, "rb"))   # the scanner's grip
            raise OSError(28, "No space left on device")
        real_write(path, rows)

    spreadsheet.write_xlsx = _fails_and_stays_held
    try:
        got, printed = _keep_old(out)
    finally:
        spreadsheet.write_xlsx = real_write
        for h in handles:
            h.close()
    backups = [p for p in folder.iterdir()
               if p not in (out, got)]
    assert len(backups) == 1, list(folder.iterdir())
    backup = backups[0]
    assert backup.read_bytes() == old_bytes, "the earlier sheet changed"
    assert got != out and spreadsheet.read_rows(got) == NEW, got
    want = (f"[TAG] {out.name} may be an incomplete file — your earlier "
            f"sheet is {backup.name} (could not move it back: ")
    assert want in printed, printed
    assert "already existed — kept as" not in printed, printed


def case_keep_old_a_bug_while_writing_puts_the_old_file_back_and_raises():
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    real_write = spreadsheet.write_xlsx

    def _bug(path, rows):
        raise ValueError("not a file-system problem")

    spreadsheet.write_xlsx = _bug
    try:
        _keep_old(out)
    except ValueError:
        pass
    else:
        raise AssertionError("a non-OSError must propagate, not be retried")
    finally:
        spreadsheet.write_xlsx = real_write
    assert out.read_bytes() == old_bytes
    assert [p.name for p in folder.iterdir()] == [out.name], \
        list(folder.iterdir())


def case_keep_old_fallback_write_failing_too_propagates():
    """Nowhere left to put the rows: the caller's own except reports it."""
    folder, out = _folder_with_old_file()
    old_bytes = out.read_bytes()
    real_write = spreadsheet.write_xlsx

    def _disk_full(path, rows):
        raise OSError(28, "No space left on device")

    spreadsheet.write_xlsx = _disk_full
    try:
        with _held_open(out):
            _keep_old(out)
    except OSError as exc:
        assert "No space left" in str(exc), exc
    else:
        raise AssertionError("the fallback's own failure must propagate")
    finally:
        spreadsheet.write_xlsx = real_write
    assert out.read_bytes() == old_bytes


CASES = [v for k, v in sorted(globals().items()) if k.startswith("case_")]


def main() -> int:
    failures = 0
    for fn in CASES:
        name = fn.__name__[5:].replace("_", " ")
        try:
            fn()
            print(f"  ok    {name}")
        except AssertionError as exc:
            print(f"  FAIL  {name}: {exc}")
            failures += 1
        except Exception as exc:
            print(f"  ERROR {name}: {type(exc).__name__}: {exc}")
            failures += 1
    print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())