asyncactions.py (12766B)
1 # mypy: allow-untyped-defs 2 3 from typing import Any, Mapping 4 webdriver = None 5 6 7 def do_delayed_imports(): 8 global webdriver 9 import webdriver 10 11 12 def get_browsing_context_id(context): 13 """ 14 :param context: Either a string representing the browsing context id, or a 15 BiDi serialized window proxy object. In the latter case, the value is 16 extracted from the serialized object. 17 :return: The browsing context id. 18 """ 19 if isinstance(context, str): 20 return context 21 elif isinstance(context, webdriver.bidi.protocol.BidiWindow): 22 # Context can be a serialized WindowProxy. 23 return context.browsing_context 24 raise ValueError("Unexpected context type: %s" % context) 25 26 class BidiBluetoothAction: 27 def __init__(self, logger, protocol): 28 do_delayed_imports() 29 self.logger = logger 30 self.protocol = protocol 31 32 async def __call__(self, payload): 33 if "context" not in payload: 34 raise ValueError("Missing required parameter: context") 35 36 context = get_browsing_context_id(payload["context"]) 37 if isinstance(context, str): 38 pass 39 elif isinstance(context, webdriver.bidi.protocol.BidiWindow): 40 # Context can be a serialized WindowProxy. 41 context = context.browsing_context 42 else: 43 raise ValueError("Unexpected context type: %s" % context) 44 return await self.execute(context, payload) 45 46 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 47 raise NotImplementedError 48 49 class BidiBluetoothHandleRequestDevicePrompt(BidiBluetoothAction): 50 name = "bidi.bluetooth.handle_request_device_prompt" 51 52 def __init__(self, logger, protocol): 53 do_delayed_imports() 54 self.logger = logger 55 self.protocol = protocol 56 57 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 58 prompt = payload["prompt"] 59 accept = payload["accept"] 60 device = payload["device"] 61 return await self.protocol.bidi_bluetooth.handle_request_device_prompt(context, prompt, accept, device) 62 63 class BidiBluetoothSimulateAdapterAction(BidiBluetoothAction): 64 name = "bidi.bluetooth.simulate_adapter" 65 66 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 67 state = payload["state"] 68 return await self.protocol.bidi_bluetooth.simulate_adapter(context, 69 state) 70 71 class BidiBluetoothDisableSimulationAction(BidiBluetoothAction): 72 name = "bidi.bluetooth.disable_simulation" 73 74 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 75 return await self.protocol.bidi_bluetooth.disable_simulation(context) 76 77 class BidiBluetoothSimulatePreconnectedPeripheralAction(BidiBluetoothAction): 78 name = "bidi.bluetooth.simulate_preconnected_peripheral" 79 80 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 81 address = payload["address"] 82 name = payload["name"] 83 manufacturer_data = payload["manufacturerData"] 84 known_service_uuids = payload["knownServiceUuids"] 85 return await self.protocol.bidi_bluetooth.simulate_preconnected_peripheral( 86 context, address, name, manufacturer_data, known_service_uuids) 87 88 class BidiBluetoothSimulateGattConnectionResponseAction(BidiBluetoothAction): 89 name = "bidi.bluetooth.simulate_gatt_connection_response" 90 91 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 92 address = payload["address"] 93 code = payload["code"] 94 return await self.protocol.bidi_bluetooth.simulate_gatt_connection_response( 95 context, address, code) 96 97 class BidiBluetoothSimulateGattDisconnectionAction(BidiBluetoothAction): 98 name = "bidi.bluetooth.simulate_gatt_disconnection" 99 100 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 101 address = payload["address"] 102 return await self.protocol.bidi_bluetooth.simulate_gatt_disconnection( 103 context, address) 104 105 class BidiBluetoothSimulateServiceAction(BidiBluetoothAction): 106 name = "bidi.bluetooth.simulate_service" 107 108 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 109 address = payload["address"] 110 uuid = payload["uuid"] 111 type = payload["type"] 112 return await self.protocol.bidi_bluetooth.simulate_service( 113 context, address, uuid, type) 114 115 class BidiBluetoothSimulateCharacteristicAction(BidiBluetoothAction): 116 name = "bidi.bluetooth.simulate_characteristic" 117 118 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 119 address = payload["address"] 120 service_uuid = payload["serviceUuid"] 121 characteristic_uuid = payload["characteristicUuid"] 122 characteristic_properties = payload["characteristicProperties"] 123 type = payload["type"] 124 return await self.protocol.bidi_bluetooth.simulate_characteristic( 125 context, address, service_uuid, characteristic_uuid, characteristic_properties, type) 126 127 class BidiBluetoothSimulateCharacteristicResponseAction(BidiBluetoothAction): 128 name = "bidi.bluetooth.simulate_characteristic_response" 129 130 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 131 address = payload["address"] 132 service_uuid = payload["serviceUuid"] 133 characteristic_uuid = payload["characteristicUuid"] 134 type = payload["type"] 135 code = payload["code"] 136 data = payload["data"] 137 return await self.protocol.bidi_bluetooth.simulate_characteristic_response( 138 context, address, service_uuid, characteristic_uuid, type, code, data) 139 140 class BidiBluetoothSimulateDescriptorAction(BidiBluetoothAction): 141 name = "bidi.bluetooth.simulate_descriptor" 142 143 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 144 address = payload["address"] 145 service_uuid = payload["serviceUuid"] 146 characteristic_uuid = payload["characteristicUuid"] 147 descriptor_uuid = payload["descriptorUuid"] 148 type = payload["type"] 149 return await self.protocol.bidi_bluetooth.simulate_descriptor( 150 context, address, service_uuid, characteristic_uuid, descriptor_uuid, type) 151 152 class BidiBluetoothSimulateDescriptorResponseAction(BidiBluetoothAction): 153 name = "bidi.bluetooth.simulate_descriptor_response" 154 155 async def execute(self, context: str, payload: Mapping[str, Any]) -> Any: 156 address = payload["address"] 157 service_uuid = payload["serviceUuid"] 158 characteristic_uuid = payload["characteristicUuid"] 159 descriptor_uuid = payload["descriptorUuid"] 160 type = payload["type"] 161 code = payload["code"] 162 data = payload["data"] 163 return await self.protocol.bidi_bluetooth.simulate_descriptor_response( 164 context, address, service_uuid, characteristic_uuid, descriptor_uuid, type, code, data) 165 166 class BidiEmulationSetGeolocationOverrideAction: 167 name = "bidi.emulation.set_geolocation_override" 168 169 def __init__(self, logger, protocol): 170 do_delayed_imports() 171 self.logger = logger 172 self.protocol = protocol 173 174 async def __call__(self, payload): 175 if "error" in payload and "coordinates" in payload: 176 raise ValueError( 177 "Params `error` and `coordinates` are mutually exclusive") 178 179 # If `error` is present, set it. Otherwise, use `UNDEFINED`. 180 error = payload['error'] if 'error' in payload else webdriver.bidi.undefined.UNDEFINED 181 coordinates = webdriver.bidi.undefined.UNDEFINED 182 if 'coordinates' in payload: 183 coordinates = payload['coordinates'] 184 elif error is webdriver.bidi.undefined.UNDEFINED: 185 # If `error` is not present, pass `coordinates` of null. 186 coordinates = None 187 188 if "contexts" not in payload: 189 raise ValueError("Missing required parameter: contexts") 190 contexts = [] 191 for context in payload["contexts"]: 192 contexts.append(get_browsing_context_id(context)) 193 if len(contexts) == 0: 194 raise ValueError("At least one context must be provided") 195 196 return await self.protocol.bidi_emulation.set_geolocation_override( 197 coordinates, error, contexts) 198 199 200 class BidiEmulationSetLocaleOverrideAction: 201 name = "bidi.emulation.set_locale_override" 202 203 def __init__(self, logger, protocol): 204 do_delayed_imports() 205 self.logger = logger 206 self.protocol = protocol 207 208 async def __call__(self, payload): 209 locale = payload['locale'] if 'locale' in payload else None 210 211 if "contexts" not in payload: 212 raise ValueError("Missing required parameter: contexts") 213 contexts = [] 214 for context in payload["contexts"]: 215 contexts.append(get_browsing_context_id(context)) 216 if len(contexts) == 0: 217 raise ValueError("At least one context must be provided") 218 219 return await self.protocol.bidi_emulation.set_locale_override(locale, 220 contexts) 221 222 223 class BidiEmulationSetScreenOrientationOverrideAction: 224 name = "bidi.emulation.set_screen_orientation_override" 225 226 def __init__(self, logger, protocol): 227 do_delayed_imports() 228 self.logger = logger 229 self.protocol = protocol 230 231 async def __call__(self, payload): 232 screen_orientation = payload['screenOrientation'] \ 233 if 'screenOrientation' in payload \ 234 else None 235 236 if "contexts" not in payload: 237 raise ValueError("Missing required parameter: contexts") 238 contexts = [] 239 for context in payload["contexts"]: 240 contexts.append(get_browsing_context_id(context)) 241 if len(contexts) == 0: 242 raise ValueError("At least one context must be provided") 243 244 return await self.protocol.bidi_emulation.set_screen_orientation_override( 245 screen_orientation, contexts) 246 247 248 class BidiSessionSubscribeAction: 249 name = "bidi.session.subscribe" 250 251 def __init__(self, logger, protocol): 252 do_delayed_imports() 253 self.logger = logger 254 self.protocol = protocol 255 256 async def __call__(self, payload): 257 events = payload["events"] 258 contexts = None 259 if "contexts" in payload and payload["contexts"] is not None: 260 contexts = [] 261 for context in payload["contexts"]: 262 contexts.append(get_browsing_context_id(context)) 263 return await self.protocol.bidi_events.subscribe(events, contexts) 264 265 266 class BidiSessionUnsubscribeAction: 267 name = "bidi.session.unsubscribe" 268 269 def __init__(self, logger, protocol): 270 do_delayed_imports() 271 self.logger = logger 272 self.protocol = protocol 273 274 async def __call__(self, payload): 275 subscriptions = payload["subscriptions"] 276 if len(subscriptions) == 0: 277 raise ValueError("At least one subscription ID should be provided") 278 279 return await self.protocol.bidi_events.unsubscribe( 280 subscriptions=subscriptions) 281 282 283 class BidiPermissionsSetPermissionAction: 284 name = "bidi.permissions.set_permission" 285 286 def __init__(self, logger, protocol): 287 do_delayed_imports() 288 self.logger = logger 289 self.protocol = protocol 290 291 async def __call__(self, payload): 292 descriptor = payload['descriptor'] 293 state = payload['state'] 294 origin = payload['origin'] 295 embedded_origin = payload.get('embeddedOrigin') 296 return await self.protocol.bidi_permissions.set_permission(descriptor, 297 state, 298 origin, 299 embedded_origin) 300 301 302 async_actions = [ 303 BidiBluetoothHandleRequestDevicePrompt, 304 BidiBluetoothSimulateAdapterAction, 305 BidiBluetoothDisableSimulationAction, 306 BidiBluetoothSimulatePreconnectedPeripheralAction, 307 BidiBluetoothSimulateGattConnectionResponseAction, 308 BidiBluetoothSimulateGattDisconnectionAction, 309 BidiBluetoothSimulateServiceAction, 310 BidiBluetoothSimulateCharacteristicAction, 311 BidiBluetoothSimulateCharacteristicResponseAction, 312 BidiBluetoothSimulateDescriptorAction, 313 BidiBluetoothSimulateDescriptorResponseAction, 314 BidiEmulationSetGeolocationOverrideAction, 315 BidiEmulationSetLocaleOverrideAction, 316 BidiEmulationSetScreenOrientationOverrideAction, 317 BidiPermissionsSetPermissionAction, 318 BidiSessionSubscribeAction, 319 BidiSessionUnsubscribeAction, 320 BidiPermissionsSetPermissionAction, 321 BidiSessionSubscribeAction]