ecw-charge-validation/tests/test_runs_from_db.py

r"""
web/runs.py from the database (operator, 2026-10-30): the Portal on Dokploy
cannot read the bot PC's Logs folder, so its Runs page said "No runs match"
for every date. With `DbEvents` the same summaries come from
`dbo.CV_RunEvent` — the same JSON lines the files hold.

    python tests\test_runs_from_db.py

No database: the repo is a stand-in.
"""

import sys
import json
import pathlib

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

from web import runs                                  # noqa: E402

CASES = []


def case(fn):
    CASES.append(fn)
    return fn


def _line(day, t, kind, **f):
    return json.dumps({**f, "t": f"2026-10-{day}T{t}.000", "kind": kind})


def _run(day, hhmmss, end_code=0):
    t = f"{hhmmss[:2]}:{hhmmss[2:4]}:{hhmmss[4:]}"
    run_id = f"202610{day}_{hhmmss}"
    lines = [_line(day, t, "run_start", source="terminal", started_by="ops",
                   mode="VERIFY", claim_limit=0, test_sheet="", pid=1,
                   log="x"),
             _line(day, t, "claim_start", n=1, total=1, claim="9103000"),
             _line(day, t, "run_end", exit_code=end_code)]
    return run_id, lines


class FakeEventsRepo:
    def __init__(self, *runs_):
        self.rows = [(rid, line) for rid, lines in runs_ for line in lines]
        self.calls = []

    def run_ids(self):
        self.calls.append(("run_ids",))
        return sorted({r for r, _ in self.rows}, reverse=True)

    def run_event_lines(self, run_ids):
        self.calls.append(("run_event_lines", tuple(run_ids)))
        return [(r, b) for r, b in self.rows if r in set(run_ids)]


@case
def runs_come_from_the_database_newest_first_and_filtered_by_day():
    repo = FakeEventsRepo(_run("28", "161611"), _run("29", "163000"),
                          _run("29", "170620", end_code=9))
    src = runs.DbEvents(repo)
    got = runs.list_runs(src, day_from="2026-10-29", day_to="2026-10-29")
    assert [r["run_id"] for r in got] == ["20261029_170620",
                                         "20261029_163000"], got
    assert got[0]["status"] == "Run Error" and got[1]["status"] == "Done"
    assert repo.calls[-1] == ("run_event_lines",
                              ("20261029_170620", "20261029_163000")), \
        "one query for the day's runs, not one per run"


@case
def one_run_and_the_latest_run_come_from_the_database():
    repo = FakeEventsRepo(_run("28", "161611"), _run("29", "163000"))
    src = runs.DbEvents(repo)
    assert runs.load_run(src, "20261028_161611")["status"] == "Done"
    assert runs.load_run(src, "20261030_000000") is None
    assert runs.latest_run(src)["run_id"] == "20261029_163000"


@case
def a_screenshot_is_never_served_from_the_database():
    repo = FakeEventsRepo(("20261029_163000", [
        _line("29", "16:30:00", "screenshot", path="C:/x/101500_shot.png")]))
    assert runs.screenshot_path(runs.DbEvents(repo), "20261029_163000", 0,
                                "C:/x") is None


@case
def the_portal_runs_page_reads_its_events_source():
    import test_web_app as t
    repo = FakeEventsRepo(_run("29", "163000"))
    client, _ = t.signed_in(t.FakeRepo(), t.FakeRunner(),
                            events=runs.DbEvents(repo))
    html = client.get("/runs", params={"day_from": "2026-10-29",
                                       "day_to": "2026-10-29"}).text
    assert "No runs match" not in html, html[-800:]
    assert "/runs/20261029_163000" in html


def _live_lines(day, hhmm, ss_last, end=False):
    """A run whose last event is at hh:mm:ss_last; ended or not."""
    run_id = f"202610{day}_{hhmm}00"
    t0 = f"{hhmm[:2]}:{hhmm[2:]}:00"
    lines = [_line(day, t0, "run_start", source="terminal", started_by="ops",
                   mode="VERIFY", claim_limit=0, test_sheet="", pid=1,
                   log="x"),
             _line(day, t0, "queue", claims=["9103000", "9104000"]),
             _line(day, t0, "claim_start", n=1, total=2, claim="9103000"),
             _line(day, f"{hhmm[:2]}:{hhmm[2:]}:{ss_last}", "claim_result",
                   n=1, claim="9103000", category="Fail")]
    if end:
        lines.append(_line(day, f"{hhmm[:2]}:{hhmm[2:]}:{ss_last}",
                           "run_end", exit_code=0))
    return run_id, lines


@case
def the_server_dashboard_shows_a_run_still_writing_events_as_running():
    from datetime import datetime
    repo = FakeEventsRepo(_live_lines("30", "1015", "40"))
    st = runs.live_status(runs.DbEvents(repo),
                          now=datetime(2026, 10, 30, 10, 17, 0))
    assert st["running"] is True and st["progress"] == [1, 2], st
    assert st["counts"] == {"Pass": 0, "Fail": 1, "Review": 0}, st
    assert st["source"] == "terminal" and st["tail"] == [], st


@case
def the_server_dashboard_shows_an_ended_or_silent_run_as_ended():
    from datetime import datetime
    repo = FakeEventsRepo(_live_lines("30", "1015", "40", end=True))
    st = runs.live_status(runs.DbEvents(repo),
                          now=datetime(2026, 10, 30, 10, 17, 0))
    assert st["running"] is False and st["exit_code"] == 0, st
    assert "Done" in st["message"] or "exit 0" in st["message"].lower() \
        or st["message"], st
    silent = FakeEventsRepo(_live_lines("30", "1015", "40"))
    st = runs.live_status(runs.DbEvents(silent),
                          now=datetime(2026, 10, 30, 11, 30, 0))
    assert st["running"] is False, "no event for over 10 minutes"
    assert runs.live_status(runs.DbEvents(FakeEventsRepo()))["running"] \
        is False


@case
def the_server_portals_bot_status_comes_from_the_database():
    import test_web_app as t
    repo = FakeEventsRepo(_live_lines("30", "1015", "40", end=True))
    client, _ = t.signed_in(t.FakeRepo(), t.FakeRunner(),
                            events=runs.DbEvents(repo))
    st = client.get("/api/bot/status").json()
    assert st["running"] is False and st["exit_code"] == 0, st
    assert st["counts"]["Fail"] == 1 and st["progress"] == [1, 2], st


def main() -> int:
    failures = 0
    for fn in CASES:
        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
    print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
    return 1 if failures else 0


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