# (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 time
import signal
import socket
import types
import webob
import logging
import warnings
import tempfile
import unittest
import threading
import subprocess
from functools import wraps
from webtest import app as testapp
from wsgiref import simple_server
from contextlib import contextmanager
from webtest.compat import PY3
from webtest.compat import urlencode
from webtest.compat import binary_type
from webtest.compat import HTTPConnection
from webtest.compat import CannotSendRequest
from webtest.compat import HTTPServer
from webtest.compat import SimpleHTTPRequestHandler
try:
import json
except ImportError:
try:
import simplejson as json # NOQA
except:
json = False
try:
unicode()
except NameError:
unicode = str
log = logging.getLogger(__name__)
if 'SELENIUM_VERBOSE':
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')
fd.write(resp.body)
fd.flush()
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()
fd.close()
[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)
[docs]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 = 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 = None
if app:
super(SeleniumApp, self).__init__(app, relative_to=relative_to)
self._run_server(self.app)
url = self.app.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
[docs] 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):
if req.method != 'GET':
raise testapp.AppError('Only GET are allowed')
if self.app:
req.host = '%s:%s' % self.app.bind
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)
trafic = json.loads(self.browser.captureNetworkTraffic('json'))
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')
def _run_server(self, app):
"""Run a wsgi server in a separate thread"""
ip, port = _free_port()
self.app = app = WSGIApplication(app, (ip, port))
def run():
httpd = simple_server.make_server(
ip, port, app,
server_class=WSGIServer,
handler_class=WSGIRequestHandler)
httpd.serve_forever()
app.thread = threading.Thread(target=run)
app.thread.start()
conn = HTTPConnection(ip, port)
time.sleep(.5)
for i in range(100):
try:
conn.request('GET', '/__application__')
conn.getresponse()
except (socket.error, CannotSendRequest):
time.sleep(.3)
else:
break
[docs] def close(self):
"""Close selenium and the WSGI server if needed"""
if self.app:
conn = HTTPConnection(*self.app.bind)
for i in range(100):
try:
conn.request('GET', '/__kill_application__')
conn.getresponse()
except socket.error:
conn.close()
break
else:
time.sleep(.3)
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)
[docs]class TestResponse(testapp.TestResponse):
[docs] 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)
@property
def forms(self):
return Forms(self)
@property
def form(self):
return Form(self, 0)
def _body__get(self):
body = self.browser.getHtmlSource()
if PY3:
return body.encode(self.charset or 'utf-8')
if isinstance(body, binary_type):
return unicode(body, self.charset or 'utf-8')
else:
return body
body = property(_body__get)
def __contains__(self, item):
if isinstance(item, Element):
return item.isElementPresent()
return super(TestResponse, self).__contains__(item)
@property
[docs] def doc(self):
"""Expose a :class:`~webtest.browser.Document`"""
return Document(self)
##########
# DOM
##########
[docs]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
[docs] def exist(self):
"""return true is the element is present"""
return self.isElementPresent()
[docs] 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
[docs] def wait_and_click(self, timeout=3000):
"""Wait for an element, click on it and return this element"""
return self.wait().click()
[docs] def hasClass(self, name):
"""True iif the class is present"""
classes = self.attr('class').split()
return name in classes
[docs] def html(self):
"""Return the innerHTML of the element"""
return self.eval('e.innerHTML')
[docs] def text(self):
"""Return the text of the element"""
return self.getText()
[docs] def attr(self, attr):
"""Return the attribute value of the element"""
return self.eval('e.getAttribute(%r)' % str(attr))
[docs] 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 = _get_value(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)
[docs] 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('"', "'"))
[docs]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)
[docs] 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)
[docs] def xpath(self, path):
"""Get an :class:`~webtest.browser.Element` using xpath"""
return Element(self.resp, 'xpath=%s' % path)
[docs] def css(self, selector):
"""Get an :class:`~webtest.browser.Element` using a css selector"""
return Element(self.resp, 'css=%s' % selector)
[docs] 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 __contains__(self, s):
if isinstance(s, Element):
return s.exist()
return self.browser.isTextPresent(_get_value(s))
def __call__(self, locator):
return Element(locator)
###########
# Forms
###########
class Field(testapp.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 = [testapp._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 = [testapp._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 = _free_port()
def run():
FileHandler.filename = filename
server = 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, testapp.Hidden):
"""Field representing ``<input type="hidden">``"""
Field.classes['hidden'] = Hidden
class Submit(Field, testapp.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 Forms(object):
def __init__(self, resp):
self.resp = resp
def __getitem__(self, key):
return Form(self.resp, key)
class WSGIApplication(object):
"""A WSGI middleware to handle special calls used to run a test app"""
def __init__(self, app, bind):
self.app = app
self.serve_forever = True
self.bind = bind
self.url = 'http://%s:%s/' % bind
self.thread = None
def __call__(self, environ, start_response):
if '__kill_application__' in environ['PATH_INFO']:
self.serve_forever = False
resp = webob.Response()
return resp(environ, start_response)
elif '__file__' in environ['PATH_INFO']:
req = webob.Request(environ)
resp = webob.Response()
resp.content_type = 'text/html; charset=UTF-8'
filename = req.params.get('__file__')
body = open(filename).read()
body.replace('http://localhost/',
'http://%s/' % req.host)
if PY3:
resp.text = body
else:
resp.body = body
return resp(environ, start_response)
elif '__application__' in environ['PATH_INFO']:
resp = webob.Response()
return resp(environ, start_response)
return self.app(environ, start_response)
def __repr__(self):
return '<WSGIApplication %r at %s>' % (self.app, self.url)
class WSGIRequestHandler(simple_server.WSGIRequestHandler):
"""A WSGIRequestHandler who log to a logger"""
def log_message(self, format, *args):
log.debug("%s - - [%s] %s" %
(self.address_string(),
self.log_date_time_string(),
format % args))
class WSGIServer(simple_server.WSGIServer):
"""A WSGIServer"""
def serve_forever(self):
while self.application.serve_forever:
self.handle_request()
class FileHandler(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_value(s):
if json:
return json.dumps(s)
else:
return repr(str(s))
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 _free_port():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('', 0))
ip, port = s.getsockname()
s.close()
ip = os.environ.get('SELENIUM_BIND', '127.0.0.1')
return ip, port
def is_available():
"""return True if the selenium module is available and a RC server is
running"""
if json == False:
warnings.warn(
('selenium is not available because no json module are '
'available. Consider installing simplejson'),
SeleniumWarning)
host = os.environ.get('SELENIUM_HOST', '127.0.0.1')
port = int(os.environ.get('SELENIUM_PORT', 4444))
try:
conn = HTTPConnection(host, port)
conn.request('GET', '/')
except socket.error:
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)
for i in range(30):
time.sleep(.3)
try:
conn = HTTPConnection(host, port)
conn.request('GET', '/')
except socket.error:
pass
else:
return True
return False
return True
