ecw-charge-validation/tests/test_input_archive.py

r"""
Input\ holds exactly one report, and it is always the current one.

    python tests\test_input_archive.py

The problem this covers, in the words it was reported in: run the bot a second
time, and the agent takes the **old** file's claims instead of the new one.

Three days of testing left six reports in `Input\`, named
`Claim Report_09-07-2026.xlsx`, `... (2).xlsx`, `... (3).xlsx` and so on. Once
more than one is sitting there, "which report is current?" stops being obvious
— and the wrong answer is silent: the run looks clean and processes claims from
a stale queue.

Two things make the answer unambiguous, and both are tested here:

  * the export moves every earlier report into `Input\Archive\` before saving
    the new one, so there is only ever one to choose from;
  * `latest_report()` picks by **modification time**, never by name. Sorting
    names is a trap: `Claim Report_09-09-2026 (2).xlsx` sorts *before*
    `Claim Report_09-09-2026.xlsx`, because a space (0x20) is lower than a dot
    (0x2e) — so the alphabetically first file can be the older one.

Runs against a temporary Input\ ; the real one is never touched.
"""

import sys
import time
import pathlib
import tempfile

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

from ecw_bot import settings as cfg, claim_file        # noqa: E402
from ecw_bot.pages import claim_report                 # noqa: E402

CHECKS = []


def check(fn):
    CHECKS.append(fn)
    return fn


def sandbox():
    """A fresh temporary Input\\ , pointed at by settings."""
    root = pathlib.Path(tempfile.mkdtemp())
    cfg.INPUTS = root / "Input"
    cfg.INPUTS_ARCHIVE = cfg.INPUTS / "Archive"
    cfg.INPUTS.mkdir(parents=True)
    return cfg.INPUTS


def report(where, name, minutes_ago=0, ids=("9115000",)):
    """Write a minimal xlsx that claim_file can actually read."""
    import zipfile
    ns = 'xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"'
    rows = ['<row r="1"><c r="A1" t="str"><v></v></c>'
            '<c r="C1" t="str"><v>CLAIMS#</v></c></row>']
    for i, cid in enumerate(ids, start=2):
        rows.append(f'<row r="{i}"><c r="A{i}" t="str"><v></v></c>'
                    f'<c r="C{i}" t="str"><v>{cid}</v></c></row>')
    sheet = (f'<?xml version="1.0"?><worksheet {ns}><sheetData>'
             f'{"".join(rows)}</sheetData></worksheet>')
    path = where / name
    with zipfile.ZipFile(path, "w") as z:
        z.writestr("xl/worksheets/sheet1.xml", sheet)
    if minutes_ago:
        old = time.time() - minutes_ago * 60
        import os
        os.utime(path, (old, old))
    return path


@check
def archive_leaves_only_the_new_report(page=None):
    inp = sandbox()
    report(inp, "Claim Report_09-07-2026.xlsx", minutes_ago=2880)
    report(inp, "Claim Report_09-07-2026 (2).xlsx", minutes_ago=2900)
    report(inp, "Claim Report_09-08-2026.xlsx", minutes_ago=1440)
    report(inp, "rescued_20260908_163651.xlsx", minutes_ago=1400)

    moved = claim_report._archive_old()
    assert len(moved) == 4, [m.name for m in moved]

    left = [f.name for f in inp.iterdir() if f.is_file()]
    assert left == [], f"Input should be empty of reports, got {left}"
    archived = sorted(f.name for f in cfg.INPUTS_ARCHIVE.iterdir())
    assert len(archived) == 4, archived
    # Each archived name carries the file's own mtime, so nothing collides and
    # the order is readable.
    assert all("__20" in n for n in archived), archived


@check
def the_new_report_is_the_one_that_gets_read(page=None):
    """The reported symptom, reproduced and then not happening."""
    inp = sandbox()
    report(inp, "Claim Report_09-08-2026.xlsx",
           minutes_ago=1440, ids=("1111111", "2222222"))

    # Second run: archive, then the new file lands.
    claim_report._archive_old()
    report(inp, "Claim Report_09-09-2026.xlsx", ids=("9115000", "9114000"))

    info = claim_file.claim_ids()          # no path - it has to find it
    assert info["file"].name == "Claim Report_09-09-2026.xlsx", info["file"]
    assert info["ids"] == ["9115000", "9114000"], info["ids"]
    assert "1111111" not in info["ids"], "read the OLD report's claims"


@check
def newest_by_mtime_not_by_name(page=None):
    """
    The name-sorting trap, with the two files that expose it.

    `Claim Report_09-09-2026 (2).xlsx` sorts FIRST alphabetically - a space is
    0x20, a dot is 0x2e - while being the OLDER file. Anything picking
    `sorted(...)[0]` reads stale claims and looks perfectly healthy doing it.
    """
    inp = sandbox()
    old = report(inp, "Claim Report_09-09-2026 (2).xlsx",
                 minutes_ago=120, ids=("1111111",))
    new = report(inp, "Claim Report_09-09-2026.xlsx", ids=("9115000",))

    names = sorted(f.name for f in inp.iterdir())
    assert names[0] == old.name, (
        f"this test assumes '(2)' sorts first; it did not: {names}")

    picked = claim_file.latest_report()
    assert picked.name == new.name, (
        f"picked {picked.name}, the alphabetically-first but OLDER file")


@check
def more_than_one_report_is_reported_not_guessed(page=None):
    import io
    import contextlib
    inp = sandbox()
    report(inp, "Claim Report_09-08-2026.xlsx", minutes_ago=1440)
    report(inp, "Claim Report_09-09-2026.xlsx")

    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        picked = claim_file.latest_report()
    out = buf.getvalue()
    assert picked.name == "Claim Report_09-09-2026.xlsx", picked
    # A guess that goes unmentioned is how a stale read stays invisible.
    assert "WARNING" in out and "should be one" in out, out


@check
def empty_input_says_where_the_old_ones_went(page=None):
    inp = sandbox()
    (inp / "Archive").mkdir()
    try:
        claim_file.latest_report()
    except FileNotFoundError as exc:
        assert "Archive" in str(exc), exc
        return
    raise AssertionError("expected FileNotFoundError naming Archive")


@check
def archiving_can_be_turned_off(page=None):
    inp = sandbox()
    report(inp, "Claim Report_09-08-2026.xlsx", minutes_ago=1440)
    was = cfg.ARCHIVE_OLD_INPUTS
    cfg.ARCHIVE_OLD_INPUTS = False
    try:
        assert claim_report._archive_old() == []
        assert (inp / "Claim Report_09-08-2026.xlsx").exists()
    finally:
        cfg.ARCHIVE_OLD_INPUTS = was


def main() -> int:
    real_in, real_arch = cfg.INPUTS, cfg.INPUTS_ARCHIVE
    failures = 0
    try:
        for fn in CHECKS:
            name = fn.__name__.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
    finally:
        cfg.INPUTS, cfg.INPUTS_ARCHIVE = real_in, real_arch

    print(f"\n{len(CHECKS) - failures}/{len(CHECKS)} passed")
    return 1 if failures else 0


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