mirror of
https://github.com/servo/servo.git
synced 2025-10-04 18:49:23 +01:00
43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
import json
|
|
|
|
import webdriver
|
|
|
|
|
|
"""WebDriver wire protocol codecs."""
|
|
|
|
|
|
class Encoder(json.JSONEncoder):
|
|
def __init__(self, *args, **kwargs):
|
|
kwargs.pop("session")
|
|
super(Encoder, self).__init__(*args, **kwargs)
|
|
|
|
def default(self, obj):
|
|
if isinstance(obj, (list, tuple)):
|
|
return [self.default(x) for x in obj]
|
|
elif isinstance(obj, webdriver.Element):
|
|
return {webdriver.Element.identifier: obj.id}
|
|
elif isinstance(obj, webdriver.Frame):
|
|
return {webdriver.Frame.identifier: obj.id}
|
|
elif isinstance(obj, webdriver.Window):
|
|
return {webdriver.Frame.identifier: obj.id}
|
|
return super(Encoder, self).default(obj)
|
|
|
|
|
|
class Decoder(json.JSONDecoder):
|
|
def __init__(self, *args, **kwargs):
|
|
self.session = kwargs.pop("session")
|
|
super(Decoder, self).__init__(
|
|
object_hook=self.object_hook, *args, **kwargs)
|
|
|
|
def object_hook(self, payload):
|
|
if isinstance(payload, (list, tuple)):
|
|
return [self.object_hook(x) for x in payload]
|
|
elif isinstance(payload, dict) and webdriver.Element.identifier in payload:
|
|
return webdriver.Element.from_json(payload, self.session)
|
|
elif isinstance(payload, dict) and webdriver.Frame.identifier in payload:
|
|
return webdriver.Frame.from_json(payload, self.session)
|
|
elif isinstance(payload, dict) and webdriver.Window.identifier in payload:
|
|
return webdriver.Window.from_json(payload, self.session)
|
|
elif isinstance(payload, dict):
|
|
return {k: self.object_hook(v) for k, v in payload.iteritems()}
|
|
return payload
|