code/apps/suite_c_scripts.py

"""Scripted (model-free) solutions for suite C, used to validate every task before it enters the suite.

    python apps/suite_c_scripts.py [name-filter]          REPEATS=3 -> each script must pass its check every run
    python apps/suite_c_scripts.py --probe URL [step ...]  interactive probe: print the observed actions after the steps

A script is a list of steps executed against jev's Browser (observe -> pick the action by label -> act):
    ("click", "label substring")          ("fill", "label substring", "text")
    ("select", "label substring", "option label substring")   ("enter",)   ("scroll", n)   ("sleep", seconds)
Labels match case-insensitively; a "re:" prefix means a regular expression; "#k" suffix picks the k-th match (0-based);
an "@role " prefix (e.g. "@button re:^Search$") restricts the match to that role.
Each task's check is run on the start page too (must FAIL there) and on the final page (must PASS).
"""
import json, os, re, sys, time
sys.path.insert(0, "/home/ckl/projects/S/jev-ultrafast")
os.environ.setdefault("BU_CDP_URL", "http://127.0.0.1:9222")
from jev_ultrafast.browser import Browser, StalePage  # noqa: E402


def observe(br, tries=12, settle=0.0):
    time.sleep(settle)
    for i in range(tries):
        try:
            return br.observe(screenshot=False)
        except StalePage:
            if i == tries - 1:
                raise
            time.sleep(0.5)


def find(page, kind, label):
    """First action of `kind` whose label matches; "@role " prefix restricts the role, "re:" = regex, "#k" = k-th match."""
    idx = 0; role = None
    if label.startswith("@"):
        role, _, label = label[1:].partition(" ")
    m = re.search(r"#(\d+)$", label)
    if m:
        idx, label = int(m.group(1)), label[: m.start()]
    if label.startswith("re:"):
        pat = re.compile(label[3:], re.I)
        hits = [a for a in page["actions"] if a["kind"] == kind and pat.search(a["label"])]
    else:
        hits = [a for a in page["actions"] if a["kind"] == kind and label.lower() in a["label"].lower()]
    if role:
        hits = [a for a in hits if a.get("role") == role]
    if len(hits) <= idx:
        raise LookupError(f"no {kind} action matching {label!r} (have {[a['label'][:40] for a in page['actions'] if a['kind']==kind][:40]})")
    return hits[idx]


def find_select(page, label, option):
    """Select action whose <select> label contains `label` and whose option label equals/contains `option`."""
    hits = [a for a in page["actions"] if a["kind"] == "select" and label.lower() in a["label"].rsplit(" → ", 1)[0].lower()]
    exact = [a for a in hits if a["label"].rsplit(" → ", 1)[1].strip().lower() == option.lower()]
    part = [a for a in hits if option.lower() in a["label"].rsplit(" → ", 1)[1].lower()]
    if not (exact or part):
        raise LookupError(f"no select option {option!r} in dropdown {label!r} (have {[a['label'][-45:] for a in hits][:30]})")
    return (exact or part)[0]


def run_steps(br, steps, verbose=False):
    """Execute the steps; returns (final page, number of browser actions performed)."""
    page = observe(br)
    n = 0
    for step in steps:
        op = step[0]
        if op == "sleep":
            page = observe(br, settle=step[1]); continue
        if op == "scroll":
            for _ in range(step[1] if len(step) > 1 else 1):
                a = next((x for x in page["actions"] if x["id"] == "scroll_down"), None)
                if a is None: break
                try:
                    br.act(a, page)
                except StalePage:                          # page changed under us: observe again, then scroll
                    page = observe(br, settle=0.5); continue
                page = observe(br); n += 1
            if verbose:
                print(f"   {n:2d}. scroll")
            continue
        for attempt in range(4):                       # retry a stale/covered target after a fresh observation
            try:
                if op == "click":
                    a = find(page, "click", step[1]); br.act(a, page)
                elif op == "fill":
                    a = find(page, "fill", step[1]); br.act(a, page, text=step[2])
                elif op == "select":
                    a = find_select(page, step[1], step[2]); br.act(a, page)
                elif op == "enter":
                    a = next(x for x in page["actions"] if x["kind"] == "key"); br.act(a, page)
                else:
                    raise ValueError(op)
                break
            except (StalePage, LookupError, StopIteration) as e:
                if attempt == 3:
                    raise
                time.sleep(0.8); page = observe(br)
        n += 1
        if verbose:
            print(f"   {n:2d}. {op} {step[1:]}")
        page = observe(br, settle=0.6)
    time.sleep(1.5)                                    # let a slow navigation land before judging (same as suite run())
    return observe(br), n


def show(page, limit=120):
    print("URL:", page["url"]); print("TITLE:", page["title"])
    print("TEXT:", page["text"][:500].replace("\n", " | "))
    for a in page["actions"][:limit]:
        extra = "" if a["kind"] != "select" else f"  [current={a.get('current_value')}]"
        chk = f" checked={a['checked']}" if "checked" in a else ""
        print(f"  {a['id']:5s} {a['kind']:6s} {a.get('role') or '':9s} {a['label'][:90]!r}{extra}{chk}")
    if len(page["actions"]) > limit:
        print(f"  ... {len(page['actions'])-limit} more")


def parse_cli_step(s):
    op, _, rest = s.partition(":")
    if op in ("enter",): return ("enter",)
    if op == "scroll": return ("scroll", int(rest or 1))
    if op == "sleep": return ("sleep", float(rest))
    if op == "fill":
        lab, _, txt = rest.partition("="); return ("fill", lab, txt)
    if op == "select":
        lab, _, opt = rest.partition("="); return ("select", lab, opt)
    return (op, rest)


# name -> steps (the start URL and the check live in browser_suite_c.TASKS)
SCRIPTS = {
    # ---- multi-step ----
    "met-sunflowers": [("fill", "Search by subject", "sunflowers"), ("enter",), ("click", "Has image"),
                       ("select", "Relevance", "Date (oldest-newest)")],
    "nuget-serilog-tool": [("fill", "Enter packages", "serilog"), ("enter",), ("click", "Package Type: .NET tool"),
                           ("select", "sort package", "Downloads")],
    "alpine-curl-filter": [("fill", "Package name", "curl"), ("select", "Branch", "v3.20"), ("select", "Repository", "main"),
                           ("select", "Architecture", "aarch64"), ("enter",)],
    "freesound-rain-cc0": [("fill", "Search sounds", "rain"), ("enter",), ("click", "Creative Commons 0"), ("click", "Soundscapes")],
    "ats-shampoo-haircare": [("fill", "Search Keywords", "shampoo"), ("enter",), ("select", "All Categories", "Hair Care"),
                             ("click", "Search in product descriptions"), ("click", "@button re:^Search$"), ("select", "Date Old", "Price High > Low")],
    "bnf-hugo-printed-p2": [("fill", "Rechercher une notice", "victor hugo"), ("click", "Submit"), ("click", "Texte imprimé"),
                            ("click", "Page suivante")],
    "vsm-python-installs": [("fill", "Search Visual Studio Code extensions", "python"), ("enter",), ("click", "Sort By"), ("click", "re:^Installs")],
    "wp-cache-commercial-redis": [("fill", "re:^Search$", "cache"), ("enter",), ("click", "Commercial"), ("click", "re:^Redis Object Cache")],
    "todomvc-active": [("fill", "What needs to be done", "buy milk"), ("enter",), ("fill", "What needs to be done", "walk the dog"),
                       ("enter",), ("click", "re:^Active$")],
    "setlist-radiohead-uk": [("fill", "Artist, Venue", "radiohead"), ("enter",), ("select", "Artist", "Radiohead ("),
                             ("select", "Country", "United Kingdom")],
    "jetbrains-rust-free": [("fill", "re:^Search", "rust"), ("enter",), ("click", "re:^free$"), ("click", "re:^plugin icon Rust  JetBrains")],
    "luarocks-rapidjson": [("fill", "Search modules", "json"), ("enter",), ("click", "Include non-root"), ("click", "@button re:^Search$"),
                           ("click", "re:^rapidjson$")],
    "letcode-dropdowns": [("select", "apple", "Apple"), ("select", "super hero", "Batman"), ("select", "programming language", "Swift"),
                          ("scroll", 1), ("select", "Select India", "India")],
    "letcode-radio": [("click", "re:^Foo$"), ("click", "re:^Going$"), ("click", "I agree"), ("click", "Remember me")],
    "clojars-ring-page3": [("fill", "Search projects", "ring"), ("enter",), ("scroll", 4), ("click", "re:^3$")],
    "fred-unemployment-pop": [("fill", "re:^Search", "unemployment rate"), ("enter",), ("click", "Sort by Relevance"), ("click", "re:^Popularity")],
    "modrinth-sodium": [("fill", "Search mods", "sodium"), ("enter",), ("click", "re:^1\\.21\\.11$"), ("click", "Sort by"), ("click", "re:^Downloads$")],
    "tvmaze-friends-episodes": [("fill", "Search Shows", "friends"), ("enter",), ("click", "re:^Friends$"), ("click", "re:^Episodes$")],
    "qaclickjet-form": [("click", "Round Trip"), ("click", "Senior Citizen"), ("select", "INR", "USD"), ("fill", "Type to Select", "India")],
    # ---- shorter ----
    "cocktail-margarita": [("fill", "Search for a Cocktail", "margarita"), ("enter",), ("click", "re:^Margarita")],
    "mealdb-arrabiata": [("fill", "Search for a Meal", "arrabiata"), ("enter",), ("click", "Spicy Arrabiata")],
    "gentoo-openrc-talk": [("fill", "Search Gentoo Wiki", "OpenRC"), ("enter",), ("click", "re:^Discussion$")],
    "webkit-css-bugs": [("click", "re:^Browse$"), ("click", "re:^WebKit \n"), ("click", "re:^CSS \n")],
    "govdata-wetter-energie": [("fill", "Suchbegriff", "Wetter"), ("enter",), ("click", "Energie")],
    "fedora-vim-common": [("fill", "re:^Search$", "vim"), ("enter",), ("click", "re:^vim-common$")],
    "racket-argo": [("fill", "Search packages", "json"), ("enter",), ("click", "re:^argo$")],
    "rdrr-ggplot": [("fill", "packages, doc text", "ggplot"), ("enter",)],
}


def main():
    if len(sys.argv) > 1 and sys.argv[1] == "--probe":
        br = Browser(sys.argv[2])
        try:
            page, n = run_steps(br, [parse_cli_step(s) for s in sys.argv[3:]], verbose=True)
            show(page)
        finally:
            br.close()
        return
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from browser_suite_c import TASKS, check_page, final_state  # noqa: E402
    flt = sys.argv[1] if len(sys.argv) > 1 else ""
    repeats = int(os.environ.get("REPEATS", "1"))
    rows = []
    shard, nshards = (int(v) for v in os.environ.get("SHARD", "0/1").split("/"))   # SHARD=i/n runs every n-th task
    for i, (name, url, goal, check, *extra) in enumerate(TASKS):
        if (flt and flt not in name) or i % nshards != shard: continue
        for rep in range(repeats):
            t0 = time.time(); ok = start_ok = None; n = 0; err = ""
            try:
                if extra: extra[0](url)                    # per-task setup (e.g. clear localStorage)
                br = Browser(url)
                try:
                    start_ok = check_page(name, check, final_state(br))
                    page, n = run_steps(br, SCRIPTS[name])
                    page = final_state(br)
                    ok = check_page(name, check, page)
                    final = page["url"]
                finally:
                    br.close()
            except Exception as e:
                err = f"{type(e).__name__}: {str(e)[:120]}"; final = ""
            verdict = "PASS" if (ok and not start_ok) else "FAIL"
            rows.append((name, verdict == "PASS", n))
            print(f"{verdict}  {name:26s} steps={n:2d} start_check={start_ok!s:5s} final_check={ok!s:5s} {time.time()-t0:5.1f}s {final[:60]} {err}", flush=True)
    good = sum(r[1] for r in rows)
    print(f"\n== {good}/{len(rows)} scripted runs passed")
    per = {}; steps = {}
    for r in rows: per.setdefault(r[0], []).append(r[1]); steps[r[0]] = max(steps.get(r[0], 0), r[2])
    print("   per task: " + "  ".join(f"{k}={sum(v)}/{len(v)}({steps[k]} steps)" for k, v in per.items()))
    json.dump({k: {"pass": sum(v), "runs": len(v), "steps": steps[k]} for k, v in per.items()},
              open(os.environ.get("SCRIPTS_OUT", "/tmp/suite_c_scripts.json"), "w"), indent=1)


if __name__ == "__main__":
    main()