feat(chv): 阶段 4 截图三级管线——a11y 角色解析 + diff_shot 逐像素裁剪 + verify_shot 全屏校验

- snapshot 纯字符串解析 a11y 角色计数/interactive/字符数/hash
  (不跑 JS、不取原始 DOM,禁 op 保持拒绝);status 带计数器
  与 verify_share
- diff_shot:PNG 全帧 Pillow 逐像素比对(单通道阈值 10),
  无变化 changed:false size 0;有变化 bbox 外扩 8px、钳制视口、
  宽高下限 32px,再 clip 截 JPEG q70;Pillow 缺失如实报错并
  指引回落 screenshot 全帧;open 不清基线(跨页局部差分靠它)
- verify_shot:全帧 JPEG q70,只用于检查点
- build.sh 与 playwright 同批 pip 装 pillow(只加 pillow);
  buildinfo BASE_CONTENTS 登记 pillow(pip 名覆盖)
- 单测 cloud/tests/test_chv_opt_stage4.py:31 项(bbox 纯函数、
  a11y 解析、计数器、Pillow 缺失回落、禁 op、lockfile 往返)
This commit is contained in:
host committed 2026-11-08 09:45:00 +00:00
1 parent 5dc75128f8
commit f723eb8a88
4 files changed
+893 -6

No files matched your search

+562
View File
@@ -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()