path: root/testing/web-platform/tests/tools/webdriver/webdriver/client.py

# 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")