thebadking/obscura
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bracket paired screenshots at capture state

Browse Source
This commit is contained in:
SGavrl committed 2026-07-31 08:59:53 +00:00
1 parent c69fafae21
commit 42d3c1191b
4 files changed
+479 -80
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/obscura-cli/src/main.rs
+87 -23
View File
@@ -636,6 +636,33 @@ async fn run_fetch(
eprintln!("Fetching {}...", url_str);
}
// The paired corpus opts into a truthful capture boundary: its read-only
// evaluation runs after all settle passes and the final scroll reassert,
// immediately before screenshot paint. Ordinary CLI evaluation retains
// its existing evaluate-then-settle behavior when this private variable is
// absent.
let eval_at_capture_boundary = screenshot.is_some()
&& eval.is_some()
&& std::env::var("OBSCURA_SHOT_EVAL_AT_CAPTURE")
.is_ok_and(|value| value == "1");
let controlled_scroll_request = screenshot.as_ref().and_then(|_| {
let raw_y = std::env::var("OBSCURA_SHOT_SCROLL_Y").ok()?;
let x = match std::env::var("OBSCURA_SHOT_SCROLL_X") {
Ok(raw) => raw.parse::<f64>().ok().filter(|value| value.is_finite())?,
Err(_) => 0.0,
};
let requested_y = if raw_y.eq_ignore_ascii_case("bottom") {
"document.documentElement.scrollHeight".to_string()
} else {
raw_y
.parse::<f64>()
.ok()
.filter(|value| value.is_finite())
.map(|value| value.to_string())?
};
Some((x, requested_y))
});
// Process-level hard deadline. A synchronous hang inside a Rust op invoked
// from page JS cannot be cancelled by tokio (there is no await to interrupt)
// nor by the V8 watchdog (terminate_execution only unwinds JS bytecode, not
@@ -644,7 +671,9 @@ async fn run_fetch(
// operation overruns navigation + every configured settle pass + grace. A
// normal fetch returns first and the process exits before this fires.
{
let settle_passes = if eval.is_some()
let settle_passes = if eval_at_capture_boundary {
1 + u64::from(controlled_scroll_request.is_some())
} else if eval.is_some()
&& (screenshot.is_some() || selector.is_some() || dump_specified)
{
2
@@ -683,6 +712,29 @@ async fn run_fetch(
page.settle(wait_secs.saturating_mul(1000)).await;
let mut deferred_eval_output = None;
let initial_controlled_scroll = if eval_at_capture_boundary {
controlled_scroll_request.as_ref().map(|(x, requested_y)| {
page.evaluate(&format!(
"(()=>{{\
const requestedX={x},requestedY={requested_y};\
const preInitial={{x:window.scrollX,y:window.scrollY}};\
window.scrollTo(requestedX,requestedY);\
return {{requested:{{x:requestedX,y:requestedY}},\
preInitialActual:preInitial,\
postInitialActual:{{x:window.scrollX,y:window.scrollY}},\
initialBehavior:'authored',\
initialPhase:'before-controlled-scroll-settle'}}\
}})()"
))
})
} else {
None
};
if initial_controlled_scroll.is_some() {
page.settle(wait_secs.saturating_mul(1000)).await;
}
if !eval_at_capture_boundary {
if let Some(ref expr) = eval {
// Bound the eval by the same budget as navigation so a runaway
// expression (infinite loop, never-settling sync work) cannot hang.
@@ -714,6 +766,7 @@ async fn run_fetch(
// of returning the still-pending eval value (issue #248).
page.settle(wait_secs.saturating_mul(1000)).await;
}
}
if let Some(ref sel) = selector {
let found = wait_for_selector(&mut page, sel, wait_secs).await;
@@ -735,26 +788,10 @@ async fn run_fetch(
// into one instant reassertion at the actual capture boundary.
// Ordinary CLI screenshots are unchanged when these private
// capture-environment variables are absent.
let controlled_scroll = std::env::var("OBSCURA_SHOT_SCROLL_Y")
.ok()
.and_then(|raw_y| {
let x = match std::env::var("OBSCURA_SHOT_SCROLL_X") {
Ok(raw) => raw
.parse::<f64>()
.ok()
.filter(|value| value.is_finite())?,
Err(_) => 0.0,
};
let requested_y = if raw_y.eq_ignore_ascii_case("bottom") {
"document.documentElement.scrollHeight".to_string()
} else {
raw_y
.parse::<f64>()
.ok()
.filter(|value| value.is_finite())
.map(|value| value.to_string())?
};
Some(page.evaluate(&format!(
let controlled_scroll = controlled_scroll_request
.as_ref()
.map(|(x, requested_y)| {
page.evaluate(&format!(
"(()=>{{\
const requestedX={x},requestedY={requested_y};\
const preReassert={{x:window.scrollX,y:window.scrollY}};\
@@ -771,8 +808,15 @@ async fn run_fetch(
behavior:'instant',\
phase:'immediately-before-capture-state-and-screenshot'}}\
}})()"
)))
))
});
if eval_at_capture_boundary {
if let Some(ref expr) = eval {
deferred_eval_output = Some(
page.evaluate_with_timeout(expr, Duration::from_secs(timeout_secs))
);
}
}
let capture_state = deferred_eval_output.as_ref().map(|_| {
page.evaluate(
"(()=>({\
@@ -796,11 +840,31 @@ async fn run_fetch(
// after the post-eval settle so automation can record the exact
// live viewport that was painted.
if let Some(result) = deferred_eval_output {
let mut controlled_scroll_report = controlled_scroll;
if let (Some(report), Some(initial)) = (
controlled_scroll_report.as_mut(),
initial_controlled_scroll.as_ref(),
) {
if let (Some(report), Some(initial)) =
(report.as_object_mut(), initial.as_object())
{
for key in [
"preInitialActual",
"postInitialActual",
"initialBehavior",
"initialPhase",
] {
if let Some(value) = initial.get(key) {
report.insert(key.to_string(), value.clone());
}
}
}
}
println!(
"{}",
serde_json::json!({
"evaluation": result,
"controlledScroll": controlled_scroll,
"controlledScroll": controlled_scroll_report,
"captureState": capture_state.unwrap_or(serde_json::Value::Null),
})
);
crates/obscura-cli/tests/screenshot_after_eval.rs
+65
View File
@@ -126,3 +126,68 @@ fn paired_capture_reasserts_scroll_after_the_post_eval_settle() {
std::fs::remove_dir_all(directory).expect("remove temporary output directory");
}
#[test]
fn paired_capture_evaluates_state_after_final_scroll_reassert() {
let directory = output_directory();
std::fs::create_dir_all(&directory).expect("temporary output directory");
let screenshot = directory.join("capture-boundary.png");
let url = concat!(
"data:text/html,<html style=\"margin:0;scroll-behavior:smooth\">",
"<body style=\"margin:0\"><div style=\"height:400px\"></div>",
"</body></html>"
);
let output = Command::new(env!("CARGO_BIN_EXE_obscura"))
.args(["fetch", url, "--screenshot"])
.arg(&screenshot)
.args([
"--eval",
"JSON.stringify({phase:'capture-boundary-before-screenshot',y:scrollY})",
"--wait",
"0",
"--timeout",
"5",
"--quiet",
])
.env("OBSCURA_SHOT_W", "100")
.env("OBSCURA_SHOT_H", "80")
.env("OBSCURA_SHOT_SCROLL_X", "0")
.env("OBSCURA_SHOT_SCROLL_Y", "80")
.env("OBSCURA_SHOT_EVAL_AT_CAPTURE", "1")
.output()
.expect("run obscura fetch");
assert!(
output.status.success(),
"capture failed: {}",
String::from_utf8_lossy(&output.stderr)
);
let report: serde_json::Value =
serde_json::from_slice(&output.stdout).expect("capture report JSON");
let evaluation: serde_json::Value = serde_json::from_str(
report["evaluation"]
.as_str()
.expect("stringified boundary state"),
)
.expect("boundary state JSON");
assert_eq!(
evaluation["phase"].as_str(),
Some("capture-boundary-before-screenshot")
);
assert_eq!(
evaluation["y"].as_f64(),
report["captureState"]["scrollY"].as_f64(),
"state and screenshot captureState must observe the same final reassert"
);
assert_eq!(
report["controlledScroll"]["initialPhase"].as_str(),
Some("before-controlled-scroll-settle")
);
assert_eq!(
report["controlledScroll"]["phase"].as_str(),
Some("immediately-before-capture-state-and-screenshot")
);
std::fs::remove_dir_all(directory).expect("remove temporary output directory");
}
render-repros/paired-corpus.py
+203 -50
View File
@@ -63,6 +63,7 @@ BRAND_PERMUTATIONS = [
[2, 1, 0],
]
GEOMETRY_PROBE_RECT_LIMIT = 200
CAPTURE_BOUNDARY_PHASE = "capture-boundary-before-screenshot"
def geometry_probe_javascript(selectors_expression):
@@ -216,8 +217,18 @@ def reassert_chromium_controlled_scroll(page, scroll):
)
def obscura_state_eval_expression(scroll, geometry_selectors=None):
"""Sample the live page after an optional scroll and immediately before paint."""
def obscura_state_eval_expression(
scroll,
geometry_selectors=None,
sampled_phase="before-cli-post-eval-settle",
):
"""Build a page-state sample with an explicit phase label.
`scroll` remains for compatibility with focused tests and ordinary callers.
Paired captures perform their authored scroll before the second settle in
the CLI, then invoke this expression without scrolling at the final capture
boundary.
"""
scroll_script = ""
requested = "null"
controlled_scroll_result = "null"
@@ -269,7 +280,7 @@ def obscura_state_eval_expression(scroll, geometry_selectors=None):
"return ('00000000'+(h>>>0).toString(16)).slice(-8)};"
+ geometry_setup
+ "return JSON.stringify({"
"sampled_phase:'before-cli-post-eval-settle',"
f"sampled_phase:{json.dumps(sampled_phase)},"
+ geometry_result
+ f"requested:{requested},"
+ f"controlled_scroll:{controlled_scroll_result},"
@@ -347,7 +358,10 @@ def parse_obscura_capture_report(stdout):
if isinstance(report.get("controlledScroll"), dict):
state["controlled_scroll_capture"] = report["controlledScroll"]
state["evaluation_sampled_phase"] = state.get("sampled_phase")
state["sampled_phase"] = "immediately-before-screenshot"
state["screenshot_sampled_phase"] = CAPTURE_BOUNDARY_PHASE
state["state_and_screenshot_share_capture_boundary"] = (
state.get("sampled_phase") == CAPTURE_BOUNDARY_PHASE
)
return state
return None
@@ -366,8 +380,14 @@ def parse_obscura_scroll_report(stdout):
or state.get("requested")
),
"pre_reassert_actual": capture_controlled.get("preReassertActual"),
"pre_initial_actual": controlled.get("pre_initial_actual"),
"post_initial_actual": controlled.get("post_initial_actual"),
"pre_initial_actual": (
capture_controlled.get("preInitialActual")
or controlled.get("pre_initial_actual")
),
"post_initial_actual": (
capture_controlled.get("postInitialActual")
or controlled.get("post_initial_actual")
),
"final_reassert_actual": capture_controlled.get(
"finalReassertActual"
),
@@ -407,6 +427,10 @@ def obscura_environment(width, height):
# randomized platform changes responsive content and font selection,
# making a renderer comparison answer the wrong question.
OBSCURA_PROFILE=str(CANONICAL_OBSCURA_PROFILE),
# Run the paired corpus's read-only state/selector evaluation only after
# all settle phases and the final scroll reassertion. No event-loop
# pumping occurs between that evaluation and screenshot paint.
OBSCURA_SHOT_EVAL_AT_CAPTURE="1",
)
return env
@@ -557,7 +581,11 @@ def capture_obscura(
"--wait",
f"{settle_ms / 1000:g}",
]
state_expression = obscura_state_eval_expression(scroll, geometry_selectors)
state_expression = obscura_state_eval_expression(
None,
geometry_selectors,
sampled_phase=CAPTURE_BOUNDARY_PHASE,
)
command.extend(["--eval", state_expression])
started = time.time()
try:
@@ -650,15 +678,15 @@ def chromium_identity_override(session):
def capture_chromium_state(page, geometry_selectors=None):
"""Sample page provenance immediately before the screenshot."""
expression = """async () => {
async function sha256(value) {
if (!globalThis.crypto || !crypto.subtle) return null;
const bytes = new TextEncoder().encode(value);
const digest = await crypto.subtle.digest("SHA-256", bytes);
return Array.from(new Uint8Array(digest), byte =>
byte.toString(16).padStart(2, "0")).join("");
}
"""Synchronously sample page provenance immediately before screenshot.
Keep the in-page evaluation free of promises and event-loop yields. Hashing
is intentionally performed on the host after the synchronous snapshot
returns; awaiting WebCrypto here would let page tasks mutate layout between
the geometry sample and screenshot paint while falsely labeling both as one
capture boundary.
"""
expression = """() => {
function fnv1a32(value) {
let hash = 2166136261;
for (let index = 0; index < value.length; index++) {
@@ -695,7 +723,12 @@ def capture_chromium_state(page, geometry_selectors=None):
matchMedia("(prefers-reduced-motion: reduce)").matches
};
return {
sampled_phase: "immediately-before-screenshot",
_hash_sources: {
dom,
normalized_dom: normalizedDom,
visible_text: normalizedText
},
sampled_phase: __CAPTURE_BOUNDARY_PHASE__,
url: location.href,
identity: {
user_agent: navigator.userAgent,
@@ -715,16 +748,13 @@ def capture_chromium_state(page, geometry_selectors=None):
outer_html_utf16: dom.length,
outer_html_fnv1a32: fnv1a32(dom),
outer_html_bytes: new TextEncoder().encode(dom).length,
outer_html_sha256: await sha256(dom),
normalized_outer_html_utf16: normalizedDom.length,
normalized_outer_html_fnv1a32: fnv1a32(normalizedDom),
normalized_outer_html_bytes:
new TextEncoder().encode(normalizedDom).length,
normalized_outer_html_sha256: await sha256(normalizedDom),
visible_text_utf16: normalizedText.length,
visible_text_fnv1a32: fnv1a32(normalizedText),
visible_text_bytes: new TextEncoder().encode(normalizedText).length,
visible_text_sha256: await sha256(normalizedText),
},
geometry: {
inner_width: innerWidth,
@@ -776,9 +806,12 @@ def capture_chromium_state(page, geometry_selectors=None):
}
};
}"""
expression = expression.replace(
"__CAPTURE_BOUNDARY_PHASE__", json.dumps(CAPTURE_BOUNDARY_PHASE), 1
)
if geometry_selectors:
expression = expression.replace(
"async () => {", "async geometrySelectors => {", 1
"() => {", "geometrySelectors => {", 1
)
expression = expression.replace(
" const root = document.documentElement;",
@@ -787,41 +820,135 @@ def capture_chromium_state(page, geometry_selectors=None):
1,
)
expression = expression.replace(
' return {\n sampled_phase: "immediately-before-screenshot",',
" return {\n _hash_sources:",
' return {\n geometry_probes: geometryProbes,\n'
' sampled_phase: "immediately-before-screenshot",',
" _hash_sources:",
1,
)
state = page.evaluate(expression, list(geometry_selectors))
else:
state = page.evaluate(expression)
return state
def finalize_chromium_state_hashes(state):
"""Finalize hashes after paint without touching the live Chromium page."""
document_state = state["document"]
if (
document_state["outer_html_sha256"] is None
or document_state["visible_text_sha256"] is None
):
# crypto.subtle is unavailable on some non-secure/file origins. Only
# transfer the potentially large strings on that fallback path.
fallback = page.evaluate(
"""() => {
const dom = document.documentElement
? document.documentElement.outerHTML
: "";
const text = document.body
? document.body.innerText.replace(/\\s+/g, " ").trim()
: "";
return {dom, text};
}"""
)
hash_sources = state.pop("_hash_sources", None)
if hash_sources is not None:
document_state["outer_html_sha256"] = hashlib.sha256(
fallback["dom"].encode()
hash_sources["dom"].encode()
).hexdigest()
document_state["normalized_outer_html_sha256"] = hashlib.sha256(
hash_sources["normalized_dom"].encode()
).hexdigest()
document_state["visible_text_sha256"] = hashlib.sha256(
fallback["text"].encode()
hash_sources["visible_text"].encode()
).hexdigest()
return state
def chromium_capture_signature(state):
"""Return the capture-critical state used to bracket a Chromium PNG.
Full DOM hashes remain useful provenance but are intentionally excluded:
analytics nodes can churn without affecting paint. The signature focuses
on viewport/content geometry, resource readiness, and every requested
geometry probe.
"""
document = state.get("document") or {}
geometry = state.get("geometry") or {}
fonts = state.get("fonts") or {}
images = state.get("images") or {}
probes = []
for probe in state.get("geometry_probes") or []:
probes.append(
{
"selector": probe.get("selector"),
"valid": probe.get("valid"),
"count": probe.get("count"),
"rects_truncated": probe.get("rects_truncated"),
"rects": [
{
key: rect.get(key)
for key in (
"x",
"y",
"width",
"height",
"visible",
"client_rect_count",
)
}
for rect in probe.get("rects") or []
],
}
)
return {
"document": {
key: document.get(key)
for key in ("ready_state", "element_count")
},
"geometry": {
key: geometry.get(key)
for key in (
"inner_width",
"inner_height",
"scroll_x",
"scroll_y",
"device_pixel_ratio",
"document_client_width",
"document_client_height",
"document_scroll_width",
"document_scroll_height",
"body_client_width",
"body_client_height",
"body_scroll_width",
"body_scroll_height",
"visual_viewport",
)
},
"fonts": {
key: fonts.get(key)
for key in ("supported", "status", "face_count", "ready_at_sample")
},
"images": {
key: images.get(key)
for key in (
"total",
"complete",
"complete_with_pixels",
"complete_without_pixels",
"pending",
"lazy",
)
},
"geometry_probes": probes,
}
def capture_chromium_image(page, screenshot, geometry_selectors=None):
"""Bracket one Chromium PNG with synchronous capture-critical samples."""
state = capture_chromium_state(page, geometry_selectors)
page.screenshot(
path=str(screenshot),
full_page=False,
timeout=50000,
)
post_capture_state = capture_chromium_state(page, geometry_selectors)
before_signature = chromium_capture_signature(state)
after_signature = chromium_capture_signature(post_capture_state)
post_capture_state.pop("_hash_sources", None)
boundary = {
"bracketed": True,
"stable": before_signature == after_signature,
"before": before_signature,
"after": after_signature,
}
finalize_chromium_state_hashes(state)
return state, boundary
def freeze_chromium_animations(page, sample_ms):
"""Pause currently exposed Web Animations at one explicit local time.
@@ -1104,7 +1231,8 @@ def main():
"after load and once after the initial scroll. Both engines record "
"the resulting anchored offset and then reassert the requested "
"offset with instant behavior immediately before capture state "
"and paint"
"and paint. Neither engine performs another settle between state/"
"selector sampling and screenshot paint"
),
"animation_sampling": (
{
@@ -1146,11 +1274,19 @@ def main():
),
},
"state_observability": {
"chromium": "same page, sampled immediately before screenshot",
"capture_boundary_phase": CAPTURE_BOUNDARY_PHASE,
"chromium": (
"same page, synchronously sampled after all settle phases and "
"immediately before screenshot. A second synchronous sample "
"after paint brackets the PNG and reports whether capture-"
"critical geometry/readiness remained stable. DOM SHA-256 "
"finalization runs on the host only after paint"
),
"obscura": (
"DOM/resource probes are sampled by the eval that begins the "
"post-eval settle; the CLI captureState is sampled afterward, "
"following the final instant scroll reassert, and records the "
"the paired-capture CLI defers DOM/resource/selector evaluation "
"until after all settle phases and the final instant scroll "
"reassert. Evaluation, captureState, and screenshot then run "
"consecutively without another settle; captureState records the "
"exact shared PreparedRender viewport, scroll offset, and "
"content size used by paint"
),
@@ -1312,6 +1448,7 @@ def main():
chrome_started = time.time()
chromium_scroll_state = None
chromium_capture_boundary = None
try:
page.goto(url, wait_until="load", timeout=50000)
page.wait_for_timeout(args.settle_ms)
@@ -1340,8 +1477,18 @@ def main():
)
chromium_state_ok = False
try:
chromium_state = capture_chromium_state(
page, args.geometry_selector
(
chromium_state,
chromium_capture_boundary,
) = capture_chromium_image(
page,
chrome_path,
args.geometry_selector,
)
if not chromium_capture_boundary["stable"]:
chrome_messages.append(
"capture boundary warning: capture-critical "
"state changed while taking the Chromium PNG"
)
if chromium_scroll_state is not None:
chromium_state["controlled_scroll"] = (
@@ -1362,8 +1509,11 @@ def main():
except Exception as error:
chromium_state = None
chrome_messages.append(f"state capture error: {error}")
if not chrome_path.is_file():
page.screenshot(
path=str(chrome_path), full_page=False, timeout=50000
path=str(chrome_path),
full_page=False,
timeout=50000,
)
chrome_ok = (
chromium_state_ok
@@ -1382,6 +1532,9 @@ def main():
"elapsed_s": round(time.time() - chrome_started, 3),
"title": page.title() if chrome_ok else None,
"state": chromium_state if chrome_ok else None,
"capture_boundary_validation": (
chromium_capture_boundary if chrome_ok else None
),
"scroll_state": (
chromium_scroll_state if chrome_ok else None
),
render-repros/test_paired_corpus.py
+124 -7
View File
@@ -1,9 +1,11 @@
#!/usr/bin/env python3
import hashlib
import importlib.util
import json
import unittest
from pathlib import Path
from unittest import mock
SCRIPT = Path(__file__).with_name("paired-corpus.py")
@@ -75,12 +77,14 @@ class ControlledScrollTests(unittest.TestCase):
def test_capture_report_uses_post_settle_state(self):
stdout = (
"diagnostic\n"
'{"evaluation":"{\\"requested\\":{\\"x\\":0,\\"y\\":2702},'
'\\"controlled_scroll\\":{\\"requested\\":{\\"x\\":0,\\"y\\":2702},'
'\\"pre_initial_actual\\":{\\"x\\":0,\\"y\\":0},'
'\\"post_initial_actual\\":{\\"x\\":0,\\"y\\":12},'
'\\"initial_behavior\\":\\"authored\\"}}",'
'{"evaluation":"{\\"sampled_phase\\":'
'\\"capture-boundary-before-screenshot\\",'
'\\"requested\\":null,\\"controlled_scroll\\":null}",'
'"controlledScroll":{"requested":{"x":0,"y":2702},'
'"preInitialActual":{"x":0,"y":0},'
'"postInitialActual":{"x":0,"y":12},'
'"initialBehavior":"authored",'
'"initialPhase":"before-controlled-scroll-settle",'
'"preReassertActual":{"x":0,"y":1791},'
'"finalReassertActual":{"x":0,"y":1802},'
'"behavior":"instant",'
@@ -102,7 +106,9 @@ class ControlledScrollTests(unittest.TestCase):
self.assertEqual(state["actual"], {"x": 0, "y": 1802})
self.assertEqual(state["reassert_behavior"], "instant")
self.assertEqual(state["content"]["height"], 2702)
self.assertEqual(state["sampled_phase"], "immediately-before-screenshot")
self.assertEqual(
state["sampled_phase"], paired_corpus.CAPTURE_BOUNDARY_PHASE
)
def test_obscura_capture_exports_final_scroll_request_to_cli(self):
environment = paired_corpus.obscura_environment(1280, 720)
@@ -111,6 +117,20 @@ class ControlledScrollTests(unittest.TestCase):
)
self.assertEqual(environment["OBSCURA_SHOT_SCROLL_X"], "12")
self.assertEqual(environment["OBSCURA_SHOT_SCROLL_Y"], "bottom")
self.assertEqual(environment["OBSCURA_SHOT_EVAL_AT_CAPTURE"], "1")
def test_paired_state_expression_is_read_only_at_capture_boundary(self):
expression = paired_corpus.obscura_state_eval_expression(
None,
[".card"],
sampled_phase=paired_corpus.CAPTURE_BOUNDARY_PHASE,
)
self.assertIn(
f'sampled_phase:"{paired_corpus.CAPTURE_BOUNDARY_PHASE}"',
expression,
)
self.assertIn("geometry_probes:geometryProbes", expression)
self.assertNotIn("window.scrollTo", expression)
def test_every_capture_expression_samples_page_state_without_scrolling(self):
expression = paired_corpus.obscura_state_eval_expression(None)
@@ -124,7 +144,9 @@ class ControlledScrollTests(unittest.TestCase):
def test_capture_report_keeps_dom_state_and_authoritative_geometry(self):
stdout = (
'{"evaluation":"{\\"document\\":{\\"ready_state\\":\\"complete\\",'
'{"evaluation":"{\\"sampled_phase\\":'
'\\"capture-boundary-before-screenshot\\",'
'\\"document\\":{\\"ready_state\\":\\"complete\\",'
'\\"element_count\\":7,\\"outer_html_fnv1a32\\":\\"12345678\\"},'
'\\"geometry\\":{\\"document_scroll_height\\":999}}",'
'"captureState":{"scrollX":3,"scrollY":40,'
@@ -135,6 +157,26 @@ class ControlledScrollTests(unittest.TestCase):
self.assertEqual(state["document"]["element_count"], 7)
self.assertEqual(state["geometry"]["scroll_y"], 40)
self.assertEqual(state["geometry"]["document_scroll_height"], 1200)
self.assertTrue(state["state_and_screenshot_share_capture_boundary"])
def test_legacy_pre_settle_report_is_not_relabelled_as_capture_state(self):
stdout = (
'{"evaluation":"{\\"sampled_phase\\":'
'\\"before-cli-post-eval-settle\\",'
'\\"document\\":{},\\"geometry\\":{}}",'
'"captureState":{"scrollX":0,"scrollY":0,'
'"innerWidth":640,"innerHeight":480,'
'"scrollWidth":640,"scrollHeight":480}}\n'
)
state = paired_corpus.parse_obscura_capture_report(stdout)
self.assertEqual(
state["sampled_phase"], "before-cli-post-eval-settle"
)
self.assertEqual(
state["screenshot_sampled_phase"],
paired_corpus.CAPTURE_BOUNDARY_PHASE,
)
self.assertFalse(state["state_and_screenshot_share_capture_boundary"])
def test_page_state_comparison_reports_provenance_and_geometry_deltas(self):
obscura = {
@@ -212,6 +254,77 @@ class GeometryProbeTests(unittest.TestCase):
self.assertEqual(args, ())
self.assertNotIn("sampleGeometrySelector", expression)
self.assertNotIn("geometry_probes", expression)
self.assertNotIn("async ", expression)
self.assertNotIn("await ", expression)
self.assertNotIn("crypto.subtle", expression)
self.assertIn(
f'sampled_phase: "{paired_corpus.CAPTURE_BOUNDARY_PHASE}"',
expression,
)
def test_chromium_snapshot_paints_before_host_hashing(self):
events = []
first_state = {
"_hash_sources": {
"dom": "<html></html>",
"normalized_dom": "<html></html>",
"visible_text": "hello",
},
"document": {},
}
second_state = {
"_hash_sources": {
"dom": "<html></html>",
"normalized_dom": "<html></html>",
"visible_text": "hello",
},
"document": {},
}
class OrderedPage:
def __init__(self):
self.states = [first_state, second_state]
def evaluate(self, expression, *args):
events.append("evaluate")
return self.states.pop(0)
def screenshot(self, **kwargs):
events.append("screenshot")
real_sha256 = hashlib.sha256
def ordered_sha256(value):
events.append("sha256")
return real_sha256(value)
with mock.patch.object(
paired_corpus.hashlib,
"sha256",
side_effect=ordered_sha256,
):
captured, boundary = paired_corpus.capture_chromium_image(
OrderedPage(), Path("/tmp/not-written.png")
)
self.assertEqual(
events,
["evaluate", "screenshot", "evaluate", "sha256", "sha256", "sha256"],
)
self.assertTrue(boundary["stable"])
self.assertNotIn("_hash_sources", captured)
self.assertEqual(
captured["document"]["outer_html_sha256"],
hashlib.sha256(b"<html></html>").hexdigest(),
)
self.assertEqual(
captured["document"]["normalized_outer_html_sha256"],
hashlib.sha256(b"<html></html>").hexdigest(),
)
self.assertEqual(
captured["document"]["visible_text_sha256"],
hashlib.sha256(b"hello").hexdigest(),
)
def test_repeatable_selectors_are_passed_safely_in_one_state_expression(self):
selectors = ["header nav a", '[data-label="a\\\"b"]', "["]
@@ -232,6 +345,10 @@ class GeometryProbeTests(unittest.TestCase):
self.assertIn("querySelectorAll(selector)", expression)
self.assertIn("catch(error)", expression)
self.assertIn("geometry_probes: geometryProbes", expression)
self.assertIn(
f'sampled_phase: "{paired_corpus.CAPTURE_BOUNDARY_PHASE}"',
expression,
)
def test_probe_comparison_reports_raw_deltas_and_invalid_errors(self):
obscura = {