Source code for webtest_selenium.sel

# (c) 2005 Ian Bicking and contributors; written for Paste
# (
# Licensed under the MIT license:
Routines for testing WSGI applications with selenium.

Most interesting is :class:`~webtest.sel.SeleniumApp` and the
:func:`~webtest.sel.selenium` decorator
import os
import cgi
import sys
import signal
import types
import webob
import logging
import tempfile
import unittest
import threading
import subprocess
from functools import wraps
from webtest import app as testapp
from webtest import forms
from webtest import utils
from contextlib import contextmanager
from six.moves import http_client
from six.moves import BaseHTTPServer
from six.moves import SimpleHTTPServer
from webtest import http
from six import binary_type
from six import PY3
from webtest.compat import urlencode
import json

except NameError:
    unicode = str

log = logging.getLogger(__name__)

if 'SELENIUM_VERBOSE' in os.environ:

class SeleniumWarning(Warning):
    """Specific warning category"""

HAS_UPLOAD_SUPPORT = ('*chrome', '*firefox')

# Decorator

def function_decorator(func):
    """run test with selenium. create a new session if needed"""
    def wrapper(*args):
        if is_available():
            if args and isinstance(args[0], unittest.TestCase):
                self = args[0]
                if isinstance(, SeleniumApp):
                    old_app =
           = SeleniumApp(
               = old_app
                # function
    return wrapper

def context_manager(resp):
    """A context mamanger to create a session inside a test"""
    resp.updated = False
    if not is_available():
        yield None
        test_app = resp.test_app
        app = SeleniumApp(
        for h, v in resp.request.headers.items():
            if h.lower() not in ('host',):
                app.browser.addCustomRequestHeader(h, v)
        fd = tempfile.NamedTemporaryFile(prefix='webtest-selenium-',
                                         suffix='.html', delete=False)
        response = app.get('/__file__', dict(
            yield response
            body = app.browser.getHtmlSource()
            if PY3:
                body = body.encode(resp.charset or 'utf-8')
            resp.body = body
            resp._forms_indexed = None
            resp.updated = True

[docs]def selenium(obj): """A callable usable as: - class decorator - function decorator - contextmanager """ if isinstance(obj, type): if is_available(): return obj elif isinstance(obj, types.FunctionType): return function_decorator(obj) elif isinstance(obj, testapp.TestResponse): return contextmanager(context_manager)(obj) else: raise RuntimeError('Unsuported type %r' % obj)
class Selenium(object): """Selenium RC control aka ``browser`` A object use to manipulate DOM nodes. This object allow to use the underlying selenium api. See Selenium `api <>`_ You can use the original method name:: browser.fireEvent('id=#myid", 'focus') Or a more pythonic name:: browser.fire_event('id=#myid", 'focus') Both are equal to:: browser.execute('fireEvent', 'id=#myid', 'focus') """ def __init__(self): = os.environ.get('SELENIUM_HOST', '') self.port = int(os.environ.get('SELENIUM_PORT', 4444)) self.session_id = None def start(self, url): self.driver = os.environ.get('SELENIUM_DRIVER', '*googlechrome') self.session_id = self.getNewBrowserSession( self.driver, url, '', "captureNetworkTraffic=true", "addCustomRequestHeader=true") def stop(self): self.testComplete() self.session_id = None def execute(self, cmd, *args): data = dict([(i + 1, a) for i, a in enumerate(args)], cmd=cmd) if self.session_id: data['sessionId'] = self.session_id data = urlencode(data) headers = { "Content-Type": "application/x-www-form-urlencoded; charset=utf-8" } conn = http_client.HTTPConnection(, self.port) try: conn.request("POST", "/selenium-server/driver/", data, headers) resp = conn.getresponse() data = finally: conn.close() if PY3: data = str(data, 'utf-8') if data.startswith('ERROR: Unknown command:'): raise AttributeError(repr(data)) elif not data.startswith('OK'): raise RuntimeError(repr(data)) data = data[3:] if data in ('true', 'false'): return data == 'true' and True or False return data def __getattr__(self, attr): cmd = _get_command(attr) def wrapped(*args): args = [cmd] + [str(a) for a in args] return self.execute(*args) wrapped.__name__ = attr return wrapped ############## # Webtest API ##############
[docs]class SeleniumApp(testapp.TestApp): """See :class:`webtest.TestApp` SeleniumApp only support ``GET`` requests """ apps = [] def __init__(self, app=None, url=None, timeout=30000, extra_environ=None, relative_to=None, **kwargs): = self.server = None if app: super(SeleniumApp, self).__init__(app, relative_to=relative_to) self.server = http.StopableWSGIServer.create(app) self.server.wait() url = self.server.application_url assert is_available() self.session_id = None self._browser = Selenium() self._browser.start(url) self.extra_environ = extra_environ or {} self.timeout = timeout self.test_app = self @property def browser(self): """The current :class:`~webtest.sel.Selenium`""" return self._browser @property def has_upload_support(self): return self._browser.driver in HAS_UPLOAD_SUPPORT def do_request(self, req, status, expect_errors, now=None): if req.method != 'GET': raise testapp.AppError('Only GET are allowed') if = '%s:%s' % (, self.server.adj.port) self.browser.captureNetworkTraffic('json') for h, v in req.headers.items(): if h.lower() not in ('host',): self.browser.addCustomRequestHeader(h, v) resp = self._get_response() if not expect_errors: self._check_status(status, resp) if not status: status = resp.status_int if not (status > 300 and status < 400): self._check_errors(resp) return resp def _get_response(self, resp=None, timeout=None): """Get responses responses from selenium""" if timeout != 0: timeout = timeout or self.timeout self.browser.waitForPageToLoad(timeout) json_string = self.browser.captureNetworkTraffic('json') # escape, otherwise json.loads() can raise ValueError json_string = json_string.replace('\\0', '\\\\0') trafic = json.loads(json_string) responses = [] errors = [] for d in trafic: if d['url'].endswith('.ico'): continue req = webob.Request.blank(d['url']) for h in d['requestHeaders']: req.headers[h['name']] = h['value'] resp = TestResponse() = resp.test_app = self.test_app resp.browser = self.test_app.browser resp.responses = responses resp.errors = errors resp.request = req resp.status = str(d['statusCode']) for h in d['responseHeaders']: resp.headers[h['name']] = h['value'] if resp.status_int == 200 and 'text/' in resp.content_type: if not resp.charset: resp.charset = 'utf-8' if resp.status_int > 400: errors.append('%s %r' % (resp.request.url, resp)) if 'html' in resp.content_type or resp.status_int != 200: responses.append(resp) if responses: resp = responses.pop(0) return resp elif resp is not None: return resp else: raise LookupError('No response found')
[docs] def close(self): """Close selenium and the WSGI server if needed""" if self.server: self.server.shutdown() if 'SELENIUM_KEEP_OPEN' not in os.environ: self.browser.stop() if 'SELENIUM_PID' in os.environ: os.kill(int(os.environ['SELENIUM_PID']), signal.SIGTERM)
class TestResponse(testapp.TestResponse): def follow(self, status=None, **kw): """If this request is a redirect, follow that redirect. It is an error if this is not a redirect response. Returns another response object. """ if not (self.status_int >= 300 and self.status_int < 400): raise ValueError( 'You can only follow 301 and 302. Not %s' % self.status_int) if len(self.responses): resp = self.responses[0] if not kw.get('expect_errors', False):, resp) if not status: return self.responses.pop(0) raise LookupError('Responses queue is empty. Nothing to follow.') def click(self, description=None, linkid=None, href=None, anchor=None, index=None, verbose=False, extra_environ=None, timeout=None): link =, linkid=linkid, href=href, index=index) if timeout == 0: return self return self.test_app._get_response(resp=self, timeout=timeout) def _parse_forms(self): forms_ = self._forms_indexed = {} for i, f in enumerate(self.html('form')): form_id = f.get('id') form = Form(self, form_id or i) forms_[i] = form if form_id: forms_[form_id] = form @property def form(self): return Form(self, 0) def _text__get(self): body = self.browser.getHtmlSource() print(body) if isinstance(body, binary_type): return body.decode(self.charset or 'utf-8') else: return body text = property(_text__get) def __contains__(self, s): if isinstance(s, Element): return s.isElementPresent() return s in self.text @property def doc(self): """Expose a :class:`~webtest.browser.Document`""" return Document(self) ########## # DOM ########## class Element(object): """A object use to manipulate DOM nodes. This object allow to use the underlying selenium api for the specified locator. See Selenium `api <>`_ You can use the original method name:: element.fireEvent('focus') Or a more pythonic name:: element.fire_event('focus') Both are equal to:: browser.execute('fireEvent', element.locator, 'focus') """ def __init__(self, resp, locator): self.browser = resp.browser self.resp = resp self.locator = locator def __getattr__(self, attr): cmd = _get_command(attr) def wrapped(*args): args = [cmd, self.locator] + [str(a) for a in args] return self.browser.execute(*args) wrapped.__name__ = attr return wrapped def exist(self): """return true is the element is present""" return self.isElementPresent() def wait(self, timeout=3000): """Wait for an element and return this element""" script = "selenium.isElementPresent(%r) || null" % str(self) try: self.browser.waitForCondition(script, timeout) except RuntimeError: raise RuntimeError("Can't find %s after %sms" % (self, timeout)) return self def wait_and_click(self, timeout=3000): """Wait for an element, click on it and return this element""" return self.wait(timeout).click() def hasClass(self, name): """True iif the class is present""" classes = self.attr('class').split() return name in classes def html(self): """Return the innerHTML of the element""" return self.eval('e.innerHTML') def text(self): """Return the text of the element""" return self.getText() def attr(self, attr): """Return the attribute value of the element""" return self.eval('e.getAttribute(%r)' % str(attr)) def drag_and_drop(self, element): """Drag and drop to element""" return self.dragAndDropToObject(element) def value__get(self): return self.getValue() def value__set(self, value): value = json.dumps(value) script = """(function() { s.doFireEvent(l, "focus"); s.doType(l, %s); e.setAttribute("value", %s); s.doFireEvent(l, "keydown"); s.doFireEvent(l, "keypress"); s.doFireEvent(l, "keyup"); }())""" % (value, value) self.eval(script) value = property(value__get, value__set) def eval(self, *expr): """Eval a javascript expression in Selenium RC. You can use the following variables: - s: the ``selenium`` object - b: the ``browserbot`` object - l: the element locator string - e: the element itself """ script = ( "(function(s) {" "var l = %r;" "var b = s.browserbot; var e = b.findElement(l);" "var res = %s; return res || 'null';" "}(this))" ) % (str(self), ''.join(expr).strip(';')) try: return self.browser.getEval(script) except RuntimeError: raise RuntimeError(script) def __contains__(self, s): if isinstance(s, Element): s = s.html() return s in self.html() def __nonzero__(self): return self.exist() __bool__ = __nonzero__ def __repr__(self): return '<%s at %s>' % (self.__class__.__name__, self.locator) def __str__(self): return str(self.locator.replace('"', "'")) class Document(object): """The browser document. ``resp.doc.myid`` is egual to ``resp.doc.css('#myid')``""" def __init__(self, resp): self.resp = resp def __getattr__(self, attr): return Element(self.resp, 'css=#%s' % attr) def get(self, tag, **kwargs): """Return an element matching ``tag``, an ``attribute`` and an ``index``. For example: resp.doc.get('input', name='go') => xpath=//input[@name="go"] resp.doc.get('li', description='Item') => xpath=//li[.="Item"] """ locator = _eval_xpath(tag, **kwargs) return Element(self.resp, locator) def xpath(self, path): """Get an :class:`~webtest.browser.Element` using xpath""" return Element(self.resp, 'xpath=%s' % path) def css(self, selector): """Get an :class:`~webtest.browser.Element` using a css selector""" return Element(self.resp, 'css=%s' % selector) def link(self, description=None, linkid=None, href=None, index=None): """Get a link""" return self.get('a', description=description, id=linkid, href=href, index=index) def input(self, value=None, name=None, inputid=None, index=None): """Get an input field""" return self.get('input', id=inputid, value=value, name=name, index=index) def button(self, description=None, buttonid=None, index=None): """Get a button""" return self.get('button', description=description, id=buttonid, index=index) def __contains__(self, s): if isinstance(s, Element): return s.exist() return self.browser.isTextPresent(json.dumps(s)) def __call__(self, locator): return Element(locator) ########### # Forms ########### class Field(forms.Field, Element): classes = {} def __init__(self, *args, **kwargs): super(Field, self).__init__(*args, **kwargs) self.browser = self.form.browser self.options = [] self.selectedIndices = [] self._forced_values = [] self.locator = _eval_xpath(self.tag, locator=self.form.locator, value = property(Element.value__get, Element.value__set) class Select(Field): """Field representing ``<select>``""" def force_value(self, value):'value=%s' % value) def value__set(self, value):'value=%s' % value) def value__get(self): return self.getSelectedValue() value = property(value__get, value__set) Field.classes['select'] = Select class MultipleSelect(Field): """Field representing ``<select multiple="multiple">``""" def force_value(self, values): self.removeAllSelections() str_values = [utils.stringify(value) for value in values] for v in str_values: self.addSelection('value=%s' % v) def value__set(self, values): self.removeAllSelections() str_values = [utils.stringify(value) for value in values] for v in str_values: self.addSelection('value=%s' % v) def value__get(self): value = self.getSelectedValues() return value.split(',') value = property(value__get, value__set) Field.classes['multiple_select'] = MultipleSelect class Radio(Field): """Field representing ``<input type="radio">``""" def value__set(self, value): if value: self.check() else: self.uncheck() def value__get(self): script = r"""(function(obj) { var name = '%s'; var element = obj.browserbot.findElement('%s'); var elements = element.getElementsByTagName('input'); var values = []; for (var i = 0, n = elements.length; i < n; ++i) { element = elements[i]; if ( == name && element.checked) { values.push('name='+element.value); } } return values.join('&'); }(this))""" % (, self.form.locator) value = self.browser.getEval(script) value = [v for k, v in cgi.parse_qsl('name=true')] if not value: return None elif len(value) == 1: return value[0] raise ValueError( 'Got more than one value for %r: %s' % (self, value)) value = property(value__get, value__set) Field.classes['radio'] = Radio class Checkbox(Radio): """Field representing ``<input type="checkbox">``""" Field.classes['checkbox'] = Checkbox class Text(Field): """Field representing ``<input type="text">``""" Field.classes['text'] = Text class File(Field): """Field representing ``<input type="file">``""" def _run_server(self, filename): """Run a simple server in a separate thread""" ip, port = http.get_free_port() def run(): FileHandler.filename = filename server = BaseHTTPServer.HTTPServer((ip, port), FileHandler) server.handle_request() thread = threading.Thread(target=run) thread.start() return 'http://%s:%s/' % (ip, port) def value__set(self, value): if isinstance(value, (list, tuple)) and len(value) == 1: value = [] + list(value) test_app = self.form.resp.test_app file_info = test_app._get_file_info(value) name, filename, content = file_info if test_app.has_upload_support: url = self._run_server(filename) url += os.path.basename(filename) self.attachFile(url) force_value = value__set value = property(Field.value__get, value__set) Field.classes['file'] = File class Textarea(Text): """Field representing ``<textarea>``""" Field.classes['textarea'] = Textarea class Hidden(Text, forms.Hidden): """Field representing ``<input type="hidden">``""" Field.classes['hidden'] = Hidden class Submit(Field, forms.Submit): """Field representing ``<input type="submit">`` and ``<button>``""" settable = False def value__get(self): return None value = property(value__get) def value_if_submitted(self): return self._value Field.classes['submit'] = Submit Field.classes['button'] = Submit Field.classes['image'] = Submit class Form(forms.Form, Element): """See :class:`~webtest.Form`""" FieldClass = Field def __init__(self, resp, id): self.resp = resp self.test_app = resp.test_app self.browser = resp.browser if isinstance(id, int): self.locator = _eval_xpath('form', index=id) else: self.locator = _eval_xpath('form', id=id) if not self: raise LookupError('No form found at %s' % self.locator) form = self.eval('e.innerHTML') super(Form, self).__init__(resp, '<form>%s</form>' % form) def _parse_fields(self): super(Form, self)._parse_fields() # Add index to locators for name, fields in self.fields.items(): if len(fields) > 1: for i, field in enumerate(fields): field.locator += '[%s]' % (i + 1,) def submit(self, name=None, index=None, extra_environ=None, timeout=None): """Submits the form. If ``name`` is given, then also select that button (using ``index`` to disambiguate)``. Returns a :class:`webtest.browser.TestResponse` object. """ if timeout != 0: self.browser.captureNetworkTraffic('json') self.test_app._make_environ(extra_environ) if name: selector = _eval_xpath('input', locator=self.locator, name=name, index=index) else: self.browser.submit(self.locator) return self.test_app._get_response(resp=self.resp, timeout=timeout) ############### # Servers ############### class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): """Handle a simple file""" def translate_path(self, path): return self.filename def log_message(self, format, *args): log.debug("%s - - [%s] %s\n" % (self.address_string(), self.log_date_time_string(), format % args)) ############### # Misc ############### def _get_command(cmd): if '_' in cmd: cmd = cmd.split('_') cmd = [cmd.pop(0)] + [c.title() for c in cmd] cmd = ''.join(cmd) return cmd def _eval_xpath(tag, locator=None, index=None, **kwargs): if not locator: locator = 'xpath=' locator += "//%s" % tag for k, v in kwargs.items(): if k in ('for_', 'class_'): k = k.strip('_') if v: if k == 'description': locator += '[.="%s"]' % v else: locator += '[@%s="%s"]' % (k, v) if index is not None: locator += '[%s]' % (index + 1,) return locator def is_available(): """return True if the selenium module is available and a RC server is running""" host = os.environ.get('SELENIUM_HOST', '') port = int(os.environ.get('SELENIUM_PORT', 4444)) if not http.check_server(host, port, retries=1): if 'SELENIUM_JAR' not in os.environ: return False else: jar = os.environ['SELENIUM_JAR'] p = subprocess.Popen(['java', '-jar', jar]) os.environ['SELENIUM_PID'] = str( return http.check_server(host, port) return True