okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

dawnextensions: add cdp browser tool7de30f7

main
branchesmainall refs →
3.1 KiB79 linesraw
1"""Explicit session recovery; only acknowledged declarative settings are retained.
2
3This is NOT a command log. Navigation, input, JavaScript, permissions, scripts,
4request interception and event subscriptions must never be replayed here.
5"""
6
7from __future__ import annotations
8
9import copy
10import json
11from typing import TYPE_CHECKING
12
13from .errors import ConnectionLost
14from .transport import CDPSession, JSON
15
16if TYPE_CHECKING:
17    from .page import Page
18
19# A closed list, not "anything beginning with Emulation.set". Each entry has
20# replacement semantics. Clear commands share a key with their matching setter.
21SETTINGS = {
22    "Emulation.setDeviceMetricsOverride": "metrics",
23    "Emulation.clearDeviceMetricsOverride": "metrics",
24    "Emulation.setTouchEmulationEnabled": "touch",
25    "Emulation.setEmulatedMedia": "media",
26    "Emulation.setTimezoneOverride": "timezone",
27    "Emulation.setLocaleOverride": "locale",
28    "Emulation.setUserAgentOverride": "user_agent",
29    "Network.setUserAgentOverride": "user_agent",
30    "Emulation.setGeolocationOverride": "geolocation",
31    "Emulation.clearGeolocationOverride": "geolocation",
32    "Emulation.setCPUThrottlingRate": "cpu",
33    "Emulation.setDefaultBackgroundColorOverride": "background",
34    "Emulation.setEmulatedVisionDeficiency": "vision",
35}
36MAX_SETTINGS_BYTES = 256 * 1024
37
38
39class PageCDPSession(CDPSession):
40    """Raw CDP plus retention of the documented root-page emulation setters.
41
42    Sending still happens exactly once. Settings are saved only after success.
43    A saved CDPSession is tied to its original attachment, never silently rebound.
44    """
45
46    def __init__(self, page: Page, session_id: str) -> None:
47        super().__init__(page.browser._connection, session_id)
48        self.page = page
49
50    async def send(
51        self, method: str, params: JSON | None = None, *, timeout: float | None = None
52    ) -> JSON:
53        if self is not self.page.cdp:
54            raise ConnectionLost(
55                "This CDP session was replaced; use page.cdp after reattachment."
56            )
57        if method not in SETTINGS:
58            return await super().send(method, params, timeout=timeout)
59        # Serialize configuration and reattachment, so a late acknowledgement
60        # cannot overwrite a newer intended setting. Copy before the first await.
61        payload = copy.deepcopy(params or {})
62        async with self.page._settings_lock:
63            self.page._check()
64            if self is not self.page.cdp:
65                raise ConnectionLost("This CDP session was replaced; use page.cdp.")
66            candidate = dict(self.page._settings)
67            key = SETTINGS[method]
68            candidate.pop(key, None)
69            candidate[key] = {"method": method, "params": payload}
70            if (
71                len(json.dumps(candidate, allow_nan=False).encode())
72                > MAX_SETTINGS_BYTES
73            ):
74                raise ValueError(
75                    "Remembered emulation settings exceed 256 KiB; command was not sent"
76                )
77            result = await super().send(method, payload, timeout=timeout)
78            self.page._settings = candidate
79            return result