""" 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())