diff --git a/chv-guest/build.sh b/chv-guest/build.sh index 9dcfbe9..02ece4d 100755 --- a/chv-guest/build.sh +++ b/chv-guest/build.sh @@ -373,8 +373,11 @@ else echo "警告：cloud/chv/guest/vsock_notify.py 不在仓库里（P1 未合并），镜像暂不含 vsock notify" >&2 fi if [ "${PLAYWRIGHT:-1}" = "1" ]; then + # 阶段 4：与 playwright 同批 pip 装 pillow（diff_shot 的逐像素 + # 比对与 JPEG 编码用）。只加 pillow，不动其它包版本；BUILDINFO + # 的 pip freeze 会自然记录 pillow 版本，lockfile 按 pip 名覆盖。 VC_ARGS+=( - --run-command '(pip3 install --no-cache-dir --break-system-packages playwright || pip3 install --no-cache-dir playwright) && PLAYWRIGHT_BROWSERS_PATH=/usr/local/share/ms-playwright python3 -m playwright install --with-deps chromium-headless-shell' + --run-command '(pip3 install --no-cache-dir --break-system-packages playwright pillow || pip3 install --no-cache-dir playwright pillow) && PLAYWRIGHT_BROWSERS_PATH=/usr/local/share/ms-playwright python3 -m playwright install --with-deps chromium-headless-shell' --run-command 'dpkg-query -W -f "\${Package} \${Version}\n" > /app/guest-versions-dpkg.txt; pip3 freeze > /app/guest-versions-pip.txt || true' --run-command 'ls /usr/local/share/ms-playwright/ > /app/guest-versions-chromium.txt 2>/dev/null || true') fi diff --git a/chv-guest/buildinfo.py b/chv-guest/buildinfo.py index bbae164..f4bd74d 100644 --- a/chv-guest/buildinfo.py +++ b/chv-guest/buildinfo.py @@ -46,6 +46,10 @@ BASE_CONTENTS = [ # feat-chv-opt-landing（阶段 2）：cell veth 的 networkd # 不管配置（须排在 80-container-ve.network 之前）。 "79-chv-cell.network", + # feat-chv-opt-landing（阶段 4）：Pillow（pip 包，diff_shot + # 逐像素比对与全帧 JPEG 编码用；只加 pillow，不动其它包版本。 + # check/generate lockfile 按 pip 名覆盖，不走 DECLARED_SOURCES）。 + "pillow", ] # 执行器默认内核 cmdline（config.py），构建产物须与它一起记录（B-12）， diff --git a/chv-guest/chv-browser.py b/chv-guest/chv-browser.py index 4e8fa8c..83e0da7 100644 --- a/chv-guest/chv-browser.py +++ b/chv-guest/chv-browser.py @@ -24,8 +24,32 @@ executable_path，不走 channel/env）。 * 截图：JPEG quality 70，viewport 宽封顶 1280（上下文创建时 定死 1280x800）；diff_only 时与上一帧比对，字节相同即不落盘、 - 返回 changed=false（整帧相同跳过；变化区裁剪是阶段 4 的事， - 这里不冒充已做）。 + 返回 changed=false（整帧相同跳过）。 + * 阶段 4（feat-chv-opt-landing）：截图三级管线。 + 一级 a11y+DOM 主通道：snapshot 在 aria_snapshot 字符串基础 + 上纯字符串解析出角色计数（button/link/textbox/heading 等） + 与 interactive 总数、a11y 字符数与 a11y_hash——不跑 JS、不取 + 原始 DOM、不新增 evaluate/content 通路，禁 op 保持拒绝。 + 二级差分截图（op `diff_shot`）：先截 PNG 全帧，用 Pillow + 解码与上一帧逐像素比对——像素判变阈值 DIFF_PIXEL_THRESHOLD + =10（单通道差值 >10 才算变，滤掉抗锯齿 1 位抖动）；无变化 + 返回 changed:false size 0；有变化算出变化区 bbox（外扩 + margin 8px、钳制到视口内、宽高下限 32px），再用 + page.screenshot(type=jpeg, quality=70, clip=bbox) 只截变 + 化区。Pillow 缺失时 diff_shot 如实报错并指引回落全帧 + screenshot，绝不静默假装裁剪。full_size 是本帧 PNG 经 + Pillow 以 JPEG q70 编码得到的全帧字节数（与浏览器直出 + JPEG q70 同参数、字节数可能有数百字节差异，仅作对照）。 + 三级全屏校验（op `verify_shot`）：全帧 JPEG q70，专用于 + 少数校验场景。事件驱动口径：日常只在动作后调 diff_shot + （变化才出图），verify_shot 只在检查点调用。计数器 + shots_full/shots_diff/shots_verify/shots_skipped 由 + status 返回，verify_share = shots_verify / + (shots_full+shots_diff+shots_verify)，分母为 0 时记 0.0。 + 计数口径：shots_full = screenshot 落盘数；shots_diff = + diff_shot 返回 changed:true 的数；shots_verify = + verify_shot 成功数；shots_skipped = screenshot diff_only + 跳过数 + diff_shot changed:false 数。 对齐 Muse P2 的不变口径：子代理只向单元返回页面无障碍快照， 不见原始 DOM，不能在页面上下文跑 JS，不暴露 evaluate/content， @@ -35,10 +59,19 @@ page.accessibility，改用 page.aria_snapshot()（返回 YAML 式 操作（Unix socket /run/chv/browser.sock，JSON 行）： open {url, session?} -> 导航并返回 {title, url, a11y, truncated} - snapshot {session?} -> 当前页的 {title, url, a11y, truncated} + snapshot {session?} -> 当前页的 {title, url, a11y, truncated, + a11y_roles, a11y_interactive, + a11y_chars, a11y_hash} screenshot {session?, diff_only?} -> JPEG 落盘 /run/chv/shots/.jpg， 返回 {path, size, format, quality, width, changed, sha256} + diff_shot {session?} -> PNG 全帧逐像素比对；无变化 + {changed:false, size:0}；有变化只截变 + 化区 JPEG，返回 {changed:true, bbox, + size, full_size, path, sha256} + verify_shot {session?} -> 全帧 JPEG q70（只用于检查点）， + 返回 {path, size, format, quality, + width, sha256} close {session?} -> 关该会话的页（浏览器进程继续跑） sessions {} -> 各会话 {session, url, title} status {} -> {engine, profile, profile_on_data_disk, @@ -95,6 +128,20 @@ MAX_REQUEST_BYTES = 65536 MAX_A11Y_NODES = 200 MAX_A11Y_CHARS = 65536 NAV_TIMEOUT_MS = 30000 +# 阶段 4 差分截图口径（写死在代码里，驱动与报告按此口径实测）： +# 像素判变阈值：单通道差值 >10 才算变（滤抗锯齿 1 位抖动）； +# bbox 外扩 margin 8px；变化区宽高下限 32px（不足则绕中心扩到 +# 32px 再钳制到视口内）。 +DIFF_PIXEL_THRESHOLD = 10 +DIFF_MARGIN_PX = 8 +DIFF_MIN_SIZE_PX = 32 +# a11y 角色里算 interactive 的集合（aria_snapshot 字符串解析用； +# searchbox 是 textbox 的搜索变体，同样算可交互）。 +INTERACTIVE_ROLES = frozenset([ + "button", "link", "textbox", "searchbox", "checkbox", "radio", + "combobox", "listbox", "menuitem", "tab", "switch", "slider", + "spinbutton", +]) # 拒绝的 URL 主机：localhost 与私有地址字面量（主机名判定）。 _PRIVATE_HOST_RE = re.compile( @@ -215,6 +262,137 @@ def truncate_a11y(node, max_nodes=MAX_A11Y_NODES, return tree, state["truncated"] +def parse_a11y(a11y): + """解析无障碍快照出角色计数（纯函数，可单测）。 + + 输入是 aria_snapshot 字符串（主路）或旧版 dict 树（回落）。 + 字符串按行解析：行形如 `- link "More information..." [ref=e3]`， + 嵌套行带前导空格；角色取 `- ` 后的第一个 token。dict 树则 + 递归数每个节点的 role 字段。返回 (roles, interactive, chars, + a11y_hash)：roles 是 {角色: 个数}，interactive 是 + INTERACTIVE_ROLES 内的角色总数，chars 是快照字符数， + a11y_hash 是快照规范文本的 sha256（字符串用原文，dict 用 + sort_keys 的 JSON）。不跑 JS、不取原始 DOM。""" + roles = {} + if isinstance(a11y, str): + text = a11y + for line in text.splitlines(): + m = re.match(r"^\s*-\s+([A-Za-z][\w-]*)", line) + if m: + role = m.group(1).lower() + roles[role] = roles.get(role, 0) + 1 + elif isinstance(a11y, dict): + text = json.dumps(a11y, ensure_ascii=False, sort_keys=True) + + def walk(n): + if not isinstance(n, dict): + return + role = n.get("role") + if isinstance(role, str) and role: + key = role.lower() + roles[key] = roles.get(key, 0) + 1 + for value in n.values(): + if isinstance(value, list): + for child in value: + walk(child) + elif isinstance(value, dict): + walk(value) + + walk(a11y) + else: + text = "" + interactive = sum(count for role, count in roles.items() + if role in INTERACTIVE_ROLES) + digest = hashlib.sha256(text.encode()).hexdigest() + return roles, interactive, len(text), digest + + +def expand_bbox(x0, y0, x1, y1, width, height, + margin=DIFF_MARGIN_PX, min_size=DIFF_MIN_SIZE_PX): + """变化区 bbox 外扩 + 钳制 + 最小尺寸（纯函数，可单测）。 + + 输入是像素差分的包围盒（左闭右开 [x0,y0,x1,y1)）。先外扩 + margin，再钳制到 [0,0,width,height] 内；宽/高不足 min_size + 时绕中心扩到 min_size 再钳制（图本身比 min_size 还小时取 + 整幅宽/高）。返回 Playwright clip 形 {"x","y","width", + "height"}（均为 int）。""" + x0 = max(0, x0 - margin) + y0 = max(0, y0 - margin) + x1 = min(width, x1 + margin) + y1 = min(height, y1 + margin) + # 宽/高不足 min_size 时绕中心扩；一侧贴边扩不动时向另一侧 + # 补足（只有图本身比 min_size 还窄/矮时才取整幅宽/高）。 + for axis in ("x", "y"): + lo, hi, size = (x0, x1, width) if axis == "x" \ + else (y0, y1, height) + if hi - lo < min_size: + need = min_size - (hi - lo) + lo = max(0, lo - need // 2) + hi = min(size, hi + (need - need // 2)) + if hi - lo < min_size: + if lo == 0: + hi = min(size, lo + min_size) + elif hi == size: + lo = max(0, hi - min_size) + if hi - lo < min_size: + lo, hi = 0, size + if axis == "x": + x0, x1 = lo, hi + else: + y0, y1 = lo, hi + return {"x": int(x0), "y": int(y0), + "width": int(x1 - x0), "height": int(y1 - y0)} + + +def _load_pillow(): + """懒加载 Pillow（diff_shot 专用）；缺失时报明错并指引回落 + 全帧 screenshot，绝不静默假装裁剪。返回 (Image, ImageChops)。""" + try: + from PIL import Image, ImageChops + except ImportError as e: + raise RuntimeError( + "diff_shot 需要 Pillow（%s）：镜像里 pip 未装 pillow，" + "无法逐像素比对与裁剪；回落：用 screenshot 全帧 JPEG " + "q70（op screenshot，不带 diff_only）" % e) + return Image, ImageChops + + +def compute_diff_bbox(prev_png, cur_png, + margin=DIFF_MARGIN_PX, + min_size=DIFF_MIN_SIZE_PX, + pillow=None): + """两帧 PNG 的逐像素比对（纯函数，可单测，需 Pillow）。 + + 返回 (changed, bbox)：无变化时 (False, None)；有变化时 + (True, clip dict)。像素判变：逐通道差值 > DIFF_PIXEL_THRESHOLD + 才算变（见模块 docstring）。两帧尺寸不同时视为整帧变化， + bbox 即当前帧全幅。pillow 供单测注入 (Image, ImageChops)， + 正常运行时用 _load_pillow()。""" + Image, ImageChops = pillow or _load_pillow() + import io + prev = Image.open(io.BytesIO(prev_png)).convert("RGB") + cur = Image.open(io.BytesIO(cur_png)).convert("RGB") + width, height = cur.size + if prev.size != cur.size: + return True, {"x": 0, "y": 0, "width": width, + "height": height} + diff = ImageChops.difference(prev, cur) + # 逐通道阈值掩码：任一通道差值 >10 即变（convert("L") 会按 + # 亮度加权、漏掉单通道小变化，故三通道分别取阈值再取并）。 + masks = [] + for channel in diff.split(): + masks.append(channel.point( + lambda p: 255 if p > DIFF_PIXEL_THRESHOLD else 0)) + mask = masks[0] + for extra in masks[1:]: + mask = ImageChops.lighter(mask, extra) + box = mask.getbbox() + if box is None: + return False, None + return True, expand_bbox(box[0], box[1], box[2], box[3], + width, height, margin, min_size) + + def _load_playwright(): """懒加载 playwright；缺失时报明错（不静默假装有浏览器）。 返回 sync_playwright() 的上下文管理器实例（须调用， @@ -278,6 +456,10 @@ class BrowserAgent: self._engine = "shell" self._mounted_full = False self._last_shot = {} + # 阶段 4：上一帧 PNG（diff_shot 的比对基线）与截图计数器。 + self._last_png = {} + self._counters = {"shots_full": 0, "shots_diff": 0, + "shots_verify": 0, "shots_skipped": 0} # -- 浏览器生命周期 ---------------------------------------------- def _close_context(self): @@ -394,6 +576,11 @@ class BrowserAgent: pass page = self._context.new_page() self._pages[session] = page + # 注意：open 不清 diff 基线——diff_shot 永远与“上一帧” + # 逐像素比对，不管上一帧是哪一页；导航后调 diff_shot + # 得到的正是新旧两页渲染的真实差异区（跨页局部变化 + # 如搜索词 Cat->Dog 就靠这个）。只有 close（会话结束） + # 才清基线，下次 open 后的首次 diff_shot 为全幅首帧。 page.goto(url, timeout=NAV_TIMEOUT_MS) page.wait_for_load_state("domcontentloaded", timeout=NAV_TIMEOUT_MS) @@ -431,8 +618,12 @@ class BrowserAgent: "aria_snapshot/accessibility）"} except Exception as e: return {"ok": False, "error": "取快照失败：%s" % e} + # 阶段 4 一级通道：纯字符串解析 enrich（不跑 JS、不取 DOM）。 + roles, interactive, chars, digest = parse_a11y(tree) return {"ok": True, "title": title, "url": url, "a11y": tree, "truncated": truncated, + "a11y_roles": roles, "a11y_interactive": interactive, + "a11y_chars": chars, "a11y_hash": digest, "session": session} def screenshot(self, session=None, diff_only=False): @@ -453,7 +644,8 @@ class BrowserAgent: digest = hashlib.sha256(data).hexdigest() if diff_only and self._last_shot.get(session) == digest: # 整帧与上一帧字节相同：跳过落盘（阶段 4 的变化区 - # 裁剪不在本阶段做，这里只做整帧相同跳过）。 + # 裁剪走 diff_shot op，这里只做整帧相同跳过）。 + self._counters["shots_skipped"] += 1 return {"ok": True, "changed": False, "size": 0, "path": None, "format": "jpeg", "quality": JPEG_QUALITY, @@ -467,11 +659,124 @@ class BrowserAgent: except OSError as e: return {"ok": False, "error": "截图落盘失败：%s" % e} self._last_shot[session] = digest + self._counters["shots_full"] += 1 return {"ok": True, "path": path, "size": len(data), "format": "jpeg", "quality": JPEG_QUALITY, "width": VIEWPORT_WIDTH, "changed": True, "sha256": digest, "session": session} + def diff_shot(self, session=None): + """二级差分截图（阶段 4）：PNG 全帧逐像素比对，只截变化区。 + + 无变化：{changed:false, size:0}（计 shots_skipped）。 + 有变化：clip=bbox 的 JPEG q70 落盘，返回 {changed:true, + bbox, size, full_size, path, sha256}（计 shots_diff）。 + 首帧（本会话无上一帧）视为整帧变化，bbox 为全视口。 + Pillow 缺失时 ok:false 并指引回落 screenshot 全帧， + 不计入计数器、不落盘、不假装裁剪。""" + session = session or DEFAULT_SESSION + page = self._page(session) + if page is None: + return {"ok": False, + "error": "会话 %s 还没有打开的页面（先 open）" + % session} + try: + png = page.screenshot(type="png") + except Exception as e: + return {"ok": False, "error": "截图失败：%s" % e} + if not isinstance(png, (bytes, bytearray)): + return {"ok": False, "error": "截图未返回字节数据"} + png = bytes(png) + try: + Image, _ImageChops = _load_pillow() + except RuntimeError as e: + return {"ok": False, "error": str(e), + "fallback": "screenshot"} + prev = self._last_png.get(session) + if prev is None: + changed, bbox = True, {"x": 0, "y": 0, + "width": VIEWPORT_WIDTH, + "height": VIEWPORT_HEIGHT} + else: + try: + changed, bbox = compute_diff_bbox(prev, png) + except Exception as e: + return {"ok": False, + "error": "差分比对失败：%s" % e} + # full_size：本帧 PNG 经 Pillow 以 JPEG q70 编码的全帧 + # 字节数（与浏览器直出同参数，仅作对照，不直接等同）。 + import io + try: + buf = io.BytesIO() + Image.open(io.BytesIO(png)).convert("RGB").save( + buf, "JPEG", quality=JPEG_QUALITY) + full_size = len(buf.getvalue()) + except Exception: + full_size = len(png) + self._last_png[session] = png + if not changed: + self._counters["shots_skipped"] += 1 + return {"ok": True, "changed": False, "size": 0, + "path": None, "bbox": None, + "full_size": full_size, "format": "jpeg", + "quality": JPEG_QUALITY, "session": session} + try: + clip = page.screenshot(type="jpeg", + quality=JPEG_QUALITY, clip=bbox) + except Exception as e: + return {"ok": False, "error": "变化区截图失败：%s" % e} + if not isinstance(clip, (bytes, bytearray)): + return {"ok": False, "error": "变化区截图未返回字节数据"} + clip = bytes(clip) + try: + os.makedirs(self._shots_dir, exist_ok=True) + path = os.path.join(self._shots_dir, "%s.jpg" + % uuid.uuid4().hex) + with open(path, "wb") as fh: + fh.write(clip) + except OSError as e: + return {"ok": False, "error": "截图落盘失败：%s" % e} + self._counters["shots_diff"] += 1 + return {"ok": True, "changed": True, "bbox": bbox, + "path": path, "size": len(clip), + "full_size": full_size, "format": "jpeg", + "quality": JPEG_QUALITY, + "sha256": hashlib.sha256(clip).hexdigest(), + "session": session} + + def verify_shot(self, session=None): + """三级全屏校验（阶段 4）：全帧 JPEG q70，只用于检查点。 + 计 shots_verify。日常动作后用 diff_shot，检查点才调它。""" + session = session or DEFAULT_SESSION + page = self._page(session) + if page is None: + return {"ok": False, + "error": "会话 %s 还没有打开的页面（先 open）" + % session} + try: + data = page.screenshot(type="jpeg", + quality=JPEG_QUALITY) + except Exception as e: + return {"ok": False, "error": "截图失败：%s" % e} + if not isinstance(data, (bytes, bytearray)): + return {"ok": False, "error": "截图未返回字节数据"} + data = bytes(data) + try: + os.makedirs(self._shots_dir, exist_ok=True) + path = os.path.join(self._shots_dir, "%s.jpg" + % uuid.uuid4().hex) + with open(path, "wb") as fh: + fh.write(data) + except OSError as e: + return {"ok": False, "error": "截图落盘失败：%s" % e} + self._last_shot[session] = hashlib.sha256(data).hexdigest() + self._counters["shots_verify"] += 1 + return {"ok": True, "path": path, "size": len(data), + "format": "jpeg", "quality": JPEG_QUALITY, + "width": VIEWPORT_WIDTH, "changed": True, + "sha256": self._last_shot[session], + "session": session} + def close(self, session=None): session = session or DEFAULT_SESSION page = self._pages.pop(session, None) @@ -481,6 +786,7 @@ class BrowserAgent: except Exception: pass self._last_shot.pop(session, None) + self._last_png.pop(session, None) return {"ok": True, "closed": True, "session": session} def sessions(self): @@ -499,6 +805,11 @@ class BrowserAgent: # -- 服务化：状态 / CDP / 引擎 -------------------------------------- def status(self): + total = (self._counters["shots_full"] + + self._counters["shots_diff"] + + self._counters["shots_verify"]) + verify_share = (self._counters["shots_verify"] / total + if total else 0.0) return {"ok": True, "engine": self._engine, "profile": self._profile_dir, "profile_on_data_disk": self._profile_on_data_disk, @@ -509,7 +820,9 @@ class BrowserAgent: "mounted_full": self._mounted_full, "full_present": full_chromium_present( self._full_browsers_dir), - "full_browsers_dir": self._full_browsers_dir} + "full_browsers_dir": self._full_browsers_dir, + "counters": dict(self._counters), + "verify_share": verify_share} def cdp(self): ok, err = self._ensure_browser() @@ -607,6 +920,10 @@ def handle_request(agent, req): if op == "screenshot": return agent.screenshot(req.get("session"), bool(req.get("diff_only"))) + if op == "diff_shot": + return agent.diff_shot(req.get("session")) + if op == "verify_shot": + return agent.verify_shot(req.get("session")) if op == "close": return agent.close(req.get("session")) if op == "sessions": @@ -623,6 +940,7 @@ def handle_request(agent, req): return agent.unmount_full() return {"ok": False, "error": "未知操作：%r（白名单：open/snapshot/screenshot/" + "diff_shot/verify_shot/" "close/sessions/status/cdp/engine/mount_full/" "unmount_full/ping）" % (op,)} diff --git a/cloud/tests/test_chv_opt_stage4.py b/cloud/tests/test_chv_opt_stage4.py new file mode 100644 index 0000000..c305985 --- /dev/null +++ b/cloud/tests/test_chv_opt_stage4.py @@ -0,0 +1,562 @@ +"""阶段 4（截图三级管线）测试：a11y 角色解析 + 差分截图 bbox + +全屏校验 + 计数器 + Pillow 缺失回落 + 禁 op 仍拒绝 + build 接线。 + +只测本轨交付物（chv-guest/chv-browser.py、build.sh、 +buildinfo.py、check/generate_lockfile.py）。全部用假 +playwright/假 PNG（标准库 zlib 编码）/假文件系统，绝不连真实 +网络、不起真实 VM。Pillow 正向路径用本机已装的 Pillow 跑； +缺失路径用注入的 _load_pillow 假失败覆盖。 +""" +import hashlib +import importlib.util +import json +import os +import struct +import subprocess +import sys +import tempfile +import unittest +import zlib +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] + + +def _load_script(name): + path = ROOT / "chv-guest" / name + spec = importlib.util.spec_from_file_location( + "chv_guest_s4_" + name[:-3], str(path)) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +BROWSER = _load_script("chv-browser.py") +BUILDINFO = _load_script("buildinfo.py") +CHECK = _load_script("check_lockfile.py") +GENERATE = _load_script("generate_lockfile.py") + + +def encode_png(width, height, pixels): + """标准库编码 RGB PNG：pixels 是 {(x, y): (r, g, b)}，缺省白。 + 供 compute_diff_bbox / diff_shot 的假帧用（不依赖 Pillow 编码）。""" + def chunk(kind, data): + out = struct.pack(">I", len(data)) + kind + data + return out + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF) + + ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) + raw = bytearray() + for y in range(height): + raw.append(0) + for x in range(width): + raw.extend(pixels.get((x, y), (255, 255, 255))) + return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + + chunk(b"IDAT", zlib.compress(bytes(raw))) + + chunk(b"IEND", b"")) + + +def solid_png(width, height, color=(255, 255, 255)): + return encode_png(width, height, { + (x, y): color for x in range(width) for y in range(height)}) + + +ARIA_SAMPLE = """- generic [ref=e1]: + - heading "Example Domain" [level=1] + - paragraph "This domain is for use in illustrative examples" + - link "More information..." [ref=e3] + - button "Search" [ref=e4] + - textbox "Keywords" [ref=e5] + - searchbox "Site search" [ref=e6] + - checkbox "Remember me" [checked] + - heading "Second heading" [level=2] +""" + + +class FakePage: + def __init__(self, title="Example Domain", aria=None): + self.url = "about:blank" + self._title = title + self._aria = aria if aria is not None else ARIA_SAMPLE + self._frame = solid_png(64, 48) + self.shots = [] + + def set_frame(self, png_bytes): + self._frame = png_bytes + + def goto(self, url, timeout=None): + self.url = url.rstrip("/") + "/" + + def wait_for_load_state(self, state, timeout=None): + pass + + def title(self): + return self._title + + def aria_snapshot(self): + return self._aria + + def screenshot(self, path=None, type=None, quality=None, clip=None): + self.shots.append((type, quality, clip)) + if type == "png": + return self._frame + if clip is not None: + return (b"\xff\xd8CLIP-%d-%d-%d-%d" + % (clip["x"], clip["y"], clip["width"], + clip["height"])) + assert type == "jpeg", type + assert quality == 70, quality + return b"\xff\xd8\xff\xe0FAKEJPEG-" + self._title.encode() + + def close(self): + pass + + +class FakeContext: + def __init__(self, pages): + self._pages = list(pages) + self._idx = 0 + self.launches = [] + + def new_page(self): + page = self._pages[self._idx % len(self._pages)] + self._idx += 1 + return page + + def close(self): + pass + + +class FakeEntered: + def __init__(self, context): + self.chromium = mock.Mock() + self._context = context + self.calls = [] + + def launch_persistent(user_data_dir, **kwargs): + self.calls.append((user_data_dir, kwargs)) + return context + + self.chromium.launch_persistent_context.side_effect = \ + launch_persistent + + +def _agent(pages, tmp): + ctx = FakeContext(pages) + holder = {} + + class PW: + def __enter__(self): + entered = FakeEntered(ctx) + holder["entered"] = entered + return entered + + def __exit__(self, *exc): + return False + + agent = BROWSER.BrowserAgent( + playwright_factory=lambda: PW(), shots_dir=tmp, + profile_dir=os.path.join(tmp, "profile")) + return agent, ctx, holder + + +class ParseA11yTest(unittest.TestCase): + def test_counts_roles_and_interactive(self): + roles, interactive, chars, digest = BROWSER.parse_a11y( + ARIA_SAMPLE) + self.assertEqual(roles.get("heading"), 2) + self.assertEqual(roles.get("link"), 1) + self.assertEqual(roles.get("button"), 1) + self.assertEqual(roles.get("textbox"), 1) + self.assertEqual(roles.get("searchbox"), 1) + self.assertEqual(roles.get("checkbox"), 1) + self.assertEqual(roles.get("paragraph"), 1) + self.assertEqual(roles.get("generic"), 1) + # interactive = button+link+textbox+searchbox+checkbox = 5 + self.assertEqual(interactive, 5) + self.assertEqual(chars, len(ARIA_SAMPLE)) + self.assertEqual( + digest, hashlib.sha256(ARIA_SAMPLE.encode()).hexdigest()) + + def test_hash_stable_and_changes_with_text(self): + _, _, _, h1 = BROWSER.parse_a11y(ARIA_SAMPLE) + _, _, _, h2 = BROWSER.parse_a11y(ARIA_SAMPLE) + self.assertEqual(h1, h2) + _, _, _, h3 = BROWSER.parse_a11y(ARIA_SAMPLE + "- link \"x\"") + self.assertNotEqual(h1, h3) + + def test_empty_and_no_roles(self): + roles, interactive, chars, _ = BROWSER.parse_a11y("") + self.assertEqual(roles, {}) + self.assertEqual(interactive, 0) + self.assertEqual(chars, 0) + roles, interactive, _, _ = BROWSER.parse_a11y( + "plain text without roles\nsecond line") + self.assertEqual(roles, {}) + + def test_dict_tree_fallback_counts_roles(self): + tree = {"role": "WebArea", "name": "x", "children": [ + {"role": "heading", "name": "h"}, + {"role": "link", "name": "l"}, + {"role": "button", "name": "b"}]} + roles, interactive, chars, _ = BROWSER.parse_a11y(tree) + self.assertEqual(roles.get("heading"), 1) + self.assertEqual(roles.get("link"), 1) + self.assertEqual(roles.get("button"), 1) + self.assertEqual(interactive, 2) + self.assertGreater(chars, 0) + + +class ExpandBboxTest(unittest.TestCase): + def test_center_small_change_expands_to_min_size(self): + # 1px 变化在 (32,24)：margin 8 -> 17x17，再扩到 32x32。 + box = BROWSER.expand_bbox(32, 24, 33, 25, 1280, 800) + self.assertEqual(box, {"x": 17, "y": 9, + "width": 32, "height": 32}) + + def test_top_edge_extends_downward_not_full_height(self): + # 贴顶边的变化：y=0 侧扩不动，向下补足 32，不拉满 800。 + box = BROWSER.expand_bbox(120, 10, 128, 20, 1280, 800) + self.assertEqual(box["y"], 0) + self.assertEqual(box["height"], 32) + self.assertEqual(box["width"], 32) + + def test_corner_clamps_to_zero(self): + box = BROWSER.expand_bbox(0, 0, 1, 1, 1280, 800) + self.assertEqual(box["x"], 0) + self.assertEqual(box["y"], 0) + self.assertEqual(box["width"], 32) + self.assertEqual(box["height"], 32) + + def test_full_frame_change_stays_full(self): + box = BROWSER.expand_bbox(0, 0, 1280, 800, 1280, 800) + self.assertEqual(box, {"x": 0, "y": 0, "width": 1280, + "height": 800}) + + def test_image_smaller_than_min_size_takes_full_extent(self): + box = BROWSER.expand_bbox(10, 10, 11, 11, 20, 20) + self.assertEqual(box, {"x": 0, "y": 0, "width": 20, + "height": 20}) + + +class ComputeDiffTest(unittest.TestCase): + def test_identical_frames_no_change(self): + png = solid_png(64, 48) + changed, bbox = BROWSER.compute_diff_bbox(png, png) + self.assertFalse(changed) + self.assertIsNone(bbox) + + def test_single_pixel_change_found_with_margin_and_min_size(self): + prev = solid_png(64, 48) + cur = encode_png(64, 48, {(32, 24): (0, 0, 0)}) + changed, bbox = BROWSER.compute_diff_bbox(prev, cur) + self.assertTrue(changed) + self.assertEqual(bbox, {"x": 17, "y": 9, + "width": 32, "height": 32}) + + def test_below_threshold_change_ignored(self): + # 每通道只差 5（阈值 10）：抗锯齿抖动级，不算变。 + base = {(x, y): (100, 100, 100) + for x in range(64) for y in range(48)} + prev = encode_png(64, 48, base) + cur_pixels = dict(base) + cur_pixels[(32, 24)] = (105, 105, 105) + cur = encode_png(64, 48, cur_pixels) + changed, _ = BROWSER.compute_diff_bbox(prev, cur) + self.assertFalse(changed) + + def test_single_channel_change_above_threshold_counts(self): + # 只蓝通道差 50：亮度加权会把它压到 ~6，逐通道阈值必须抓到。 + base = {(x, y): (100, 100, 100) + for x in range(64) for y in range(48)} + prev = encode_png(64, 48, base) + cur_pixels = dict(base) + cur_pixels[(32, 24)] = (100, 100, 150) + cur = encode_png(64, 48, cur_pixels) + changed, bbox = BROWSER.compute_diff_bbox(prev, cur) + self.assertTrue(changed) + self.assertIsNotNone(bbox) + + def test_size_mismatch_is_full_frame_change(self): + prev = solid_png(64, 48) + cur = solid_png(32, 24) + changed, bbox = BROWSER.compute_diff_bbox(prev, cur) + self.assertTrue(changed) + self.assertEqual(bbox, {"x": 0, "y": 0, "width": 32, + "height": 24}) + + +class SnapshotEnrichTest(unittest.TestCase): + def test_snapshot_carries_role_counts(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + agent.open("https://example.com") + snap = agent.snapshot() + self.assertTrue(snap["ok"]) + self.assertEqual(snap["a11y_roles"]["heading"], 2) + self.assertEqual(snap["a11y_roles"]["link"], 1) + self.assertEqual(snap["a11y_interactive"], 5) + self.assertEqual(snap["a11y_chars"], len(ARIA_SAMPLE)) + self.assertEqual( + snap["a11y_hash"], + hashlib.sha256(ARIA_SAMPLE.encode()).hexdigest()) + + def test_snapshot_hash_changes_when_page_changes(self): + with tempfile.TemporaryDirectory() as tmp: + page = FakePage(aria="- link \"a\"") + agent, _c, _h = _agent([page], tmp) + agent.open("https://example.com") + h1 = agent.snapshot()["a11y_hash"] + page._aria = "- link \"b\"" + h2 = agent.snapshot()["a11y_hash"] + self.assertNotEqual(h1, h2) + + +class DiffShotTest(unittest.TestCase): + def test_first_shot_is_full_frame_change(self): + with tempfile.TemporaryDirectory() as tmp: + page = FakePage() + agent, _c, _h = _agent([page], tmp) + agent.open("https://example.com") + resp = agent.diff_shot() + self.assertTrue(resp["ok"]) + self.assertTrue(resp["changed"]) + self.assertEqual(resp["bbox"], {"x": 0, "y": 0, + "width": 1280, + "height": 800}) + self.assertGreater(resp["size"], 0) + self.assertGreater(resp["full_size"], 0) + self.assertTrue(resp["path"].endswith(".jpg")) + with open(resp["path"], "rb") as fh: + self.assertEqual(fh.read(2), b"\xff\xd8") + # clip 调的是全视口 bbox。 + self.assertEqual(page.shots[-1], + ("jpeg", 70, resp["bbox"])) + + def test_identical_second_shot_skipped(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + agent.open("https://example.com") + first = agent.diff_shot() + self.assertTrue(first["changed"]) + second = agent.diff_shot() + self.assertTrue(second["ok"]) + self.assertFalse(second["changed"]) + self.assertEqual(second["size"], 0) + self.assertIsNone(second["path"]) + self.assertIsNone(second["bbox"]) + self.assertEqual(agent.status()["counters"]["shots_skipped"], + 1) + self.assertEqual(agent.status()["counters"]["shots_diff"], + 1) + + def test_local_change_clips_to_bbox(self): + with tempfile.TemporaryDirectory() as tmp: + page = FakePage() + agent, _c, _h = _agent([page], tmp) + agent.open("https://example.com") + agent.diff_shot() + page.set_frame(encode_png(64, 48, {(32, 24): (0, 0, 0)})) + resp = agent.diff_shot() + self.assertTrue(resp["changed"]) + self.assertEqual(resp["bbox"], {"x": 17, "y": 9, + "width": 32, + "height": 32}) + # 裁剪图字节里带 bbox 坐标（假实现按 clip 编码）。 + self.assertIn(b"CLIP-17-9-32-32", + open(resp["path"], "rb").read()) + self.assertLess(resp["size"], resp["full_size"]) + + def test_open_keeps_diff_baseline_across_navigation(self): + # open 不清基线：同像素帧重导航后 diff_shot 应判无变化 + # （上一帧就是这帧）；帧变了才判变化，且 bbox 为真实差异区。 + with tempfile.TemporaryDirectory() as tmp: + page = FakePage() + agent, _c, _h = _agent([page, page], tmp) + agent.open("https://example.com") + agent.diff_shot() + agent.diff_shot() + agent.open("https://example.org") + same = agent.diff_shot() + self.assertTrue(same["ok"]) + self.assertFalse(same["changed"]) + page.set_frame(encode_png(64, 48, {(32, 24): (0, 0, 0)})) + changed = agent.diff_shot() + self.assertTrue(changed["changed"]) + self.assertEqual(changed["bbox"], {"x": 17, "y": 9, + "width": 32, + "height": 32}) + + def test_close_clears_diff_baseline(self): + with tempfile.TemporaryDirectory() as tmp: + page = FakePage() + agent, _c, _h = _agent([page], tmp) + agent.open("https://example.com") + agent.diff_shot() + agent.diff_shot() + agent.close() + self.assertFalse(agent.diff_shot()["ok"]) + agent.open("https://example.com") + resp = agent.diff_shot() + self.assertTrue(resp["changed"]) + self.assertEqual(resp["bbox"], {"x": 0, "y": 0, + "width": 1280, + "height": 800}) + + +class VerifyShotTest(unittest.TestCase): + def test_verify_shot_is_full_frame(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + agent.open("https://example.com") + resp = agent.verify_shot() + self.assertTrue(resp["ok"]) + self.assertEqual(resp["format"], "jpeg") + self.assertEqual(resp["quality"], 70) + self.assertEqual(resp["width"], 1280) + self.assertTrue(resp["path"].endswith(".jpg")) + self.assertEqual(agent.status()["counters"]["shots_verify"], + 1) + # verify 不带 clip：假实现最后一次截图无 clip 参数。 + self.assertEqual(agent._pages["default"].shots[-1], + ("jpeg", 70, None)) + + +class CountersTest(unittest.TestCase): + def test_counters_and_verify_share(self): + with tempfile.TemporaryDirectory() as tmp: + page = FakePage() + agent, _c, _h = _agent([page], tmp) + agent.open("https://example.com") + agent.screenshot() # full +1 + agent.screenshot(diff_only=True) # skip +1（同帧） + agent.diff_shot() # diff +1（首帧） + agent.diff_shot() # skip +1（同帧） + agent.verify_shot() # verify +1 + st = agent.status() + self.assertEqual(st["counters"], { + "shots_full": 1, "shots_diff": 1, + "shots_verify": 1, "shots_skipped": 2}) + self.assertAlmostEqual(st["verify_share"], 1 / 3) + + def test_verify_share_zero_when_no_shots(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + st = agent.status() + self.assertEqual(st["verify_share"], 0.0) + self.assertEqual(st["counters"]["shots_full"], 0) + + +class PillowMissingTest(unittest.TestCase): + def test_diff_shot_without_pillow_errors_and_names_fallback(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + agent.open("https://example.com") + before = agent.status()["counters"] + + def _no_pillow(): + raise RuntimeError( + "diff_shot 需要 Pillow（No module named 'PIL'）：" + "镜像里 pip 未装 pillow，无法逐像素比对与裁剪；" + "回落：用 screenshot 全帧 JPEG q70") + + with mock.patch.object(BROWSER, "_load_pillow", + side_effect=_no_pillow): + resp = agent.diff_shot() + self.assertFalse(resp["ok"]) + self.assertIn("Pillow", resp["error"]) + self.assertIn("screenshot", resp["error"]) + self.assertEqual(resp["fallback"], "screenshot") + # 不计入计数器、不落盘。 + self.assertEqual(agent.status()["counters"], before) + self.assertEqual( + [f for f in os.listdir(tmp) if f.endswith(".jpg")], + []) + + +class ForbiddenOpsTest(unittest.TestCase): + def test_forbidden_ops_still_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + for op in ("evaluate", "content", "devtools", "js"): + resp = BROWSER.handle_request(agent, {"op": op}) + self.assertFalse(resp["ok"], op) + + def test_new_ops_routed(self): + with tempfile.TemporaryDirectory() as tmp: + agent, _c, _h = _agent([FakePage()], tmp) + agent.open("https://example.com") + self.assertTrue(BROWSER.handle_request( + agent, {"op": "diff_shot"})["ok"]) + self.assertTrue(BROWSER.handle_request( + agent, {"op": "verify_shot"})["ok"]) + resp = BROWSER.handle_request(agent, {"op": "nope"}) + self.assertFalse(resp["ok"]) + self.assertIn("diff_shot", resp["error"]) + self.assertIn("verify_shot", resp["error"]) + + +class BuildWiringTest(unittest.TestCase): + def test_build_sh_installs_pillow_with_playwright(self): + text = (ROOT / "chv-guest" / "build.sh").read_text() + self.assertIn("playwright pillow", text) + # 两分支（--break-system-packages 与回落）都加 pillow。 + self.assertEqual(text.count("playwright pillow"), 2) + + def test_buildinfo_declares_pillow(self): + self.assertIn("pillow", BUILDINFO.BASE_CONTENTS) + + def test_lockfile_still_declares_browser_script(self): + # chv-browser.py 改动仍随 repo_commit 登记（阶段 3 做法）。 + for mod in (CHECK, GENERATE): + self.assertEqual(mod.DECLARED_SOURCES["chv-browser"], + "repo_commit") + + def test_lockfile_roundtrip_covers_pillow_via_pip(self): + with tempfile.TemporaryDirectory() as tmp: + info = { + "tag": "chv-guest-24.04-0.164.0-alpha.9-test123", + "codex_version": "0.164.0-alpha.9", + "repo_commit": "test123", + "gt_version": "1.0.0", + "image_sha256": "a" * 64, + "chromium_pinned_rev": None, + "contents": list(BUILDINFO.BASE_CONTENTS), + "versions": { + "dpkg": ["git 1:2.43.0-1ubuntu7", + "ca-certificates 20240203", + "cloud-init 24.1.3-0ubuntu3", + "curl 8.5.0-2ubuntu10", + "fontconfig 2.15.0-1.1ubuntu6", + "fonts-wqy-microhei 0.2.0-beta-3.1", + "openssh-client 1:9.6p1-3ubuntu13", + "openssh-server 1:9.6p1-3ubuntu13", + "python3 3.12.3-0ubuntu2", + "python3-pip 24.0+dfsg-1ubuntu1", + "systemd-container 255.4-1ubuntu8", + "xz-utils 5.6.1+really5.4.5-1build0.1"], + "pip": ["playwright==1.49.1", + "pillow==10.2.0"], + "gt": ["gt version 1.0.0"], + "chromium": ["chromium_headless_shell-1248"], + }, + } + buildinfo_path = os.path.join(tmp, "BUILDINFO.json") + with open(buildinfo_path, "w") as fh: + json.dump(info, fh) + lock_path = os.path.join(tmp, "lockfile.json") + self.assertEqual( + GENERATE.generate(buildinfo_path, lock_path), 0) + with open(lock_path) as fh: + lock = json.load(fh) + self.assertEqual(lock["unresolved"], []) + self.assertIn("pillow 10.2.0", + lock["versions"]["pip"]) + self.assertEqual( + GENERATE.verify(lock_path, buildinfo_path), 0) + self.assertEqual( + CHECK.main([sys.argv[0], buildinfo_path]), 0) + + +if __name__ == "__main__": + unittest.main()