Add closed shadow DOM capture via CDP

percy/screenshot.py

@@ -96,6 +96,91 @@ def fetch_percy_dom():
     return response.text
 
 
+def _walk_nodes(node, closed_pairs):
+    """Walk CDP DOM tree to find closed shadow roots, skipping iframe boundaries."""
+    if "contentDocument" in node:
+        return
+    if "shadowRoots" in node:
+        for sr in node["shadowRoots"]:
+            if sr.get("shadowRootType") == "closed":
+                closed_pairs.append({
+                    "hostBackendNodeId": node["backendNodeId"],
+                    "shadowBackendNodeId": sr["backendNodeId"]
+                })
+            _walk_nodes(sr, closed_pairs)
+    if "children" in node:
+        for child in node["children"]:
+            _walk_nodes(child, closed_pairs)
+
+
+def expose_closed_shadow_roots(page):
+    """Use CDP to discover closed shadow roots and expose them to PercyDOM.serialize().
+    Closed shadow roots are inaccessible from JS (element.shadowRoot === null),
+    but CDP's DOM domain can pierce them."""
+    cdp_session = None
+    try:
+        cdp_session = page.context.new_cdp_session(page)
+
+        cdp_session.send("DOM.enable")
+        doc_result = cdp_session.send(
+            "DOM.getDocument", {"depth": -1, "pierce": True}
+        )
+        root = doc_result["root"]
+
+        closed_pairs = []
+        _walk_nodes(root, closed_pairs)
+
+        if not closed_pairs:
+            return
+
+        log(
+            f"Found {len(closed_pairs)} closed shadow root(s),"
+            " exposing via CDP",
+            lvl="debug"
+        )
+
+        weakmap_script = (
+            "() => { window.__percyClosedShadowRoots ="
+            " window.__percyClosedShadowRoots || new WeakMap(); }"
+        )
+        page.evaluate(weakmap_script)
+
+        fn_decl = (
+            "function(shadowRoot) {"
+            " window.__percyClosedShadowRoots"
+            ".set(this, shadowRoot); }"
+        )
+        for pair in closed_pairs:
+            host_id = pair["hostBackendNodeId"]
+            host_result = cdp_session.send(
+                "DOM.resolveNode", {"backendNodeId": host_id}
+            )
+            host_object_id = host_result["object"]["objectId"]
+
+            shadow_id = pair["shadowBackendNodeId"]
+            shadow_result = cdp_session.send(
+                "DOM.resolveNode", {"backendNodeId": shadow_id}
+            )
+            shadow_object_id = shadow_result["object"]["objectId"]
+
+            cdp_session.send("Runtime.callFunctionOn", {
+                "functionDeclaration": fn_decl,
+                "objectId": host_object_id,
+                "arguments": [{"objectId": shadow_object_id}]
+            })
+    except Exception as err:
+        log(
+            f"Could not expose closed shadow roots via CDP: {err}",
+            lvl="debug"
+        )
+    finally:
+        if cdp_session:  # pragma: no branch
+            try:
+                cdp_session.detach()
+            except Exception:  # pragma: no cover
+                pass
+
+
 def process_frame(page, frame, options, percy_dom_script):
     """
     Processes a single cross-origin frame to capture its snapshot and resources.
@@ -392,6 +477,10 @@ def percy_snapshot(page, name, **kwargs):
         # Inject the DOM serialization script
         percy_dom_script = fetch_percy_dom()
         page.evaluate(percy_dom_script)
+
+        # Expose closed shadow roots via CDP before serialization
+        expose_closed_shadow_roots(page)
+
         cookies = page.context.cookies()
 
         # Serialize and capture the DOM

tests/test_screenshot.py

@@ -24,6 +24,8 @@
     change_window_dimension_and_wait,
     get_serialized_dom,
     process_frame,
+    expose_closed_shadow_roots,
+    _walk_nodes,
     log
 )
 import percy.screenshot as local
@@ -1290,5 +1292,105 @@ def test_create_region_with_invalid_algorithm(self):
         self.assertEqual(result, expected_result)
 
 
+class TestClosedShadowDOM(unittest.TestCase):
+    """Tests for expose_closed_shadow_roots and _walk_nodes."""
+
+    def test_walk_nodes_finds_closed_shadow_roots(self):
+        # uses top-level _walk_nodes import
+        node = {
+            "backendNodeId": 1,
+            "shadowRoots": [
+                {"backendNodeId": 2, "shadowRootType": "closed", "children": []},
+                {"backendNodeId": 3, "shadowRootType": "open", "children": []}
+            ],
+            "children": []
+        }
+        pairs = []
+        _walk_nodes(node, pairs)
+        self.assertEqual(len(pairs), 1)
+        self.assertEqual(pairs[0]["hostBackendNodeId"], 1)
+        self.assertEqual(pairs[0]["shadowBackendNodeId"], 2)
+
+    def test_walk_nodes_skips_content_document(self):
+        # uses top-level _walk_nodes import
+        node = {
+            "backendNodeId": 1,
+            "contentDocument": {"backendNodeId": 2, "children": [
+                {"backendNodeId": 3, "shadowRoots": [
+                    {"backendNodeId": 4, "shadowRootType": "closed", "children": []}
+                ], "children": []}
+            ]},
+            "children": []
+        }
+        pairs = []
+        _walk_nodes(node, pairs)
+        self.assertEqual(len(pairs), 0)
+
+    def test_expose_non_chromium_browser(self):
+        # uses top-level expose_closed_shadow_roots import
+        page = MagicMock()
+        page.context.new_cdp_session.side_effect = Exception("Not Chromium")
+        # Should not throw
+        expose_closed_shadow_roots(page)
+
+    def test_expose_no_closed_roots(self):
+        # uses top-level expose_closed_shadow_roots import
+        page = MagicMock()
+        cdp = MagicMock()
+        page.context.new_cdp_session.return_value = cdp
+        cdp.send.side_effect = lambda method, params=None: (
+            {"root": {"backendNodeId": 1, "children": []}} if method == "DOM.getDocument" else None
+        )
+        expose_closed_shadow_roots(page)
+        cdp.detach.assert_called_once()
+        page.evaluate.assert_not_called()
+
+    def test_expose_closed_roots_found(self):
+        # uses top-level expose_closed_shadow_roots import
+        page = MagicMock()
+        cdp = MagicMock()
+        page.context.new_cdp_session.return_value = cdp
+
+        def cdp_send(method, params=None):
+            if method == "DOM.getDocument":
+                return {"root": {"backendNodeId": 1, "children": [
+                    {"backendNodeId": 10, "shadowRoots": [
+                        {"backendNodeId": 20, "shadowRootType": "closed", "children": []}
+                    ], "children": []}
+                ]}}
+            if method == "DOM.resolveNode":
+                return {"object": {"objectId": f"obj-{params['backendNodeId']}"}}
+            return None
+
+        cdp.send.side_effect = cdp_send
+        expose_closed_shadow_roots(page)
+        page.evaluate.assert_called_once()
+        cdp.detach.assert_called_once()
+
+    def test_expose_cdp_error_non_fatal(self):
+        # uses top-level expose_closed_shadow_roots import
+        page = MagicMock()
+        cdp = MagicMock()
+        page.context.new_cdp_session.return_value = cdp
+        cdp.send.side_effect = Exception("CDP failed")
+        # Should not throw
+        expose_closed_shadow_roots(page)
+        cdp.detach.assert_called_once()
+
+    def test_expose_detach_error_suppressed(self):
+        # covers lines 174-175: except Exception: pass in finally
+        # uses top-level expose_closed_shadow_roots import
+        page = MagicMock()
+        cdp = MagicMock()
+        page.context.new_cdp_session.return_value = cdp
+        cdp.send.side_effect = lambda method, params=None: (
+            {"root": {"backendNodeId": 1, "children": []}}
+            if method == "DOM.getDocument" else None
+        )
+        cdp.detach.side_effect = Exception("Detach failed")
+        # Should not throw even when detach fails
+        expose_closed_shadow_roots(page)
+
+
 if __name__ == "__main__":
     unittest.main()