okami.mom/albedo

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

dawnextensions: add cdp browser tool7de30f7

main
branchesmainall refs →
1.8 KiB58 linesraw
1"""Subscriptions are explicit scopes, separate from observation and interaction."""
2
3from __future__ import annotations
4from typing import Any
5from .transport import CDPSession, Subscription
6
7
8class EventScope:
9    def __init__(
10        self,
11        cdp: CDPSession,
12        method: str,
13        *,
14        capacity: int = 256,
15        max_bytes: int = 2 * 1024 * 1024,
16        enable: bool = True,
17    ) -> None:
18        self.cdp = cdp
19        self.method = method
20        self.capacity = capacity
21        self.max_bytes = max_bytes
22        self.enable = enable
23        self.subscription: Subscription | None = None
24
25    async def __aenter__(self) -> Subscription:
26        if self.subscription is not None:
27            raise RuntimeError("An event scope cannot be entered twice")
28        self.subscription = self.cdp.connection.subscribe(
29            self.method,
30            self.cdp.session_id,
31            capacity=self.capacity,
32            max_bytes=self.max_bytes,
33        )
34        try:
35            if self.enable:
36                domain = self.method.split(".")[0]
37                if domain not in {
38                    "Network",
39                    "Runtime",
40                    "Page",
41                    "DOM",
42                    "Accessibility",
43                    "Log",
44                    "Performance",
45                    "Security",
46                }:
47                    raise ValueError(
48                        f"Enable {domain} explicitly with CDP, then use enable=False"
49                    )
50                await self.cdp.send(f"{domain}.enable")
51            return self.subscription
52        except BaseException:
53            self.subscription.close()
54            raise
55
56    async def __aexit__(self, *exc: Any) -> None:
57        if self.subscription is not None:
58            self.subscription.close()