# (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.
Most interesting is the `TestApp <class-paste.fixture.TestApp.html>`_
for testing WSGI applications, and the `TestFileEnvironment
<class-paste.fixture.TestFileEnvironment.html>`_ class for testing the
effects of command-line scripts.
"""
from __future__ import print_function
import sys
import random
import mimetypes
import time
import os
import shutil
import smtplib
import shlex
import re
import six
import subprocess
from six.moves import cStringIO as StringIO
from six.moves.urllib.parse import urlencode
from six.moves.urllib import parse as urlparse
try:
# Python 3
from http.cookies import BaseCookie
from urllib.parse import splittype, splithost
except ImportError:
# Python 2
from Cookie import BaseCookie
from urllib import splittype, splithost
from paste import wsgilib
from paste import lint
from paste.response import HeaderDict
def tempnam_no_warning(*args):
"""
An os.tempnam with the warning turned off, because sometimes
you just need to use this and don't care about the stupid
security warning.
"""
return os.tempnam(*args)
class NoDefault(object):
pass
def sorted(l):
l = list(l)
l.sort()
return l
class Dummy_smtplib(object):
existing = None
def __init__(self, server):
import warnings
warnings.warn(
'Dummy_smtplib is not maintained and is deprecated',
DeprecationWarning, 2)
assert not self.existing, (
"smtplib.SMTP() called again before Dummy_smtplib.existing.reset() "
"called.")
self.server = server
self.open = True
self.__class__.existing = self
def quit(self):
assert self.open, (
"Called %s.quit() twice" % self)
self.open = False
def sendmail(self, from_address, to_addresses, msg):
self.from_address = from_address
self.to_addresses = to_addresses
self.message = msg
def install(cls):
smtplib.SMTP = cls
install = classmethod(install)
def reset(self):
assert not self.open, (
"SMTP connection not quit")
self.__class__.existing = None
class AppError(Exception):
pass
class TestApp(object):
# for py.test
disabled = True
def __init__(self, app, namespace=None, relative_to=None,
extra_environ=None, pre_request_hook=None,
post_request_hook=None):
"""
Wraps a WSGI application in a more convenient interface for
testing.
``app`` may be an application, or a Paste Deploy app
URI, like ``'config:filename.ini#test'``.
``namespace`` is a dictionary that will be written to (if
provided). This can be used with doctest or some other
system, and the variable ``res`` will be assigned everytime
you make a request (instead of returning the request).
``relative_to`` is a directory, and filenames used for file
uploads are calculated relative to this. Also ``config:``
URIs that aren't absolute.
``extra_environ`` is a dictionary of values that should go
into the environment for each request. These can provide a
communication channel with the application.
``pre_request_hook`` is a function to be called prior to
making requests (such as ``post`` or ``get``). This function
must take one argument (the instance of the TestApp).
``post_request_hook`` is a function, similar to
``pre_request_hook``, to be called after requests are made.
"""
if isinstance(app, (six.binary_type, six.text_type)):
from paste.deploy import loadapp
# @@: Should pick up relative_to from calling module's
# __file__
app = loadapp(app, relative_to=relative_to)
self.app = app
self.namespace = namespace
self.relative_to = relative_to
if extra_environ is None:
extra_environ = {}
self.extra_environ = extra_environ
self.pre_request_hook = pre_request_hook
self.post_request_hook = post_request_hook
self.reset()
def reset(self):
"""
Resets the state of the application; currently just clears
saved cookies.
"""
self.cookies = {}
def _make_environ(self):
environ = self.extra_environ.copy()
environ['paste.throw_errors'] = True
return environ
def get(self, url, params=None, headers=None, extra_environ=None,
status=None, expect_errors=False):
"""
Get the given url (well, actually a path like
``'/page.html'``).
``params``:
A query string, or a dictionary that will be encoded
into a query string. You may also include a query
string on the ``url``.
``headers``:
A dictionary of extra headers to send.
``extra_environ``:
A dictionary of environmental variables that should
be added to the request.
``status``:
The integer status code you expect (if not 200 or 3xx).
If you expect a 404 response, for instance, you must give
``status=404`` or it will be an error. You can also give
a wildcard, like ``'3*'`` or ``'*'``.
``expect_errors``:
If this is not true, then if anything is written to
``wsgi.errors`` it will be an error. If it is true, then
non-200/3xx responses are also okay.
Returns a `response object
<class-paste.fixture.TestResponse.html>`_
"""
if extra_environ is None:
extra_environ = {}
# Hide from py.test:
__tracebackhide__ = True
if params:
if not isinstance(params, (six.binary_type, six.text_type)):
params = urlencode(params, doseq=True)
if '?' in url:
url += '&'
else:
url += '?'
url += params
environ = self._make_environ()
url = str(url)
if '?' in url:
url, environ['QUERY_STRING'] = url.split('?', 1)
else:
environ['QUERY_STRING'] = ''
self._set_headers(headers, environ)
environ.update(extra_environ)
req = TestRequest(url, environ, expect_errors)
return self.do_request(req, status=status)
def _gen_request(self, method, url, params=b'', headers=None, extra_environ=None,
status=None, upload_files=None, expect_errors=False):
"""
Do a generic request.
"""
if headers is None:
headers = {}
if extra_environ is None:
extra_environ = {}
environ = self._make_environ()
# @@: Should this be all non-strings?
if isinstance(params, (list, tuple, dict)):
params = urlencode(params)
if hasattr(params, 'items'):
# Some other multi-dict like format
params = urlencode(params.items())
if six.PY3:
params = params.encode('utf8')
if upload_files:
params = urlparse.parse_qsl(params, keep_blank_values=True)
content_type, params = self.encode_multipart(
params, upload_files)
environ['CONTENT_TYPE'] = content_type
elif params:
environ.setdefault('CONTENT_TYPE', 'application/x-www-form-urlencoded')
if '?' in url:
url, environ['QUERY_STRING'] = url.split('?', 1)
else:
environ['QUERY_STRING'] = ''
environ['CONTENT_LENGTH'] = str(len(params))
environ['REQUEST_METHOD'] = method
environ['wsgi.input'] = six.BytesIO(params)
self._set_headers(headers, environ)
environ.update(extra_environ)
req = TestRequest(url, environ, expect_errors)
return self.do_request(req, status=status)
def post(self, url, params=b'', headers=None, extra_environ=None,
status=None, upload_files=None, expect_errors=False):
"""
Do a POST request. Very like the ``.get()`` method.
``params`` are put in the body of the request.
``upload_files`` is for file uploads. It should be a list of
``[(fieldname, filename, file_content)]``. You can also use
just ``[(fieldname, filename)]`` and the file content will be
read from disk.
Returns a `response object
<class-paste.fixture.TestResponse.html>`_
"""
return self._gen_request('POST', url, params=params, headers=headers,
extra_environ=extra_environ,status=status,
upload_files=upload_files,
expect_errors=expect_errors)
def put(self, url, params=b'', headers=None, extra_environ=None,
status=None, upload_files=None, expect_errors=False):
"""
Do a PUT request. Very like the ``.get()`` method.
``params`` are put in the body of the request.
``upload_files`` is for file uploads. It should be a list of
``[(fieldname, filename, file_content)]``. You can also use
just ``[(fieldname, filename)]`` and the file content will be
read from disk.
Returns a `response object
<class-paste.fixture.TestResponse.html>`_
"""
return self._gen_request('PUT', url, params=params, headers=headers,
extra_environ=extra_environ,status=status,
upload_files=upload_files,
expect_errors=expect_errors)
def delete(self, url, params=b'', headers=None, extra_environ=None,
status=None, expect_errors=False):
"""
Do a DELETE request. Very like the ``.get()`` method.
``params`` are put in the body of the request.
Returns a `response object
<class-paste.fixture.TestResponse.html>`_
"""
return self._gen_request('DELETE', url, params=params, headers=headers,
extra_environ=extra_environ,status=status,
upload_files=None, expect_errors=expect_errors)
def _set_headers(self, headers, environ):
"""
Turn any headers into environ variables
"""
if not headers:
return
for header, value in headers.items():
if header.lower() == 'content-type':
var = 'CONTENT_TYPE'
elif header.lower() == 'content-length':
var = 'CONTENT_LENGTH'
else:
var = 'HTTP_%s' % header.replace('-', '_').upper()
environ[var] = value
def encode_multipart(self, params, files):
"""
Encodes a set of parameters (typically a name/value list) and
a set of files (a list of (name, filename, file_body)) into a
typical POST body, returning the (content_type, body).
"""
boundary = '----------a_BoUnDaRy%s$' % random.random()
content_type = 'multipart/form-data; boundary=%s' % boundary
if six.PY3:
boundary = boundary.encode('ascii')
lines = []
for key, value in params:
lines.append(b'--'+boundary)
line = 'Content-Disposition: form-data; name="%s"' % key
if six.PY3:
line = line.encode('utf8')
lines.append(line)
lines.append(b'')
line = value
if six.PY3 and isinstance(line, six.text_type):
line = line.encode('utf8')
lines.append(line)
for file_info in files:
key, filename, value = self._get_file_info(file_info)
lines.append(b'--'+boundary)
line = ('Content-Disposition: form-data; name="%s"; filename="%s"'
% (key, filename))
if six.PY3:
line = line.encode('utf8')
lines.append(line)
fcontent = mimetypes.guess_type(filename)[0]
line = ('Content-Type: %s'
% (fcontent or 'application/octet-stream'))
if six.PY3:
line = line.encode('utf8')
lines.append(line)
lines.append(b'')
lines.append(value)
lines.append(b'--' + boundary + b'--')
lines.append(b'')
body = b'\r\n'.join(lines)
return content_type, body
def _get_file_info(self, file_info):
if len(file_info) == 2:
# It only has a filename
filename = file_info[1]
if self.relative_to:
filename = os.path.join(self.relative_to, filename)
f = open(filename, 'rb')
content = f.read()
f.close()
return (file_info[0], filename, content)
elif len(file_info) == 3:
return file_info
else:
raise ValueError(
"upload_files need to be a list of tuples of (fieldname, "
"filename, filecontent) or (fieldname, filename); "
"you gave: %r"
% repr(file_info)[:100])
def do_request(self, req, status):
"""
Executes the given request (``req``), with the expected
``status``. Generally ``.get()`` and ``.post()`` are used
instead.
"""
if self.pre_request_hook:
self.pre_request_hook(self)
__tracebackhide__ = True
if self.cookies:
c = BaseCookie()
for name, value in self.cookies.items():
c[name] = value
hc = '; '.join(['='.join([m.key, m.value]) for m in c.values()])
req.environ['HTTP_COOKIE'] = hc
req.environ['paste.testing'] = True
req.environ['paste.testing_variables'] = {}
app = lint.middleware(self.app)
old_stdout = sys.stdout
out = CaptureStdout(old_stdout)
try:
sys.stdout = out
start_time = time.time()
raise_on_wsgi_error = not req.expect_errors
raw_res = wsgilib.raw_interactive(
app, req.url,
raise_on_wsgi_error=raise_on_wsgi_error,
**req.environ)
end_time = time.time()
finally:
sys.stdout = old_stdout
sys.stderr.write(out.getvalue())
res = self._make_response(raw_res, end_time - start_time)
res.request = req
for name, value in req.environ['paste.testing_variables'].items():
if hasattr(res, name):
raise ValueError(
"paste.testing_variables contains the variable %r, but "
"the response object already has an attribute by that "
"name" % name)
setattr(res, name, value)
if self.namespace is not None:
self.namespace['res'] = res
if not req.expect_errors:
self._check_status(status, res)
self._check_errors(res)
res.cookies_set = {}
for header in res.all_headers('set-cookie'):
c = BaseCookie(header)
for key, morsel in c.items():
self.cookies[key] = morsel.value
res.cookies_set[key] = morsel.value
if self.post_request_hook:
self.post_request_hook(self)
if self.namespace is None:
# It's annoying to return the response in doctests, as it'll
# be printed, so we only return it is we couldn't assign
# it anywhere
return res
def _check_status(self, status, res):
__tracebackhide__ = True
if status == '*':
return
if isinstance(status, (list, tuple)):
if res.status not in status:
raise AppError(
"Bad response: %s (not one of %s for %s)\n%s"
% (res.full_status, ', '.join(map(str, status)),
res.request.url, res.body))
return
if status is None:
if res.status >= 200 and res.status < 400:
return
body = res.body
if six.PY3:
body = body.decode('utf8', 'xmlcharrefreplace')
raise AppError(
"Bad response: %s (not 200 OK or 3xx redirect for %s)\n%s"
% (res.full_status, res.request.url,
body))
if status != res.status:
raise AppError(
"Bad response: %s (not %s)" % (res.full_status, status))
def _check_errors(self, res):
if res.errors:
raise AppError(
"Application had errors logged:\n%s" % res.errors)
def _make_response(self, resp, total_time):
status, headers, body, errors = resp
return TestResponse(self, status, headers, body, errors,
total_time)
class CaptureStdout(object):
def __init__(self, actual):
self.captured = StringIO()
self.actual = actual
def write(self, s):
self.captured.write(s)
self.actual.write(s)
def flush(self):
self.actual.flush()
def writelines(self, lines):
for item in lines:
self.write(item)
def getvalue(self):
return self.captured.getvalue()
class TestResponse(object):
# for py.test
disabled = True
"""
Instances of this class are return by `TestApp
<class-paste.fixture.TestApp.html>`_
"""
def __init__(self, test_app, status, headers, body, errors,
total_time):
self.test_app = test_app
self.status = int(status.split()[0])
self.full_status = status
self.headers = headers
self.header_dict = HeaderDict.fromlist(self.headers)
self.body = body
self.errors = errors
self._normal_body = None
self.time = total_time
self._forms_indexed = None
def forms__get(self):
"""
Returns a dictionary of ``Form`` objects. Indexes are both in
order (from zero) and by form id (if the form is given an id).
"""
if self._forms_indexed is None:
self._parse_forms()
return self._forms_indexed
forms = property(forms__get,
doc="""
A list of <form>s found on the page (instances of
`Form <class-paste.fixture.Form.html>`_)
""")
def form__get(self):
forms = self.forms
if not forms:
raise TypeError(
"You used response.form, but no forms exist")
if 1 in forms:
# There is more than one form
raise TypeError(
"You used response.form, but more than one form exists")
return forms[0]
form = property(form__get,
doc="""
Returns a single `Form
<class-paste.fixture.Form.html>`_ instance; it
is an error if there are multiple forms on the
page.
""")
_tag_re = re.compile(r'<(/?)([:a-z0-9_\-]*)(.*?)>', re.S|re.I)
def _parse_forms(self):
forms = self._forms_indexed = {}
form_texts = []
started = None
for match in self._tag_re.finditer(self.body):
end = match.group(1) == '/'
tag = match.group(2).lower()
if tag != 'form':
continue
if end:
assert started, (
"</form> unexpected at %s" % match.start())
form_texts.append(self.body[started:match.end()])
started = None
else:
assert not started, (
"Nested form tags at %s" % match.start())
started = match.start()
assert not started, (
"Danging form: %r" % self.body[started:])
for i, text in enumerate(form_texts):
form = Form(self, text)
forms[i] = form
if form.id:
forms[form.id] = form
def header(self, name, default=NoDefault):
"""
Returns the named header; an error if there is not exactly one
matching header (unless you give a default -- always an error
if there is more than one header)
"""
found = None
for cur_name, value in self.headers:
if cur_name.lower() == name.lower():
assert not found, (
"Ambiguous header: %s matches %r and %r"
% (name, found, value))
found = value
if found is None:
if default is NoDefault:
raise KeyError(
"No header found: %r (from %s)"
% (name, ', '.join([n for n, v in self.headers])))
else:
return default
return found
def all_headers(self, name):
"""
Gets all headers by the ``name``, returns as a list
"""
found = []
for cur_name, value in self.headers:
if cur_name.lower() == name.lower():
found.append(value)
return found
def follow(self, **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.
"""
assert self.status >= 300 and self.status < 400, (
"You can only follow redirect responses (not %s)"
% self.full_status)
location = self.header('location')
type, rest = splittype(location)
host, path = splithost(rest)
# @@: We should test that it's not a remote redirect
return self.test_app.get(location, **kw)
def click(self, description=None, linkid=None, href=None,
anchor=None, index=None, verbose=False):
"""
Click the link as described. Each of ``description``,
``linkid``, and ``url`` are *patterns*, meaning that they are
either strings (regular expressions), compiled regular
expressions (objects with a ``search`` method), or callables
returning true or false.
All the given patterns are ANDed together:
* ``description`` is a pattern that matches the contents of the
anchor (HTML and all -- everything between ``<a...>`` and
``</a>``)
* ``linkid`` is a pattern that matches the ``id`` attribute of
the anchor. It will receive the empty string if no id is
given.
* ``href`` is a pattern that matches the ``href`` of the anchor;
the literal content of that attribute, not the fully qualified
attribute.
* ``anchor`` is a pattern that matches the entire anchor, with
its contents.
If more than one link matches, then the ``index`` link is
followed. If ``index`` is not given and more than one link
matches, or if no link matches, then ``IndexError`` will be
raised.
If you give ``verbose`` then messages will be printed about
each link, and why it does or doesn't match. If you use
``app.click(verbose=True)`` you'll see a list of all the
links.
You can use multiple criteria to essentially assert multiple
aspects about the link, e.g., where the link's destination is.
"""
__tracebackhide__ = True
found_html, found_desc, found_attrs = self._find_element(
tag='a', href_attr='href',
href_extract=None,
content=description,
id=linkid,
href_pattern=href,
html_pattern=anchor,
index=index, verbose=verbose)
return self.goto(found_attrs['uri'])
def clickbutton(self, description=None, buttonid=None, href=None,
button=None, index=None, verbose=False):
"""
Like ``.click()``, except looks for link-like buttons.
This kind of button should look like
``<button onclick="...location.href='url'...">``.
"""
__tracebackhide__ = True
found_html, found_desc, found_attrs = self._find_element(
tag='button', href_attr='onclick',
href_extract=re.compile(r"location\.href='(.*?)'"),
content=description,
id=buttonid,
href_pattern=href,
html_pattern=button,
index=index, verbose=verbose)
return self.goto(found_attrs['uri'])
def _find_element(self, tag, href_attr, href_extract,
content, id,
href_pattern,
html_pattern,
index, verbose):
content_pat = _make_pattern(content)
id_pat = _make_pattern(id)
href_pat = _make_pattern(href_pattern)
html_pat = _make_pattern(html_pattern)
_tag_re = re.compile(r'<%s\s+(.*?)>(.*?)</%s>' % (tag, tag),
re.I+re.S)
def printlog(s):
if verbose:
print(s)
found_links = []
total_links = 0
for match in _tag_re.finditer(self.body):
el_html = match.group(0)
el_attr = match.group(1)
el_content = match.group(2)
attrs = _parse_attrs(el_attr)
if verbose:
printlog('Element: %r' % el_html)
if not attrs.get(href_attr):
printlog(' Skipped: no %s attribute' % href_attr)
continue
el_href = attrs[href_attr]
if href_extract:
m = href_extract.search(el_href)
if not m:
printlog(" Skipped: doesn't match extract pattern")
continue
el_href = m.group(1)
attrs['uri'] = el_href
if el_href.startswith('#'):
printlog(' Skipped: only internal fragment href')
continue
if el_href.startswith('javascript:'):
printlog(' Skipped: cannot follow javascript:')
continue
total_links += 1
if content_pat and not content_pat(el_content):
printlog(" Skipped: doesn't match description")
continue
if id_pat and not id_pat(attrs.get('id', '')):
printlog(" Skipped: doesn't match id")
continue
if href_pat and not href_pat(el_href):
printlog(" Skipped: doesn't match href")
continue
if html_pat and not html_pat(el_html):
printlog(" Skipped: doesn't match html")
continue
printlog(" Accepted")
found_links.append((el_html, el_content, attrs))
if not found_links:
raise IndexError(
"No matching elements found (from %s possible)"
% total_links)
if index is None:
if len(found_links) > 1:
raise IndexError(
"Multiple links match: %s"
% ', '.join([repr(anc) for anc, d, attr in found_links]))
found_link = found_links[0]
else:
try:
found_link = found_links[index]
except IndexError:
raise IndexError(
"Only %s (out of %s) links match; index %s out of range"
% (len(found_links), total_links, index))
return found_link
def goto(self, href, method='get', **args):
"""
Go to the (potentially relative) link ``href``, using the
given method (``'get'`` or ``'post'``) and any extra arguments
you want to pass to the ``app.get()`` or ``app.post()``
methods.
All hostnames and schemes will be ignored.
"""
scheme, host, path, query, fragment = urlparse.urlsplit(href)
# We
scheme = host = fragment = ''
href = urlparse.urlunsplit((scheme, host, path, query, fragment))
href = urlparse.urljoin(self.request.full_url, href)
method = method.lower()
assert method in ('get', 'post'), (
'Only "get" or "post" are allowed for method (you gave %r)'
% method)
if method == 'get':
method = self.test_app.get
else:
method = self.test_app.post
return method(href, **args)
_normal_body_regex = re.compile(br'[ \n\r\t]+')
def normal_body__get(self):
if self._normal_body is None:
self._normal_body = self._normal_body_regex.sub(
b' ', self.body)
return self._normal_body
normal_body = property(normal_body__get,
doc="""
Return the whitespace-normalized body
""")
def __contains__(self, s):
"""
A response 'contains' a string if it is present in the body
of the response. Whitespace is normalized when searching
for a string.
"""
if not isinstance(s, (six.binary_type, six.text_type)):
s = str(s)
if isinstance(s, six.text_type):
## FIXME: we don't know that this response uses utf8:
s = s.encode('utf8')
return (self.body.find(s) != -1
or self.normal_body.find(s) != -1)
def mustcontain(self, *strings, **kw):
"""
Assert that the response contains all of the strings passed
in as arguments.
Equivalent to::
assert string in res
"""
if 'no' in kw:
no = kw['no']
del kw['no']
if isinstance(no, (six.binary_type, six.text_type)):
no = [no]
else:
no = []
if kw:
raise TypeError(
"The only keyword argument allowed is 'no'")
for s in strings:
if not s in self:
print("Actual response (no %r):" % s, file=sys.stderr)
print(self, file=sys.stderr)
raise IndexError(
"Body does not contain string %r" % s)
for no_s in no:
if no_s in self:
print("Actual response (has %r)" % no_s, file=sys.stderr)
print(self, file=sys.stderr)
raise IndexError(
"Body contains string %r" % s)
def __repr__(self):
body = self.body
if six.PY3:
body = body.decode('utf8', 'xmlcharrefreplace')
body = body[:20]
return '<Response %s %r>' % (self.full_status, body)
def __str__(self):
simple_body = b'\n'.join([l for l in self.body.splitlines()
if l.strip()])
if six.PY3:
simple_body = simple_body.decode('utf8', 'xmlcharrefreplace')
return 'Response: %s\n%s\n%s' % (
self.status,
'\n'.join(['%s: %s' % (n, v) for n, v in self.headers]),
simple_body)
def showbrowser(self):
"""
Show this response in a browser window (for debugging purposes,
when it's hard to read the HTML).
"""
import webbrowser
fn = tempnam_no_warning(None, 'paste-fixture') + '.html'
f = open(fn, 'wb')
f.write(self.body)
f.close()
url = 'file:' + fn.replace(os.sep, '/')
webbrowser.open_new(url)
class TestRequest(object):
# for py.test
disabled = True
"""
Instances of this class are created by `TestApp
<class-paste.fixture.TestApp.html>`_ with the ``.get()`` and
``.post()`` methods, and are consumed there by ``.do_request()``.
Instances are also available as a ``.req`` attribute on
`TestResponse <class-paste.fixture.TestResponse.html>`_ instances.
Useful attributes:
``url``:
The url (actually usually the path) of the request, without
query string.
``environ``:
The environment dictionary used for the request.
``full_url``:
The url/path, with query string.
"""
def __init__(self, url, environ, expect_errors=False):
if url.startswith('http://localhost'):
url = url[len('http://localhost'):]
self.url = url
self.environ = environ
if environ.get('QUERY_STRING'):
self.full_url = url + '?' + environ['QUERY_STRING']
else:
self.full_url = url
self.expect_errors = expect_errors
class Form(object):
"""
This object represents a form that has been found in a page.
This has a couple useful attributes:
``text``:
the full HTML of the form.
``action``:
the relative URI of the action.
``method``:
the method (e.g., ``'GET'``).
``id``:
the id, or None if not given.
``fields``:
a dictionary of fields, each value is a list of fields by
that name. ``<input type=\"radio\">`` and ``<select>`` are
both represented as single fields with multiple options.
"""
# @@: This really should be using Mechanize/ClientForm or
# something...
_tag_re = re.compile(r'<(/?)([:a-z0-9_\-]*)([^>]*?)>', re.I)
def __init__(self, response, text):
self.response = response
self.text = text
self._parse_fields()
self._parse_action()
def _parse_fields(self):
in_select = None
in_textarea = None
fields = {}
for match in self._tag_re.finditer(self.text):
end = match.group(1) == '/'
tag = match.group(2).lower()
if tag not in ('input', 'select', 'option', 'textarea',
'button'):
continue
if tag == 'select' and end:
assert in_select, (
'%r without starting select' % match.group(0))
in_select = None
continue
if tag == 'textarea' and end:
assert in_textarea, (
"</textarea> with no <textarea> at %s" % match.start())
in_textarea[0].value = html_unquote(self.text[in_textarea[1]:match.start()])
in_textarea = None
continue
if end:
continue
attrs = _parse_attrs(match.group(3))
if 'name' in attrs:
name = attrs.pop('name')
else:
name = None
if tag == 'option':
in_select.options.append((attrs.get('value'),
'selected' in attrs))
continue
if tag == 'input' and attrs.get('type') == 'radio':
field = fields.get(name)
if not field:
field = Radio(self, tag, name, match.start(), **attrs)
fields.setdefault(name, []).append(field)
else:
field = field[0]
assert isinstance(field, Radio)
field.options.append((attrs.get('value'),
'checked' in attrs))
continue
tag_type = tag
if tag == 'input':
tag_type = attrs.get('type', 'text').lower()
FieldClass = Field.classes.get(tag_type, Field)
field = FieldClass(self, tag, name, match.start(), **attrs)
if tag == 'textarea':
assert not in_textarea, (
"Nested textareas: %r and %r"
% (in_textarea, match.group(0)))
in_textarea = field, match.end()
elif tag == 'select':
assert not in_select, (
"Nested selects: %r and %r"
% (in_select, match.group(0)))
in_select = field
fields.setdefault(name, []).append(field)
self.fields = fields
def _parse_action(self):
self.action = None
for match in self._tag_re.finditer(self.text):
end = match.group(1) == '/'
tag = match.group(2).lower()
if tag != 'form':
continue
if end:
break
attrs = _parse_attrs(match.group(3))
self.action = attrs.get('action', '')
self.method = attrs.get('method', 'GET')
self.id = attrs.get('id')
# @@: enctype?
else:
assert 0, "No </form> tag found"
assert self.action is not None, (
"No <form> tag found")
def __setitem__(self, name, value):
"""
Set the value of the named field. If there is 0 or multiple
fields by that name, it is an error.
Setting the value of a ``<select>`` selects the given option
(and confirms it is an option). Setting radio fields does the
same. Checkboxes get boolean values. You cannot set hidden
fields or buttons.
Use ``.set()`` if there is any ambiguity and you must provide
an index.
"""
fields = self.fields.get(name)
assert fields is not None, (
"No field by the name %r found (fields: %s)"
% (name, ', '.join(map(repr, self.fields.keys()))))
assert len(fields) == 1, (
"Multiple fields match %r: %s"
% (name, ', '.join(map(repr, fields))))
fields[0].value = value
def __getitem__(self, name):
"""
Get the named field object (ambiguity is an error).
"""
fields = self.fields.get(name)
assert fields is not None, (
"No field by the name %r found" % name)
assert len(fields) == 1, (
"Multiple fields match %r: %s"
% (name, ', '.join(map(repr, fields))))
return fields[0]
def set(self, name, value, index=None):
"""
Set the given name, using ``index`` to disambiguate.
"""
if index is None:
self[name] = value
else:
fields = self.fields.get(name)
assert fields is not None, (
"No fields found matching %r" % name)
field = fields[index]
field.value = value
def get(self, name, index=None, default=NoDefault):
"""
Get the named/indexed field object, or ``default`` if no field
is found.
"""
fields = self.fields.get(name)
if fields is None and default is not NoDefault:
return default
if index is None:
return self[name]
else:
fields = self.fields.get(name)
assert fields is not None, (
"No fields found matching %r" % name)
field = fields[index]
return field
def select(self, name, value, index=None):
"""
Like ``.set()``, except also confirms the target is a
``<select>``.
"""
field = self.get(name, index=index)
assert isinstance(field, Select)
field.value = value
def submit(self, name=None, index=None, **args):
"""
Submits the form. If ``name`` is given, then also select that
button (using ``index`` to disambiguate)``.
Any extra keyword arguments are passed to the ``.get()`` or
``.post()`` method.
Returns a response object.
"""
fields = self.submit_fields(name, index=index)
return self.response.goto(self.action, method=self.method,
params=fields, **args)
def submit_fields(self, name=None, index=None):
"""
Return a list of ``[(name, value), ...]`` for the current
state of the form.
"""
submit = []
if name is not None:
field = self.get(name, index=index)
submit.append((field.name, field.value_if_submitted()))
for name, fields in self.fields.items():
if name is None:
continue
for field in fields:
value = field.value
if value is None:
continue
submit.append((name, value))
return submit
_attr_re = re.compile(r'([^= \n\r\t]+)[ \n\r\t]*(?:=[ \n\r\t]*(?:"([^"]*)"|([^"][^ \n\r\t>]*)))?', re.S)
def _parse_attrs(text):
attrs = {}
for match in _attr_re.finditer(text):
attr_name = match.group(1).lower()
attr_body = match.group(2) or match.group(3)
attr_body = html_unquote(attr_body or '')
attrs[attr_name] = attr_body
return attrs
class Field(object):
"""
Field object.
"""
# Dictionary of field types (select, radio, etc) to classes
classes = {}
settable = True
def __init__(self, form, tag, name, pos,
value=None, id=None, **attrs):
self.form = form
self.tag = tag
self.name = name
self.pos = pos
self._value = value
self.id = id
self.attrs = attrs
def value__set(self, value):
if not self.settable:
raise AttributeError(
"You cannot set the value of the <%s> field %r"
% (self.tag, self.name))
self._value = value
def force_value(self, value):
"""
Like setting a value, except forces it even for, say, hidden
fields.
"""
self._value = value
def value__get(self):
return self._value
value = property(value__get, value__set)
class Select(Field):
"""
Field representing ``<select>``
"""
def __init__(self, *args, **attrs):
super(Select, self).__init__(*args, **attrs)
self.options = []
self.multiple = attrs.get('multiple')
assert not self.multiple, (
"<select multiple> not yet supported")
# Undetermined yet:
self.selectedIndex = None
def value__set(self, value):
for i, (option, checked) in enumerate(self.options):
if option == str(value):
self.selectedIndex = i
break
else:
raise ValueError(
"Option %r not found (from %s)"
% (value, ', '.join(
[repr(o) for o, c in self.options])))
def value__get(self):
if self.selectedIndex is not None:
return self.options[self.selectedIndex][0]
else:
for option, checked in self.options:
if checked:
return option
else:
if self.options:
return self.options[0][0]
else:
return None
value = property(value__get, value__set)
Field.classes['select'] = Select
class Radio(Select):
"""
Field representing ``<input type="radio">``
"""
Field.classes['radio'] = Radio
class Checkbox(Field):
"""
Field representing ``<input type="checkbox">``
"""
def __init__(self, *args, **attrs):
super(Checkbox, self).__init__(*args, **attrs)
self.checked = 'checked' in attrs
def value__set(self, value):
self.checked = not not value
def value__get(self):
if self.checked:
if self._value is None:
return 'on'
else:
return self._value
else:
return None
value = property(value__get, value__set)
Field.classes['checkbox'] = Checkbox
class Text(Field):
"""
Field representing ``<input type="text">``
"""
def __init__(self, form, tag, name, pos,
value='', id=None, **attrs):
#text fields default to empty string
Field.__init__(self, form, tag, name, pos,
value=value, id=id, **attrs)
Field.classes['text'] = Text
class Textarea(Text):
"""
Field representing ``<textarea>``
"""
Field.classes['textarea'] = Textarea
class Hidden(Text):
"""
Field representing ``<input type="hidden">``
"""
Field.classes['hidden'] = Hidden
class Submit(Field):
"""
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
############################################################
## Command-line testing
############################################################
class TestFileEnvironment(object):
"""
This represents an environment in which files will be written, and
scripts will be run.
"""
# for py.test
disabled = True
def __init__(self, base_path, template_path=None,
script_path=None,
environ=None, cwd=None, start_clear=True,
ignore_paths=None, ignore_hidden=True):
"""
Creates an environment. ``base_path`` is used as the current
working directory, and generally where changes are looked for.
``template_path`` is the directory to look for *template*
files, which are files you'll explicitly add to the
environment. This is done with ``.writefile()``.
``script_path`` is the PATH for finding executables. Usually
grabbed from ``$PATH``.
``environ`` is the operating system environment,
``os.environ`` if not given.
``cwd`` is the working directory, ``base_path`` by default.
If ``start_clear`` is true (default) then the ``base_path``
will be cleared (all files deleted) when an instance is
created. You can also use ``.clear()`` to clear the files.
``ignore_paths`` is a set of specific filenames that should be
ignored when created in the environment. ``ignore_hidden``
means, if true (default) that filenames and directories
starting with ``'.'`` will be ignored.
"""
self.base_path = base_path
self.template_path = template_path
if environ is None:
environ = os.environ.copy()
self.environ = environ
if script_path is None:
if sys.platform == 'win32':
script_path = environ.get('PATH', '').split(';')
else:
script_path = environ.get('PATH', '').split(':')
self.script_path = script_path
if cwd is None:
cwd = base_path
self.cwd = cwd
if start_clear:
self.clear()
elif not os.path.exists(base_path):
os.makedirs(base_path)
self.ignore_paths = ignore_paths or []
self.ignore_hidden = ignore_hidden
def run(self, script, *args, **kw):
"""
Run the command, with the given arguments. The ``script``
argument can have space-separated arguments, or you can use
the positional arguments.
Keywords allowed are:
``expect_error``: (default False)
Don't raise an exception in case of errors
``expect_stderr``: (default ``expect_error``)
Don't raise an exception if anything is printed to stderr
``stdin``: (default ``""``)
Input to the script
``printresult``: (default True)
Print the result after running
``cwd``: (default ``self.cwd``)
The working directory to run in
Returns a `ProcResponse
<class-paste.fixture.ProcResponse.html>`_ object.
"""
__tracebackhide__ = True
expect_error = _popget(kw, 'expect_error', False)
expect_stderr = _popget(kw, 'expect_stderr', expect_error)
cwd = _popget(kw, 'cwd', self.cwd)
stdin = _popget(kw, 'stdin', None)
printresult = _popget(kw, 'printresult', True)
args = list(map(str, args))
assert not kw, (
"Arguments not expected: %s" % ', '.join(kw.keys()))
if ' ' in script:
assert not args, (
"You cannot give a multi-argument script (%r) "
"and arguments (%s)" % (script, args))
script, args = script.split(None, 1)
args = shlex.split(args)
script = self._find_exe(script)
all = [script] + args
files_before = self._find_files()
proc = subprocess.Popen(all, stdin=subprocess.PIPE,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
cwd=cwd,
env=self.environ)
stdout, stderr = proc.communicate(stdin)
files_after = self._find_files()
result = ProcResult(
self, all, stdin, stdout, stderr,
returncode=proc.returncode,
files_before=files_before,
files_after=files_after)
if printresult:
print(result)
print('-'*40)
if not expect_error:
result.assert_no_error()
if not expect_stderr:
result.assert_no_stderr()
return result
def _find_exe(self, script_name):
if self.script_path is None:
script_name = os.path.join(self.cwd, script_name)
if not os.path.exists(script_name):
raise OSError(
"Script %s does not exist" % script_name)
return script_name
for path in self.script_path:
fn = os.path.join(path, script_name)
if os.path.exists(fn):
return fn
raise OSError(
"Script %s could not be found in %s"
% (script_name, ':'.join(self.script_path)))
def _find_files(self):
result = {}
for fn in os.listdir(self.base_path):
if self._ignore_file(fn):
continue
self._find_traverse(fn, result)
return result
def _ignore_file(self, fn):
if fn in self.ignore_paths:
return True
if self.ignore_hidden and os.path.basename(fn).startswith('.'):
return True
return False
def _find_traverse(self, path, result):
full = os.path.join(self.base_path, path)
if os.path.isdir(full):
result[path] = FoundDir(self.base_path, path)
for fn in os.listdir(full):
fn = os.path.join(path, fn)
if self._ignore_file(fn):
continue
self._find_traverse(fn, result)
else:
result[path] = FoundFile(self.base_path, path)
def clear(self):
"""
Delete all the files in the base directory.
"""
if os.path.exists(self.base_path):
shutil.rmtree(self.base_path)
os.mkdir(self.base_path)
def writefile(self, path, content=None,
frompath=None):
"""
Write a file to the given path. If ``content`` is given then
that text is written, otherwise the file in ``frompath`` is
used. ``frompath`` is relative to ``self.template_path``
"""
full = os.path.join(self.base_path, path)
if not os.path.exists(os.path.dirname(full)):
os.makedirs(os.path.dirname(full))
f = open(full, 'wb')
if content is not None:
f.write(content)
if frompath is not None:
if self.template_path:
frompath = os.path.join(self.template_path, frompath)
f2 = open(frompath, 'rb')
f.write(f2.read())
f2.close()
f.close()
return FoundFile(self.base_path, path)
class ProcResult(object):
"""
Represents the results of running a command in
`TestFileEnvironment
<class-paste.fixture.TestFileEnvironment.html>`_.
Attributes to pay particular attention to:
``stdout``, ``stderr``:
What is produced
``files_created``, ``files_deleted``, ``files_updated``:
Dictionaries mapping filenames (relative to the ``base_dir``)
to `FoundFile <class-paste.fixture.FoundFile.html>`_ or
`FoundDir <class-paste.fixture.FoundDir.html>`_ objects.
"""
def __init__(self, test_env, args, stdin, stdout, stderr,
returncode, files_before, files_after):
self.test_env = test_env
self.args = args
self.stdin = stdin
self.stdout = stdout
self.stderr = stderr
self.returncode = returncode
self.files_before = files_before
self.files_after = files_after
self.files_deleted = {}
self.files_updated = {}
self.files_created = files_after.copy()
for path, f in files_before.items():
if path not in files_after:
self.files_deleted[path] = f
continue
del self.files_created[path]
if f.mtime < files_after[path].mtime:
self.files_updated[path] = files_after[path]
def assert_no_error(self):
__tracebackhide__ = True
assert self.returncode == 0, (
"Script returned code: %s" % self.returncode)
def assert_no_stderr(self):
__tracebackhide__ = True
if self.stderr:
print('Error output:')
print(self.stderr)
raise AssertionError("stderr output not expected")
def __str__(self):
s = ['Script result: %s' % ' '.join(self.args)]
if self.returncode:
s.append(' return code: %s' % self.returncode)
if self.stderr:
s.append('-- stderr: --------------------')
s.append(self.stderr)
if self.stdout:
s.append('-- stdout: --------------------')
s.append(self.stdout)
for name, files, show_size in [
('created', self.files_created, True),
('deleted', self.files_deleted, True),
('updated', self.files_updated, True)]:
if files:
s.append('-- %s: -------------------' % name)
files = files.items()
files.sort()
last = ''
for path, f in files:
t = ' %s' % _space_prefix(last, path, indent=4,
include_sep=False)
last = path
if show_size and f.size != 'N/A':
t += ' (%s bytes)' % f.size
s.append(t)
return '\n'.join(s)
class FoundFile(object):
"""
Represents a single file found as the result of a command.
Has attributes:
``path``:
The path of the file, relative to the ``base_path``
``full``:
The full path
``stat``:
The results of ``os.stat``. Also ``mtime`` and ``size``
contain the ``.st_mtime`` and ``st_size`` of the stat.
``bytes``:
The contents of the file.
You may use the ``in`` operator with these objects (tested against
the contents of the file), and the ``.mustcontain()`` method.
"""
file = True
dir = False
def __init__(self, base_path, path):
self.base_path = base_path
self.path = path
self.full = os.path.join(base_path, path)
self.stat = os.stat(self.full)
self.mtime = self.stat.st_mtime
self.size = self.stat.st_size
self._bytes = None
def bytes__get(self):
if self._bytes is None:
f = open(self.full, 'rb')
self._bytes = f.read()
f.close()
return self._bytes
bytes = property(bytes__get)
def __contains__(self, s):
return s in self.bytes
def mustcontain(self, s):
__tracebackhide__ = True
bytes_ = self.bytes
if s not in bytes_:
print('Could not find %r in:' % s)
print(bytes_)
assert s in bytes_
def __repr__(self):
return '<%s %s:%s>' % (
self.__class__.__name__,
self.base_path, self.path)
class FoundDir(object):
"""
Represents a directory created by a command.
"""
file = False
dir = True
def __init__(self, base_path, path):
self.base_path = base_path
self.path = path
self.full = os.path.join(base_path, path)
self.size = 'N/A'
self.mtime = 'N/A'
def __repr__(self):
return '<%s %s:%s>' % (
self.__class__.__name__,
self.base_path, self.path)
def _popget(d, key, default=None):
"""
Pop the key if found (else return default)
"""
if key in d:
return d.pop(key)
return default
def _space_prefix(pref, full, sep=None, indent=None, include_sep=True):
"""
Anything shared by pref and full will be replaced with spaces
in full, and full returned.
"""
if sep is None:
sep = os.path.sep
pref = pref.split(sep)
full = full.split(sep)
padding = []
while pref and full and pref[0] == full[0]:
if indent is None:
padding.append(' ' * (len(full[0]) + len(sep)))
else:
padding.append(' ' * indent)
full.pop(0)
pref.pop(0)
if padding:
if include_sep:
return ''.join(padding) + sep + sep.join(full)
else:
return ''.join(padding) + sep.join(full)
else:
return sep.join(full)
def _make_pattern(pat):
if pat is None:
return None
if isinstance(pat, (six.binary_type, six.text_type)):
pat = re.compile(pat)
if hasattr(pat, 'search'):
return pat.search
if callable(pat):
return pat
assert 0, (
"Cannot make callable pattern object out of %r" % pat)
def setup_module(module=None):
"""
This is used by py.test if it is in the module, so you can
import this directly.
Use like::
from paste.fixture import setup_module
"""
# Deprecated June 2008
import warnings
warnings.warn(
'setup_module is deprecated',
DeprecationWarning, 2)
if module is None:
# The module we were called from must be the module...
module = sys._getframe().f_back.f_globals['__name__']
if isinstance(module, (six.binary_type, six.text_type)):
module = sys.modules[module]
if hasattr(module, 'reset_state'):
module.reset_state()
def html_unquote(v):
"""
Unquote (some) entities in HTML. (incomplete)
"""
for ent, repl in [(' ', ' '), ('>', '>'),
('<', '<'), ('"', '"'),
('&', '&')]:
v = v.replace(ent, repl)
return v