# (c) 2005 Ian Bicking and contributors; written for Paste
# (http://pythonpaste.org)
# Licensed under the MIT license:
# http://www.opensource.org/licenses/mit-license.php
"""
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
try:
unicode()
except NameError:
unicode = str
log = logging.getLogger(__name__)
if 'SELENIUM_VERBOSE' in os.environ:
log.addHandler(logging.StreamHandler(sys.stderr))
log.setLevel(logging.DEBUG)
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"""
@wraps(func)
def wrapper(*args):
if is_available():
if args and isinstance(args[0], unittest.TestCase):
self = args[0]
if isinstance(self.app, SeleniumApp):
func(self)
else:
old_app = self.app
self.app = SeleniumApp(self.app.app)
try:
func(self)
finally:
self.app.close()
self.app = old_app
else:
# function
func(*args)
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
else:
test_app = resp.test_app
app = SeleniumApp(test_app.app)
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)
fd.write(resp.body)
fd.close()
response = app.get('/__file__', dict(__file__=fd.name))
try:
yield response
finally:
body = app.browser.getHtmlSource()
if PY3:
body = body.encode(resp.charset or 'utf-8')
resp.body = body
resp._forms_indexed = None
resp.updated = True
app.close()
os.unlink(fd.name)
[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
<http://goo.gl/IecEk>`_
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):
self.host = os.environ.get('SELENIUM_HOST', '127.0.0.1')
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.host, self.port)
try:
conn.request("POST", "/selenium-server/driver/", data, headers)
resp = conn.getresponse()
data = resp.read()
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.app = 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 self.app:
req.host = '%s:%s' % (self.server.adj.host, 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)
self.browser.open(req.url)
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.app = 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):
self.app._check_status(status, resp)
if not status:
self.app._check_errors(resp)
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 = self.doc.link(description=description, linkid=linkid,
href=href, index=index)
link.click()
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
<http://goo.gl/IecEk>`_
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,
name=self.name)
value = property(Element.value__get, Element.value__set)
class Select(Field):
"""Field representing ``<select>``"""
def force_value(self, value):
self.select('value=%s' % value)
def value__set(self, value):
self.select('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 (element.name == name && element.checked) {
values.push('name='+element.value);
}
}
return values.join('&');
}(this))""" % (self.name, 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 = [self.name] + 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)
self.browser.click(selector)
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', '127.0.0.1')
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(p.pid)
return http.check_server(host, port)
return True