summaryrefslogtreecommitdiff
path: root/paste/fixture.py
diff options
context:
space:
mode:
Diffstat (limited to 'paste/fixture.py')
-rw-r--r--paste/fixture.py1755
1 files changed, 1755 insertions, 0 deletions
diff --git a/paste/fixture.py b/paste/fixture.py
new file mode 100644
index 0000000..df1c75d
--- /dev/null
+++ b/paste/fixture.py
@@ -0,0 +1,1755 @@
+# (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 [('&nbsp;', ' '), ('&gt;', '>'),
+ ('&lt;', '<'), ('&quot;', '"'),
+ ('&amp;', '&')]:
+ v = v.replace(ent, repl)
+ return v
View Lorry Controller Status