AX: AXIsolatedTree::updateChildren sometimes fails to update isolated subtrees when...
[WebKit-https.git] / WebDriverTests / imported / w3c / tools / webdriver / webdriver / bidi.py
1 import copy
2 import websockets
3
4 from . import client
5
6 class BidiSession(client.Session):
7     def __init__(self,
8                  host,
9                  port,
10                  url_prefix="/",
11                  capabilities=None,
12                  extension=None):
13         """
14         Add a capability of "webSocketUrl": True to enable
15         Bidirectional connection in session creation.
16         """
17         self.websocket_transport = None
18         capabilities = self._enable_websocket(capabilities)
19         super().__init__(host, port, url_prefix, capabilities, extension)
20
21     def _enable_websocket(self, caps):
22         if caps:
23             caps.setdefault("alwaysMatch", {}).update({"webSocketUrl": True})
24         else:
25             caps = {"alwaysMatch": {"webSocketUrl": True}}
26         return caps
27
28     def match(self, capabilities):
29         """Expensive match to see if capabilities is the same as previously
30         requested capabilities if websocket would be enabled.
31
32         :return Boolean.
33         """
34         caps = copy.deepcopy(capabilities)
35         caps = self._enable_websocket(caps)
36         return super().match(caps)
37
38     async def start(self):
39         """Start a new WebDriver Bidirectional session
40         with websocket connected.
41
42         :return: Dictionary with `capabilities` and `sessionId`.
43         """
44         value = super().start()
45
46         if not self.websocket_transport or not self.websocket_transport.open:
47             self.websocket_transport = await websockets.connect(self.capabilities["webSocketUrl"])
48         return value
49
50     async def end(self):
51         """Close websocket connection first before closing session.
52         """
53         if self.websocket_transport:
54             await self.websocket_transport.close()
55             self.websocket_transport = None
56         super().end()