File size: 12,444 Bytes
454b3e6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 | """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()
|