mirror of
https://repo.dactyloidae.xyz/Dactyloidae/UXP.git
synced 2026-08-20 11:23:07 +09:00
595 lines
16 KiB
Python
595 lines
16 KiB
Python
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
import urlparse
|
|
|
|
import error
|
|
import transport
|
|
|
|
|
|
element_key = "element-6066-11e4-a52e-4f735466cecf"
|
|
|
|
|
|
def command(func):
|
|
def inner(self, *args, **kwargs):
|
|
if hasattr(self, "session"):
|
|
session = self.session
|
|
else:
|
|
session = self
|
|
|
|
if session.session_id is None:
|
|
session.start()
|
|
assert session.session_id != None
|
|
|
|
return func(self, *args, **kwargs)
|
|
|
|
inner.__name__ = func.__name__
|
|
inner.__doc__ = func.__doc__
|
|
|
|
return inner
|
|
|
|
|
|
class Timeouts(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
self._script = 30
|
|
self._load = 0
|
|
self._implicit_wait = 0
|
|
|
|
def _set_timeouts(self, name, value):
|
|
body = {"type": name,
|
|
"ms": value * 1000}
|
|
return self.session.send_command("POST", "timeouts", body)
|
|
|
|
@property
|
|
def script(self):
|
|
return self._script
|
|
|
|
@script.setter
|
|
def script(self, value):
|
|
self._set_timeouts("script", value)
|
|
self._script = value
|
|
|
|
@property
|
|
def load(self):
|
|
return self._load
|
|
|
|
@load.setter
|
|
def set_load(self, value):
|
|
self._set_timeouts("page load", value)
|
|
self._script = value
|
|
|
|
@property
|
|
def implicit_wait(self):
|
|
return self._implicit_wait
|
|
|
|
@implicit_wait.setter
|
|
def implicit_wait(self, value):
|
|
self._set_timeouts("implicit wait", value)
|
|
self._implicit_wait = value
|
|
|
|
|
|
class ActionSequence(object):
|
|
"""API for creating and performing action sequences.
|
|
|
|
Each action method adds one or more actions to a queue. When perform()
|
|
is called, the queued actions fire in order.
|
|
|
|
May be chained together as in::
|
|
|
|
ActionSequence(session, "key", id) \
|
|
.key_down("a") \
|
|
.key_up("a") \
|
|
.perform()
|
|
"""
|
|
def __init__(self, session, action_type, input_id, pointer_params=None):
|
|
"""Represents a sequence of actions of one type for one input source.
|
|
|
|
:param session: WebDriver session.
|
|
:param action_type: Action type; may be "none", "key", or "pointer".
|
|
:param input_id: ID of input source.
|
|
:param pointer_params: Optional dictionary of pointer parameters.
|
|
"""
|
|
self.session = session
|
|
self._id = input_id
|
|
self._type = action_type
|
|
self._actions = []
|
|
self._pointer_params = pointer_params
|
|
|
|
@property
|
|
def dict(self):
|
|
d = {
|
|
"type": self._type,
|
|
"id": self._id,
|
|
"actions": self._actions,
|
|
}
|
|
if self._pointer_params is not None:
|
|
d["parameters"] = self._pointer_params
|
|
return d
|
|
|
|
@command
|
|
def perform(self):
|
|
"""Perform all queued actions."""
|
|
self.session.actions.perform([self.dict])
|
|
|
|
def _key_action(self, subtype, value):
|
|
self._actions.append({"type": subtype, "value": value})
|
|
|
|
def _pointer_action(self, subtype, button):
|
|
self._actions.append({"type": subtype, "button": button})
|
|
|
|
def pointer_move(self, x, y, duration=None, origin=None):
|
|
"""Queue a pointerMove action.
|
|
|
|
:param x: Destination x-axis coordinate of pointer in CSS pixels.
|
|
:param y: Destination y-axis coordinate of pointer in CSS pixels.
|
|
:param duration: Number of milliseconds over which to distribute the
|
|
move. If None, remote end defaults to 0.
|
|
:param origin: Origin of coordinates, either "viewport", "pointer" or
|
|
an Element. If None, remote end defaults to "viewport".
|
|
"""
|
|
# TODO change to pointerMove once geckodriver > 0.14 is available on mozilla-central
|
|
action = {
|
|
"type": "move",
|
|
"x": x,
|
|
"y": y
|
|
}
|
|
if duration is not None:
|
|
action["duration"] = duration
|
|
if origin is not None:
|
|
action["origin"] = origin if isinstance(origin, basestring) else origin.json()
|
|
self._actions.append(action)
|
|
return self
|
|
|
|
def pointer_up(self, button):
|
|
"""Queue a pointerUp action for `button`.
|
|
|
|
:param button: Pointer button to perform action with.
|
|
"""
|
|
self._pointer_action("pointerUp", button)
|
|
return self
|
|
|
|
def pointer_down(self, button):
|
|
"""Queue a pointerDown action for `button`.
|
|
|
|
:param button: Pointer button to perform action with.
|
|
"""
|
|
self._pointer_action("pointerDown", button)
|
|
return self
|
|
|
|
def key_up(self, value):
|
|
"""Queue a keyUp action for `value`.
|
|
|
|
:param value: Character to perform key action with.
|
|
"""
|
|
self._key_action("keyUp", value)
|
|
return self
|
|
|
|
def key_down(self, value):
|
|
"""Queue a keyDown action for `value`.
|
|
|
|
:param value: Character to perform key action with.
|
|
"""
|
|
self._key_action("keyDown", value)
|
|
return self
|
|
|
|
def send_keys(self, keys):
|
|
"""Queue a keyDown and keyUp action for each character in `keys`.
|
|
|
|
:param keys: String of keys to perform key actions with.
|
|
"""
|
|
for c in keys:
|
|
self.key_down(c)
|
|
self.key_up(c)
|
|
return self
|
|
|
|
|
|
class Actions(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
@command
|
|
def perform(self, actions=None):
|
|
"""Performs actions by tick from each action sequence in `actions`.
|
|
|
|
:param actions: List of input source action sequences. A single action
|
|
sequence may be created with the help of
|
|
``ActionSequence.dict``.
|
|
"""
|
|
body = {"actions": [] if actions is None else actions}
|
|
return self.session.send_command("POST", "actions", body)
|
|
|
|
@command
|
|
def release(self):
|
|
return self.session.send_command("DELETE", "actions")
|
|
|
|
def sequence(self, *args, **kwargs):
|
|
"""Return an empty ActionSequence of the designated type.
|
|
|
|
See ActionSequence for parameter list.
|
|
"""
|
|
return ActionSequence(self.session, *args, **kwargs)
|
|
|
|
class Window(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
@property
|
|
@command
|
|
def size(self):
|
|
resp = self.session.send_command("GET", "window/size")
|
|
return (resp["width"], resp["height"])
|
|
|
|
@size.setter
|
|
@command
|
|
def size(self, (width, height)):
|
|
body = {"width": width, "height": height}
|
|
self.session.send_command("POST", "window/size", body)
|
|
|
|
@property
|
|
@command
|
|
def position(self):
|
|
resp = self.session.send_command("GET", "window/position")
|
|
return (resp["x"], resp["y"])
|
|
|
|
@position.setter
|
|
@command
|
|
def position(self, (x, y)):
|
|
body = {"x": x, "y": y}
|
|
self.session.send_command("POST", "window/position", body)
|
|
|
|
@property
|
|
@command
|
|
def maximize(self):
|
|
return self.session.send_command("POST", "window/maximize")
|
|
|
|
|
|
class Find(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
@command
|
|
def css(self, selector, all=True):
|
|
return self._find_element("css selector", selector, all)
|
|
|
|
def _find_element(self, strategy, selector, all):
|
|
route = "elements" if all else "element"
|
|
|
|
body = {"using": strategy,
|
|
"value": selector}
|
|
|
|
data = self.session.send_command("POST", route, body, key="value")
|
|
|
|
if all:
|
|
rv = [self.session._element(item) for item in data]
|
|
else:
|
|
rv = self.session._element(data)
|
|
|
|
return rv
|
|
|
|
|
|
class Cookies(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
def __getitem__(self, name):
|
|
self.session.send_command("GET", "cookie/%s" % name, {}, key="value")
|
|
|
|
def __setitem__(self, name, value):
|
|
cookie = {"name": name,
|
|
"value": None}
|
|
|
|
if isinstance(name, (str, unicode)):
|
|
cookie["value"] = value
|
|
elif hasattr(value, "value"):
|
|
cookie["value"] = value.value
|
|
self.session.send_command("POST", "cookie/%s" % name, {}, key="value")
|
|
|
|
|
|
class UserPrompt(object):
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
@command
|
|
def dismiss(self):
|
|
self.session.send_command("POST", "alert/dismiss")
|
|
|
|
@command
|
|
def accept(self):
|
|
self.session.send_command("POST", "alert/accept")
|
|
|
|
@property
|
|
@command
|
|
def text(self):
|
|
return self.session.send_command("GET", "alert/text", key="value")
|
|
|
|
@text.setter
|
|
@command
|
|
def text(self, value):
|
|
body = {"value": list(value)}
|
|
self.session.send_command("POST", "alert/text", body=body)
|
|
|
|
|
|
class Session(object):
|
|
def __init__(self, host, port, url_prefix="/", desired_capabilities=None,
|
|
required_capabilities=None, timeout=transport.HTTP_TIMEOUT,
|
|
extension=None):
|
|
self.transport = transport.HTTPWireProtocol(
|
|
host, port, url_prefix, timeout=timeout)
|
|
self.desired_capabilities = desired_capabilities
|
|
self.required_capabilities = required_capabilities
|
|
self.session_id = None
|
|
self.timeouts = None
|
|
self.window = None
|
|
self.find = None
|
|
self._element_cache = {}
|
|
self.extension = None
|
|
self.extension_cls = extension
|
|
|
|
self.timeouts = Timeouts(self)
|
|
self.window = Window(self)
|
|
self.find = Find(self)
|
|
self.alert = UserPrompt(self)
|
|
self.actions = Actions(self)
|
|
|
|
def __enter__(self):
|
|
self.start()
|
|
return self
|
|
|
|
def __exit__(self, *args, **kwargs):
|
|
self.end()
|
|
|
|
def __del__(self):
|
|
self.end()
|
|
|
|
def start(self):
|
|
if self.session_id is not None:
|
|
return
|
|
|
|
body = {}
|
|
|
|
caps = {}
|
|
if self.desired_capabilities is not None:
|
|
caps["desiredCapabilities"] = self.desired_capabilities
|
|
if self.required_capabilities is not None:
|
|
caps["requiredCapabilities"] = self.required_capabilities
|
|
#body["capabilities"] = caps
|
|
body = caps
|
|
|
|
resp = self.transport.send("POST", "session", body=body)
|
|
self.session_id = resp["sessionId"]
|
|
|
|
if self.extension_cls:
|
|
self.extension = self.extension_cls(self)
|
|
|
|
return resp["value"]
|
|
|
|
def end(self):
|
|
if self.session_id is None:
|
|
return
|
|
|
|
url = "session/%s" % self.session_id
|
|
self.transport.send("DELETE", url)
|
|
|
|
self.session_id = None
|
|
self.timeouts = None
|
|
self.window = None
|
|
self.find = None
|
|
self.extension = None
|
|
|
|
def send_command(self, method, url, body=None, key=None):
|
|
if self.session_id is None:
|
|
raise error.SessionNotCreatedException()
|
|
url = urlparse.urljoin("session/%s/" % self.session_id, url)
|
|
return self.transport.send(method, url, body, key=key)
|
|
|
|
@property
|
|
@command
|
|
def url(self):
|
|
return self.send_command("GET", "url", key="value")
|
|
|
|
@url.setter
|
|
@command
|
|
def url(self, url):
|
|
if urlparse.urlsplit(url).netloc is None:
|
|
return self.url(url)
|
|
body = {"url": url}
|
|
return self.send_command("POST", "url", body)
|
|
|
|
@command
|
|
def back(self):
|
|
return self.send_command("POST", "back")
|
|
|
|
@command
|
|
def forward(self):
|
|
return self.send_command("POST", "forward")
|
|
|
|
@command
|
|
def refresh(self):
|
|
return self.send_command("POST", "refresh")
|
|
|
|
@property
|
|
@command
|
|
def title(self):
|
|
return self.send_command("GET", "title", key="value")
|
|
|
|
@property
|
|
@command
|
|
def window_handle(self):
|
|
return self.send_command("GET", "window_handle", key="value")
|
|
|
|
@window_handle.setter
|
|
@command
|
|
def window_handle(self, handle):
|
|
body = {"handle": handle}
|
|
return self.send_command("POST", "window", body=body)
|
|
|
|
def switch_frame(self, frame):
|
|
if frame == "parent":
|
|
url = "frame/parent"
|
|
body = None
|
|
else:
|
|
url = "frame"
|
|
if isinstance(frame, Element):
|
|
body = {"id": frame.json()}
|
|
else:
|
|
body = {"id": frame}
|
|
|
|
return self.send_command("POST", url, body)
|
|
|
|
@command
|
|
def close(self):
|
|
return self.send_command("DELETE", "window_handle")
|
|
|
|
@property
|
|
@command
|
|
def handles(self):
|
|
return self.send_command("GET", "window_handles", key="value")
|
|
|
|
@property
|
|
@command
|
|
def active_element(self):
|
|
data = self.send_command("GET", "element/active", key="value")
|
|
if data is not None:
|
|
return self._element(data)
|
|
|
|
def _element(self, data):
|
|
elem_id = data[element_key]
|
|
assert elem_id
|
|
if elem_id in self._element_cache:
|
|
return self._element_cache[elem_id]
|
|
return Element(self, elem_id)
|
|
|
|
@command
|
|
def cookies(self, name=None):
|
|
if name is None:
|
|
url = "cookie"
|
|
else:
|
|
url = "cookie/%s" % name
|
|
return self.send_command("GET", url, {}, key="value")
|
|
|
|
@command
|
|
def set_cookie(self, name, value, path=None, domain=None, secure=None, expiry=None):
|
|
body = {"name": name,
|
|
"value": value}
|
|
if path is not None:
|
|
body["path"] = path
|
|
if domain is not None:
|
|
body["domain"] = domain
|
|
if secure is not None:
|
|
body["secure"] = secure
|
|
if expiry is not None:
|
|
body["expiry"] = expiry
|
|
self.send_command("POST", "cookie", {"cookie": body})
|
|
|
|
def delete_cookie(self, name=None):
|
|
if name is None:
|
|
url = "cookie"
|
|
else:
|
|
url = "cookie/%s" % name
|
|
self.send_command("DELETE", url, {}, key="value")
|
|
|
|
#[...]
|
|
|
|
@command
|
|
def execute_script(self, script, args=None):
|
|
if args is None:
|
|
args = []
|
|
|
|
body = {
|
|
"script": script,
|
|
"args": args
|
|
}
|
|
return self.send_command("POST", "execute", body, key="value")
|
|
|
|
@command
|
|
def execute_async_script(self, script, args=None):
|
|
if args is None:
|
|
args = []
|
|
|
|
body = {
|
|
"script": script,
|
|
"args": args
|
|
}
|
|
return self.send_command("POST", "execute_async", body, key="value")
|
|
|
|
#[...]
|
|
|
|
@command
|
|
def screenshot(self):
|
|
return self.send_command("GET", "screenshot", key="value")
|
|
|
|
|
|
class Element(object):
|
|
def __init__(self, session, id):
|
|
self.session = session
|
|
self.id = id
|
|
assert id not in self.session._element_cache
|
|
self.session._element_cache[self.id] = self
|
|
|
|
def json(self):
|
|
return {element_key: self.id}
|
|
|
|
@property
|
|
def session_id(self):
|
|
return self.session.session_id
|
|
|
|
def url(self, suffix):
|
|
return "element/%s/%s" % (self.id, suffix)
|
|
|
|
@command
|
|
def find_element(self, strategy, selector):
|
|
body = {"using": strategy,
|
|
"value": selector}
|
|
|
|
elem = self.session.send_command("POST", self.url("element"), body, key="value")
|
|
return self.session.element(elem)
|
|
|
|
@command
|
|
def click(self):
|
|
self.session.send_command("POST", self.url("click"), {})
|
|
|
|
@command
|
|
def tap(self):
|
|
self.session.send_command("POST", self.url("tap"), {})
|
|
|
|
@command
|
|
def clear(self):
|
|
self.session.send_command("POST", self.url("clear"), {})
|
|
|
|
@command
|
|
def send_keys(self, keys):
|
|
if isinstance(keys, (str, unicode)):
|
|
keys = [char for char in keys]
|
|
|
|
body = {"value": keys}
|
|
|
|
return self.session.send_command("POST", self.url("value"), body)
|
|
|
|
@property
|
|
@command
|
|
def text(self):
|
|
return self.session.send_command("GET", self.url("text"), key="value")
|
|
|
|
@property
|
|
@command
|
|
def name(self):
|
|
return self.session.send_command("GET", self.url("name"), key="value")
|
|
|
|
@command
|
|
def style(self, property_name):
|
|
return self.session.send_command("GET", self.url("css/%s" % property_name), key="value")
|
|
|
|
@property
|
|
@command
|
|
def rect(self):
|
|
return self.session.send_command("GET", self.url("rect"))
|
|
|
|
@command
|
|
def property(self, name):
|
|
return self.session.send_command("GET", self.url("property/%s" % name), key="value")
|
|
|
|
@command
|
|
def attribute(self, name):
|
|
return self.session.send_command("GET", self.url("attribute/%s" % name), key="value")
|