summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorBert JW Regeer <xistence@0x58.com>2020-11-28 17:56:52 -0800
committerGitHub <noreply@github.com>2020-11-28 17:56:52 -0800
commit536df8686cc815f0e0ed22e6a7402b83dbec5a00 (patch)
tree1e4ba57db9579d499273344d317c34422acd1c6f
parent07ce90f2da58990b83d06f41b035d3b8ee22547b (diff)
parent0db42eb954a65c759aaaf12490f4b2611b2439b5 (diff)
downloadwebob-536df8686cc815f0e0ed22e6a7402b83dbec5a00.tar.gz
Merge pull request #390 from Pylons/webob-ng
WebOb 2.0 - Now with less Python 2
-rw-r--r--.github/workflows/ci-tests.yml89
-rw-r--r--.travis.yml49
-rw-r--r--appveyor.yml23
-rw-r--r--docs/comment-example-code/example.py105
-rw-r--r--docs/conf.py107
-rw-r--r--docs/doctests.py28
-rw-r--r--docs/jsonrpc-example-code/jsonrpc.py128
-rw-r--r--docs/jsonrpc-example-code/test_jsonrpc.py5
-rw-r--r--docs/wiki-example-code/example.py115
-rw-r--r--pyproject.toml25
-rw-r--r--setup.py12
-rw-r--r--src/webob/__init__.py3
-rw-r--r--src/webob/acceptparse.py15
-rw-r--r--src/webob/byterange.py4
-rw-r--r--src/webob/cachecontrol.py6
-rw-r--r--src/webob/client.py40
-rw-r--r--src/webob/compat.py322
-rw-r--r--src/webob/cookies.py83
-rw-r--r--src/webob/datetime_utils.py30
-rw-r--r--src/webob/dec.py41
-rw-r--r--src/webob/descriptors.py43
-rw-r--r--src/webob/etag.py10
-rw-r--r--src/webob/exc.py44
-rw-r--r--src/webob/headers.py31
-rw-r--r--src/webob/multidict.py142
-rw-r--r--src/webob/request.py414
-rw-r--r--src/webob/response.py260
-rw-r--r--src/webob/static.py14
-rw-r--r--src/webob/util.py105
-rw-r--r--tests/conftest.py20
-rw-r--r--tests/performance_test.py10
-rw-r--r--tests/test_acceptparse.py181
-rw-r--r--tests/test_byterange.py7
-rw-r--r--tests/test_cachecontrol.py22
-rw-r--r--tests/test_client.py10
-rw-r--r--tests/test_client_functional.py2
-rw-r--r--tests/test_compat.py30
-rw-r--r--tests/test_cookies.py60
-rw-r--r--tests/test_datetime_utils.py77
-rw-r--r--tests/test_dec.py14
-rw-r--r--tests/test_descriptors.py152
-rw-r--r--tests/test_etag.py12
-rw-r--r--tests/test_etag_nose.py2
-rw-r--r--tests/test_exc.py20
-rw-r--r--tests/test_in_wsgiref.py20
-rw-r--r--tests/test_misc.py37
-rw-r--r--tests/test_multidict.py27
-rw-r--r--tests/test_request.py920
-rw-r--r--tests/test_response.py90
-rw-r--r--tests/test_static.py10
-rw-r--r--tests/test_transcode.py8
-rw-r--r--tests/test_util.py54
-rw-r--r--tox.ini82
53 files changed, 1564 insertions, 2596 deletions
diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml
new file mode 100644
index 0000000..108a543
--- /dev/null
+++ b/.github/workflows/ci-tests.yml
@@ -0,0 +1,89 @@
+name: Build and test
+
+on:
+ # Only on pushes to master or one of the release branches we build on push
+ push:
+ branches:
+ - master
+ - "[0-9].[0-9]+-branch"
+ tags:
+ # Build pull requests
+ pull_request:
+
+jobs:
+ test:
+ strategy:
+ matrix:
+ py:
+ - "3.6"
+ - "3.7"
+ - "3.8"
+ - "3.9"
+ - "pypy3"
+ os:
+ - "ubuntu-latest"
+ - "windows-latest"
+ - "macos-latest"
+ architecture:
+ - x64
+ - x86
+
+ exclude:
+ # Linux and macOS don't have x86 python
+ - os: "ubuntu-latest"
+ architecture: x86
+ - os: "macos-latest"
+ architecture: x86
+ # Building on PyPy3 on Windows is broken
+ - os: "windows-latest"
+ py: "pypy3"
+
+ name: "Python: ${{ matrix.py }}-${{ matrix.architecture }} on ${{ matrix.os }}"
+ runs-on: ${{ matrix.os }}
+ steps:
+ - uses: actions/checkout@v2
+ - name: Setup python
+ uses: actions/setup-python@v2
+ with:
+ python-version: ${{ matrix.py }}
+ architecture: ${{ matrix.architecture }}
+ - run: pip install tox
+ - name: Running tox
+ run: tox -e py
+ coverage:
+ runs-on: ubuntu-latest
+ name: Validate coverage
+ steps:
+ - uses: actions/checkout@v2
+ - name: Setup python 3.8
+ uses: actions/setup-python@v2
+ with:
+ python-version: 3.8
+ architecture: x64
+
+ - run: pip install tox
+ - run: tox -e py38,coverage
+ docs:
+ runs-on: ubuntu-latest
+ name: Build the documentation
+ steps:
+ - uses: actions/checkout@v2
+ - name: Setup python
+ uses: actions/setup-python@v2
+ with:
+ python-version: 3.8
+ architecture: x64
+ - run: pip install tox
+ - run: tox -e docs
+ lint:
+ runs-on: ubuntu-latest
+ name: Lint the package
+ steps:
+ - uses: actions/checkout@v2
+ - name: Setup python
+ uses: actions/setup-python@v2
+ with:
+ python-version: 3.8
+ architecture: x64
+ - run: pip install tox
+ - run: tox -e lint
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 006033c..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,49 +0,0 @@
-language: python
-sudo: false
-matrix:
- include:
- - python: 2.7
- env: TOXENV=py27
- - python: 3.4
- env: TOXENV=py34
- - python: 3.5
- env: TOXENV=py35
- - python: 3.6
- env: TOXENV=py36
- - python: pypy
- env: TOXENV=pypy
- - python: pypy3
- env: TOXENV=pypy3
- - python: 3.6
- env: TOXENV=py27,py36,coverage
- - python: 3.6
- env: TOXENV=docs
- - python: 3.7
- env: TOXENV=lint
- dist: xenial
- sudo: true
- - python: 3.7
- env: TOXENV=py37
- dist: xenial
- sudo: true
- - python: 3.8-dev
- env: TOXENV=py38
- dist: xenial
- sudo: true
- allow_failures:
- - env: TOXENV=py38
-
-install:
- - travis_retry pip install tox
-
-script:
- - travis_retry tox
-
-notifications:
- email:
- # pyramid-checkins@lists.repoze.org
- - secure: "g8L8cMCg+rStkVmnPLA2cVQWAJUd7Z24wd/mPULT4G9j7T9vgquXETBTKIpCF3vxTfgOSEWEWEPeAQ7RnhkCpkYMGqcfgLK/6TniTcfGlWdFQnS0bc3gvmrTp4bmgiGLd3Yi+ELhrI0P7Pk+1F3HDy2bgzqiOwqBZXjnkb+xYDQ="
- irc:
- channels:
- # chat.freenode.net#pyramid
- - secure: S2iWooGJoLZgkYsj5vWbD0HGbl+4XkuaTPbbLSDdkfGtKrZIcY3jQMZxaNDtXXbChqkxnA+KW0vg5TjpvspGTZnFlZ59csizhwSFWAdLJR8WhdwroWFFOyH6gcSDM3wtaR7w3HHe6Ps0lHiwjljh5UjmK5YxyXuDtnYEF0JJvww=
diff --git a/appveyor.yml b/appveyor.yml
deleted file mode 100644
index bc7aa9b..0000000
--- a/appveyor.yml
+++ /dev/null
@@ -1,23 +0,0 @@
-environment:
- matrix:
- - PYTHON: "C:\\Python35"
- TOXENV: "py35"
- - PYTHON: "C:\\Python27"
- TOXENV: "py27"
- - PYTHON: "C:\\Python27-x64"
- TOXENV: "py27"
- - PYTHON: "C:\\Python35-x64"
- TOXENV: "py35"
-
-cache:
- - '%LOCALAPPDATA%\pip\Cache'
-
-version: '{branch}.{build}'
-
-install:
- - "%PYTHON%\\python.exe -m pip install tox"
-
-build: off
-
-test_script:
- - "%PYTHON%\\Scripts\\tox.exe"
diff --git a/docs/comment-example-code/example.py b/docs/comment-example-code/example.py
index 5b50759..97486b1 100644
--- a/docs/comment-example-code/example.py
+++ b/docs/comment-example-code/example.py
@@ -6,8 +6,8 @@ from cPickle import load, dump
from webob import Request, Response, html_escape
from webob import exc
-class Commenter(object):
+class Commenter(object):
def __init__(self, app, storage_dir):
self.app = app
self.storage_dir = storage_dir
@@ -16,12 +16,12 @@ class Commenter(object):
def __call__(self, environ, start_response):
req = Request(environ)
- if req.path_info_peek() == '.comments':
+ if req.path_info_peek() == ".comments":
return self.process_comment(req)(environ, start_response)
# This is the base path of *this* middleware:
base_url = req.application_url
resp = req.get_response(self.app)
- if resp.content_type != 'text/html' or resp.status_code != 200:
+ if resp.content_type != "text/html" or resp.status_code != 200:
# Not an HTML response, we don't want to
# do anything to it
return resp(environ, start_response)
@@ -40,21 +40,21 @@ class Commenter(object):
if not os.path.exists(filename):
return []
else:
- f = open(filename, 'rb')
+ f = open(filename, "rb")
data = load(f)
f.close()
return data
def save_data(self, url, data):
filename = self.url_filename(url)
- f = open(filename, 'wb')
+ f = open(filename, "wb")
dump(data, f)
f.close()
def url_filename(self, url):
- return os.path.join(self.storage_dir, urllib.quote(url, ''))
+ return os.path.join(self.storage_dir, urllib.quote(url, ""))
- _end_body_re = re.compile(r'</body.*?>', re.I|re.S)
+ _end_body_re = re.compile(r"</body.*?>", re.I | re.S)
def add_to_end(self, html, extra_html):
"""
@@ -64,24 +64,31 @@ class Commenter(object):
if not match:
return html + extra_html
else:
- return html[:match.start()] + extra_html + html[match.start():]
+ return html[: match.start()] + extra_html + html[match.start() :]
def format_comments(self, comments):
if not comments:
- return ''
+ return ""
text = []
- text.append('<hr>')
- text.append('<h2><a name="comment-area"></a>Comments (%s):</h2>' % len(comments))
+ text.append("<hr>")
+ text.append(
+ '<h2><a name="comment-area"></a>Comments (%s):</h2>' % len(comments)
+ )
for comment in comments:
- text.append('<h3><a href="%s">%s</a> at %s:</h3>' % (
- html_escape(comment['homepage']), html_escape(comment['name']),
- time.strftime('%c', comment['time'])))
+ text.append(
+ '<h3><a href="%s">%s</a> at %s:</h3>'
+ % (
+ html_escape(comment["homepage"]),
+ html_escape(comment["name"]),
+ time.strftime("%c", comment["time"]),
+ )
+ )
# Susceptible to XSS attacks!:
- text.append(comment['comments'])
- return ''.join(text)
+ text.append(comment["comments"])
+ return "".join(text)
def submit_form(self, base_path, req):
- return '''<h2>Leave a comment:</h2>
+ return """<h2>Leave a comment:</h2>
<form action="%s/.comments" method="POST">
<input type="hidden" name="url" value="%s">
<table width="100%%">
@@ -94,54 +101,60 @@ class Commenter(object):
<textarea name="comments" rows=10 style="width: 100%%"></textarea><br>
<input type="submit" value="Submit comment">
</form>
- ''' % (base_path, html_escape(req.url))
+ """ % (
+ base_path,
+ html_escape(req.url),
+ )
def process_comment(self, req):
try:
- url = req.params['url']
- name = req.params['name']
- homepage = req.params['homepage']
- comments = req.params['comments']
+ url = req.params["url"]
+ name = req.params["name"]
+ homepage = req.params["homepage"]
+ comments = req.params["comments"]
except KeyError, e:
- resp = exc.HTTPBadRequest('Missing parameter: %s' % e)
+ resp = exc.HTTPBadRequest("Missing parameter: %s" % e)
return resp
data = self.get_data(url)
- data.append(dict(
- name=name,
- homepage=homepage,
- comments=comments,
- time=time.gmtime()))
+ data.append(
+ dict(name=name, homepage=homepage, comments=comments, time=time.gmtime())
+ )
self.save_data(url, data)
- resp = exc.HTTPSeeOther(location=url+'#comment-area')
+ resp = exc.HTTPSeeOther(location=url + "#comment-area")
return resp
-if __name__ == '__main__':
+
+if __name__ == "__main__":
import optparse
- parser = optparse.OptionParser(
- usage='%prog --port=PORT BASE_DIRECTORY'
- )
+
+ parser = optparse.OptionParser(usage="%prog --port=PORT BASE_DIRECTORY")
parser.add_option(
- '-p', '--port',
- default='8080',
- dest='port',
- type='int',
- help='Port to serve on (default 8080)')
+ "-p",
+ "--port",
+ default="8080",
+ dest="port",
+ type="int",
+ help="Port to serve on (default 8080)",
+ )
parser.add_option(
- '--comment-data',
- default='./comments',
- dest='comment_data',
- help='Place to put comment data into (default ./comments/)')
+ "--comment-data",
+ default="./comments",
+ dest="comment_data",
+ help="Place to put comment data into (default ./comments/)",
+ )
options, args = parser.parse_args()
if not args:
- parser.error('You must give a BASE_DIRECTORY')
+ parser.error("You must give a BASE_DIRECTORY")
base_dir = args[0]
from paste.urlparser import StaticURLParser
+
app = StaticURLParser(base_dir)
app = Commenter(app, options.comment_data)
from wsgiref.simple_server import make_server
- httpd = make_server('localhost', options.port, app)
- print 'Serving on http://localhost:%s' % options.port
+
+ httpd = make_server("localhost", options.port, app)
+ print("Serving on http://localhost:%s" % options.port)
try:
httpd.serve_forever()
except KeyboardInterrupt:
- print '^C'
+ print("^C")
diff --git a/docs/conf.py b/docs/conf.py
index 914d0f9..db67971 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -4,30 +4,30 @@ import os
import shlex
extensions = [
- 'sphinx.ext.autodoc',
- 'sphinx.ext.intersphinx',
+ "sphinx.ext.autodoc",
+ "sphinx.ext.intersphinx",
]
# Add any paths that contain templates here, relative to this directory.
-templates_path = ['_templates']
+templates_path = ["_templates"]
# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:
# source_suffix = ['.rst', '.md']
-source_suffix = ['.txt', '.rst']
+source_suffix = [".txt", ".rst"]
# The encoding of source files.
-#source_encoding = 'utf-8-sig'
+# source_encoding = 'utf-8-sig'
# The master toctree document.
-master_doc = 'index'
+master_doc = "index"
# General information about the project.
-project = u'WebOb'
-copyright = u'2018, Ian Bicking, Pylons Project and contributors'
-author = u'Ian Bicking, Pylons Project, and contributors'
+project = "WebOb"
+copyright = "2018, Ian Bicking, Pylons Project and contributors"
+author = "Ian Bicking, Pylons Project, and contributors"
-version = release = pkg_resources.get_distribution('webob').version
+version = release = pkg_resources.get_distribution("webob").version
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
@@ -38,84 +38,83 @@ language = None
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
-exclude_patterns = ['_build', 'jsonrpc-example-code/*', 'file-example-code/*']
+exclude_patterns = ["_build", "jsonrpc-example-code/*", "file-example-code/*"]
# The name of the Pygments (syntax highlighting) style to use.
-pygments_style = 'sphinx'
+pygments_style = "sphinx"
# If true, `todo` and `todoList` produce output, else they produce nothing.
todo_include_todos = False
-modindex_common_prefix = ['webob.']
+modindex_common_prefix = ["webob."]
-autodoc_member_order = 'bysource'
+autodoc_member_order = "bysource"
# -- Options for HTML output ---------------------------------------------
-html_theme = 'alabaster'
+html_theme = "alabaster"
-html_static_path = ['_static']
+html_static_path = ["_static"]
-htmlhelp_basename = 'WebObdoc'
+htmlhelp_basename = "WebObdoc"
-smartquotes=False
+smartquotes = False
# -- Options for LaTeX output ---------------------------------------------
latex_elements = {
-# The paper size ('letterpaper' or 'a4paper').
-#'papersize': 'letterpaper',
-
-# The font size ('10pt', '11pt' or '12pt').
-#'pointsize': '10pt',
-
-# Additional stuff for the LaTeX preamble.
-#'preamble': '',
-
-# Latex figure (float) alignment
-#'figure_align': 'htbp',
+ # The paper size ('letterpaper' or 'a4paper').
+ #'papersize': 'letterpaper',
+ # The font size ('10pt', '11pt' or '12pt').
+ #'pointsize': '10pt',
+ # Additional stuff for the LaTeX preamble.
+ #'preamble': '',
+ # Latex figure (float) alignment
+ #'figure_align': 'htbp',
}
# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
- (master_doc, 'WebOb.tex', u'WebOb Documentation',
- u'Ian Bicking and contributors', 'manual'),
+ (
+ master_doc,
+ "WebOb.tex",
+ "WebOb Documentation",
+ "Ian Bicking and contributors",
+ "manual",
+ ),
]
# The name of an image file (relative to this directory) to place at the top of
# the title page.
-#latex_logo = None
+# latex_logo = None
# For "manual" documents, if this is true, then toplevel headings are parts,
# not chapters.
-#latex_use_parts = False
+# latex_use_parts = False
# If true, show page references after internal links.
-#latex_show_pagerefs = False
+# latex_show_pagerefs = False
# If true, show URL addresses after external links.
-#latex_show_urls = False
+# latex_show_urls = False
# Documents to append as an appendix to all manuals.
-#latex_appendices = []
+# latex_appendices = []
# If false, no module index is generated.
-#latex_domain_indices = True
+# latex_domain_indices = True
# -- Options for manual page output ---------------------------------------
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
-man_pages = [
- (master_doc, 'webob', u'WebOb Documentation',
- [author], 1)
-]
+man_pages = [(master_doc, "webob", "WebOb Documentation", [author], 1)]
# If true, show URL addresses after external links.
-#man_show_urls = False
+# man_show_urls = False
# -- Options for Texinfo output -------------------------------------------
@@ -124,22 +123,28 @@ man_pages = [
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
- (master_doc, 'WebOb', u'WebOb Documentation',
- author, 'WebOb', 'One line description of project.',
- 'Miscellaneous'),
+ (
+ master_doc,
+ "WebOb",
+ "WebOb Documentation",
+ author,
+ "WebOb",
+ "One line description of project.",
+ "Miscellaneous",
+ ),
]
# Documents to append as an appendix to all manuals.
-#texinfo_appendices = []
+# texinfo_appendices = []
# If false, no module index is generated.
-#texinfo_domain_indices = True
+# texinfo_domain_indices = True
# How to display URL addresses: 'footnote', 'no', or 'inline'.
-#texinfo_show_urls = 'footnote'
+# texinfo_show_urls = 'footnote'
# If true, do not generate a @detailmenu in the "Top" node's menu.
-#texinfo_no_detailmenu = False
+# texinfo_no_detailmenu = False
# -- Options for Epub output ----------------------------------------------
@@ -150,9 +155,9 @@ epub_author = author
epub_publisher = author
epub_copyright = copyright
-epub_exclude_files = ['search.html']
+epub_exclude_files = ["search.html"]
# Example configuration for intersphinx: refer to the Python standard library.
intersphinx_mapping = {
- 'python': ('https://docs.python.org/3', None),
+ "python": ("https://docs.python.org/3", None),
}
diff --git a/docs/doctests.py b/docs/doctests.py
index 9aecb31..a84a19b 100644
--- a/docs/doctests.py
+++ b/docs/doctests.py
@@ -1,17 +1,21 @@
import unittest
import doctest
+
def test_suite():
- flags = doctest.ELLIPSIS|doctest.NORMALIZE_WHITESPACE
- return unittest.TestSuite((
- doctest.DocFileSuite('test_request.txt', optionflags=flags),
- doctest.DocFileSuite('test_response.txt', optionflags=flags),
- doctest.DocFileSuite('test_dec.txt', optionflags=flags),
- doctest.DocFileSuite('do-it-yourself.txt', optionflags=flags),
- doctest.DocFileSuite('file-example.txt', optionflags=flags),
- doctest.DocFileSuite('index.txt', optionflags=flags),
- doctest.DocFileSuite('reference.txt', optionflags=flags),
- ))
+ flags = doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE
+ return unittest.TestSuite(
+ (
+ doctest.DocFileSuite("test_request.txt", optionflags=flags),
+ doctest.DocFileSuite("test_response.txt", optionflags=flags),
+ doctest.DocFileSuite("test_dec.txt", optionflags=flags),
+ doctest.DocFileSuite("do-it-yourself.txt", optionflags=flags),
+ doctest.DocFileSuite("file-example.txt", optionflags=flags),
+ doctest.DocFileSuite("index.txt", optionflags=flags),
+ doctest.DocFileSuite("reference.txt", optionflags=flags),
+ )
+ )
+
-if __name__ == '__main__':
- unittest.main(defaultTest='test_suite')
+if __name__ == "__main__":
+ unittest.main(defaultTest="test_suite")
diff --git a/docs/jsonrpc-example-code/jsonrpc.py b/docs/jsonrpc-example-code/jsonrpc.py
index c038992..dbb95ba 100644
--- a/docs/jsonrpc-example-code/jsonrpc.py
+++ b/docs/jsonrpc-example-code/jsonrpc.py
@@ -5,6 +5,7 @@ from simplejson import loads, dumps
import traceback
import sys
+
class JsonRpcApp(object):
"""
Serve the given object via json-rpc (http://json-rpc.org/)
@@ -24,53 +25,45 @@ class JsonRpcApp(object):
return resp(environ, start_response)
def process(self, req):
- if not req.method == 'POST':
- raise exc.HTTPMethodNotAllowed(
- "Only POST allowed",
- allowed='POST')
+ if not req.method == "POST":
+ raise exc.HTTPMethodNotAllowed("Only POST allowed", allowed="POST")
try:
json = loads(req.body)
except ValueError, e:
- raise ValueError('Bad JSON: %s' % e)
+ raise ValueError("Bad JSON: %s" % e)
try:
- method = json['method']
- params = json['params']
- id = json['id']
+ method = json["method"]
+ params = json["params"]
+ id = json["id"]
except KeyError, e:
- raise ValueError(
- "JSON body missing parameter: %s" % e)
- if method.startswith('_'):
+ raise ValueError("JSON body missing parameter: %s" % e)
+ if method.startswith("_"):
raise exc.HTTPForbidden(
- "Bad method name %s: must not start with _" % method)
+ "Bad method name %s: must not start with _" % method
+ )
if not isinstance(params, list):
- raise ValueError(
- "Bad params %r: must be a list" % params)
+ raise ValueError("Bad params %r: must be a list" % params)
try:
method = getattr(self.obj, method)
except AttributeError:
- raise ValueError(
- "No such method %s" % method)
+ raise ValueError("No such method %s" % method)
try:
result = method(*params)
except:
text = traceback.format_exc()
exc_value = sys.exc_info()[1]
error_value = dict(
- name='JSONRPCError',
- code=100,
- message=str(exc_value),
- error=text)
+ name="JSONRPCError", code=100, message=str(exc_value), error=text
+ )
return Response(
status=500,
- content_type='application/json',
- body=dumps(dict(result=None,
- error=error_value,
- id=id)))
+ content_type="application/json",
+ body=dumps(dict(result=None, error=error_value, id=id)),
+ )
return Response(
- content_type='application/json',
- body=dumps(dict(result=result,
- error=None,
- id=id)))
+ content_type="application/json",
+ body=dumps(dict(result=result, error=None, id=id)),
+ )
class ServerProxy(object):
@@ -82,112 +75,127 @@ class ServerProxy(object):
self._url = url
if proxy is None:
from wsgiproxy.exactproxy import proxy_exact_request
+
proxy = proxy_exact_request
self.proxy = proxy
def __getattr__(self, name):
- if name.startswith('_'):
+ if name.startswith("_"):
raise AttributeError(name)
return _Method(self, name)
def __repr__(self):
- return '<%s for %s>' % (
- self.__class__.__name__, self._url)
+ return "<%s for %s>" % (self.__class__.__name__, self._url)
-class _Method(object):
+class _Method(object):
def __init__(self, parent, name):
self.parent = parent
self.name = name
def __call__(self, *args):
- json = dict(method=self.name,
- id=None,
- params=list(args))
+ json = dict(method=self.name, id=None, params=list(args))
req = Request.blank(self.parent._url)
- req.method = 'POST'
- req.content_type = 'application/json'
+ req.method = "POST"
+ req.content_type = "application/json"
req.body = dumps(json)
resp = req.get_response(self.parent.proxy)
if resp.status_code != 200 and not (
- resp.status_code == 500
- and resp.content_type == 'application/json'):
+ resp.status_code == 500 and resp.content_type == "application/json"
+ ):
raise ProxyError(
- "Error from JSON-RPC client %s: %s"
- % (self.parent._url, resp.status),
- resp)
+ "Error from JSON-RPC client %s: %s" % (self.parent._url, resp.status),
+ resp,
+ )
json = loads(resp.body)
- if json.get('error') is not None:
+ if json.get("error") is not None:
e = Fault(
- json['error'].get('message'),
- json['error'].get('code'),
- json['error'].get('error'),
- resp)
+ json["error"].get("message"),
+ json["error"].get("code"),
+ json["error"].get("error"),
+ resp,
+ )
raise e
- return json['result']
+ return json["result"]
+
class ProxyError(Exception):
"""
Raised when a request via ServerProxy breaks
"""
+
def __init__(self, message, response):
Exception.__init__(self, message)
self.response = response
+
class Fault(Exception):
"""
Raised when there is a remote error
"""
+
def __init__(self, message, code, error, response):
Exception.__init__(self, message)
self.code = code
self.error = error
self.response = response
+
def __str__(self):
- return 'Method error calling %s: %s\n%s' % (
+ return "Method error calling %s: %s\n%s" % (
self.response.request.url,
self.args[0],
- self.error)
+ self.error,
+ )
+
class DemoObject(object):
"""
Something interesting to attach to
"""
+
def add(self, *args):
return sum(args)
+
def average(self, *args):
return sum(args) / float(len(args))
+
def divide(self, a, b):
return a / b
+
def make_app(expr):
- module, expression = expr.split(':', 1)
+ module, expression = expr.split(":", 1)
__import__(module)
module = sys.modules[module]
obj = eval(expression, module.__dict__)
return JsonRpcApp(obj)
+
def main(args=None):
import optparse
from wsgiref import simple_server
- parser = optparse.OptionParser(
- usage='%prog [OPTIONS] MODULE:EXPRESSION')
+
+ parser = optparse.OptionParser(usage="%prog [OPTIONS] MODULE:EXPRESSION")
parser.add_option(
- '-p', '--port', default='8080',
- help='Port to serve on (default 8080)')
+ "-p", "--port", default="8080", help="Port to serve on (default 8080)"
+ )
parser.add_option(
- '-H', '--host', default='127.0.0.1',
- help='Host to serve on (default localhost; 0.0.0.0 to make public)')
+ "-H",
+ "--host",
+ default="127.0.0.1",
+ help="Host to serve on (default localhost; 0.0.0.0 to make public)",
+ )
options, args = parser.parse_args()
if not args or len(args) > 1:
- print 'You must give a single object reference'
+ print("You must give a single object reference")
parser.print_help()
sys.exit(2)
app = make_app(args[0])
server = simple_server.make_server(options.host, int(options.port), app)
- print 'Serving on http://%s:%s' % (options.host, options.port)
+ print("Serving on http://%s:%s" % (options.host, options.port))
server.serve_forever()
# Try python jsonrpc.py 'jsonrpc:DemoObject()'
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/docs/jsonrpc-example-code/test_jsonrpc.py b/docs/jsonrpc-example-code/test_jsonrpc.py
index a418516..a688f1d 100644
--- a/docs/jsonrpc-example-code/test_jsonrpc.py
+++ b/docs/jsonrpc-example-code/test_jsonrpc.py
@@ -1,3 +1,4 @@
-if __name__ == '__main__':
+if __name__ == "__main__":
import doctest
- doctest.testfile('test_jsonrpc.txt')
+
+ doctest.testfile("test_jsonrpc.txt")
diff --git a/docs/wiki-example-code/example.py b/docs/wiki-example-code/example.py
index 2d452c7..f5edd9f 100644
--- a/docs/wiki-example-code/example.py
+++ b/docs/wiki-example-code/example.py
@@ -4,7 +4,8 @@ from webob import Request, Response
from webob import exc
from tempita import HTMLTemplate
-VIEW_TEMPLATE = HTMLTemplate("""\
+VIEW_TEMPLATE = HTMLTemplate(
+ """\
<html>
<head>
<title>{{page.title}}</title>
@@ -21,9 +22,11 @@ VIEW_TEMPLATE = HTMLTemplate("""\
<a href="{{req.url}}?action=edit">Edit</a>
</body>
</html>
-""")
+"""
+)
-EDIT_TEMPLATE = HTMLTemplate("""\
+EDIT_TEMPLATE = HTMLTemplate(
+ """\
<html>
<head>
<title>Edit: {{page.title}}</title>
@@ -47,7 +50,9 @@ EDIT_TEMPLATE = HTMLTemplate("""\
<a href="{{req.path_url}}">Cancel</a>
</form>
</body></html>
-""")
+"""
+)
+
class WikiApp(object):
@@ -59,67 +64,63 @@ class WikiApp(object):
def __call__(self, environ, start_response):
req = Request(environ)
- action = req.params.get('action', 'view')
+ action = req.params.get("action", "view")
page = self.get_page(req.path_info)
try:
try:
- meth = getattr(self, 'action_%s_%s' % (action, req.method))
+ meth = getattr(self, "action_%s_%s" % (action, req.method))
except AttributeError:
- raise exc.HTTPBadRequest('No such action %r' % action)
+ raise exc.HTTPBadRequest("No such action %r" % action)
resp = meth(req, page)
except exc.HTTPException, e:
resp = e
return resp(environ, start_response)
def get_page(self, path):
- path = path.lstrip('/')
+ path = path.lstrip("/")
if not path:
- path = 'index'
+ path = "index"
path = os.path.join(self.storage_dir, path)
path = os.path.normpath(path)
- if path.endswith('/'):
- path += 'index'
+ if path.endswith("/"):
+ path += "index"
if not path.startswith(self.storage_dir):
raise exc.HTTPBadRequest("Bad path")
- path += '.html'
+ path += ".html"
return Page(path)
def action_view_GET(self, req, page):
if not page.exists:
- return exc.HTTPTemporaryRedirect(
- location=req.url + '?action=edit')
- if req.cookies.get('message'):
- message = req.cookies['message']
+ return exc.HTTPTemporaryRedirect(location=req.url + "?action=edit")
+ if req.cookies.get("message"):
+ message = req.cookies["message"]
else:
message = None
- text = self.view_template.substitute(
- page=page, req=req, message=message)
+ text = self.view_template.substitute(page=page, req=req, message=message)
resp = Response(text)
if message:
- resp.delete_cookie('message')
+ resp.delete_cookie("message")
else:
resp.last_modified = page.mtime
resp.conditional_response = True
return resp
def action_view_POST(self, req, page):
- submit_mtime = int(req.params.get('mtime') or '0') or None
+ submit_mtime = int(req.params.get("mtime") or "0") or None
if page.mtime != submit_mtime:
return exc.HTTPPreconditionFailed(
- "The page has been updated since you started editing it")
- page.set(
- title=req.params['title'],
- content=req.params['content'])
- resp = exc.HTTPSeeOther(
- location=req.path_url)
- resp.set_cookie('message', 'Page updated')
+ "The page has been updated since you started editing it"
+ )
+ page.set(title=req.params["title"], content=req.params["content"])
+ resp = exc.HTTPSeeOther(location=req.path_url)
+ resp.set_cookie("message", "Page updated")
return resp
def action_edit_GET(self, req, page):
- text = self.edit_template.substitute(
- page=page, req=req)
+ text = self.edit_template.substitute(page=page, req=req)
return Response(text)
+
class Page(object):
def __init__(self, filename):
self.filename = filename
@@ -133,15 +134,15 @@ class Page(object):
if not self.exists:
# we need to guess the title
basename = os.path.splitext(os.path.basename(self.filename))[0]
- basename = re.sub(r'[_-]', ' ', basename)
+ basename = re.sub(r"[_-]", " ", basename)
return basename.capitalize()
content = self.full_content
- match = re.search(r'<title>(.*?)</title>', content, re.I|re.S)
+ match = re.search(r"<title>(.*?)</title>", content, re.I | re.S)
return match.group(1)
@property
def full_content(self):
- f = open(self.filename, 'rb')
+ f = open(self.filename, "rb")
try:
return f.read()
finally:
@@ -150,9 +151,9 @@ class Page(object):
@property
def content(self):
if not self.exists:
- return ''
+ return ""
content = self.full_content
- match = re.search(r'<body[^>]*>(.*?)</body>', content, re.I|re.S)
+ match = re.search(r"<body[^>]*>(.*?)</body>", content, re.I | re.S)
return match.group(1)
@property
@@ -166,35 +167,41 @@ class Page(object):
dir = os.path.dirname(self.filename)
if not os.path.exists(dir):
os.makedirs(dir)
- new_content = """<html><head><title>%s</title></head><body>%s</body></html>""" % (
- title, content)
- f = open(self.filename, 'wb')
+ new_content = (
+ """<html><head><title>%s</title></head><body>%s</body></html>"""
+ % (title, content)
+ )
+ f = open(self.filename, "wb")
f.write(new_content)
f.close()
-if __name__ == '__main__':
+
+if __name__ == "__main__":
import optparse
- parser = optparse.OptionParser(
- usage='%prog --port=PORT'
- )
+
+ parser = optparse.OptionParser(usage="%prog --port=PORT")
parser.add_option(
- '-p', '--port',
- default='8080',
- dest='port',
- type='int',
- help='Port to serve on (default 8080)')
+ "-p",
+ "--port",
+ default="8080",
+ dest="port",
+ type="int",
+ help="Port to serve on (default 8080)",
+ )
parser.add_option(
- '--wiki-data',
- default='./wiki',
- dest='wiki_data',
- help='Place to put wiki data into (default ./wiki/)')
+ "--wiki-data",
+ default="./wiki",
+ dest="wiki_data",
+ help="Place to put wiki data into (default ./wiki/)",
+ )
options, args = parser.parse_args()
- print 'Writing wiki pages to %s' % options.wiki_data
+ print("Writing wiki pages to %s" % options.wiki_data)
app = WikiApp(options.wiki_data)
from wsgiref.simple_server import make_server
- httpd = make_server('localhost', options.port, app)
- print 'Serving on http://localhost:%s' % options.port
+
+ httpd = make_server("localhost", options.port, app)
+ print("Serving on http://localhost:%s" % options.port)
try:
httpd.serve_forever()
except KeyboardInterrupt:
- print '^C'
+ print("^C")
diff --git a/pyproject.toml b/pyproject.toml
index 6f26c1c..5bdee1e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,26 +1,27 @@
[build-system]
-requires = ["setuptools", "wheel"]
+requires = ["setuptools >= 41"]
+build-backend = "setuptools.build_meta"
[tool.black]
-py36 = false
+target-version = ['py36', 'py37', 'py38']
exclude = '''
/(
- \.git
- | \.mypy_cache
- | \.tox
- | \.venv
- | \.pytest_cache
- | dist
- | build
- | docs
+ \.git
+ | .tox
)/
'''
-# This next section only exists for people that have their editors
+ # This next section only exists for people that have their editors
# automatically call isort, black already sorts entries on its own when run.
[tool.isort]
+profile = "black"
multi_line_output = 3
+src_paths = ["src", "tests"]
+skip_glob = ["docs/*"]
include_trailing_comma = true
-force_grid_wrap = 0
+force_grid_wrap = false
combine_as_imports = true
line_length = 88
+force_sort_within_sections = true
+default_section = "THIRDPARTY"
+known_first_party = "webob"
diff --git a/setup.py b/setup.py
index e6bfe82..8c81dc7 100644
--- a/setup.py
+++ b/setup.py
@@ -1,6 +1,6 @@
import os
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
here = os.path.abspath(os.path.dirname(__file__))
try:
@@ -22,7 +22,7 @@ docs_extras = ["Sphinx >= 1.7.5", "pylons-sphinx-themes"]
setup(
name="WebOb",
- version="1.9.0dev0",
+ version="2.0.0dev0",
description="WSGI request and response object",
long_description=README + "\n\n" + CHANGES,
classifiers=[
@@ -32,11 +32,11 @@ setup(
"Topic :: Internet :: WWW/HTTP :: WSGI",
"Topic :: Internet :: WWW/HTTP :: WSGI :: Application",
"Topic :: Internet :: WWW/HTTP :: WSGI :: Middleware",
- "Programming Language :: Python :: 2.7",
- "Programming Language :: Python :: 3.4",
- "Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
+ "Programming Language :: Python :: 3.7",
+ "Programming Language :: Python :: 3.8",
+ "Programming Language :: Python :: 3.9",
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
],
@@ -48,7 +48,7 @@ setup(
license="MIT",
packages=find_packages("src", exclude=["tests"]),
package_dir={"": "src"},
- python_requires=">=2.7,!=3.0.*,!=3.1.*,!=3.2.*",
+ python_requires=">=3.6",
zip_safe=True,
extras_require={"testing": testing_extras, "docs": docs_extras},
)
diff --git a/src/webob/__init__.py b/src/webob/__init__.py
index 5a70ea4..627b1b5 100644
--- a/src/webob/__init__.py
+++ b/src/webob/__init__.py
@@ -13,13 +13,12 @@ from webob.datetime_utils import ( # noqa: F401
week,
year,
)
-from webob.request import BaseRequest, LegacyRequest, Request
+from webob.request import BaseRequest, Request
from webob.response import Response
from webob.util import html_escape
__all__ = [
"Request",
- "LegacyRequest",
"Response",
"UTC",
"day",
diff --git a/src/webob/acceptparse.py b/src/webob/acceptparse.py
index 0ac3a5b..ee5e78b 100644
--- a/src/webob/acceptparse.py
+++ b/src/webob/acceptparse.py
@@ -10,7 +10,6 @@ import re
import textwrap
import warnings
-
# RFC 7230 Section 3.2.3 "Whitespace"
# OWS = *( SP / HTAB )
# ; optional whitespace
@@ -113,7 +112,7 @@ class AcceptOffer(namedtuple("AcceptOffer", ["type", "subtype", "params"])):
return Accept._form_media_range(value, self.params)
-class Accept(object):
+class Accept:
"""
Represent an ``Accept`` header.
@@ -1790,7 +1789,7 @@ def accept_property():
return property(fget, fset, fdel, textwrap.dedent(doc))
-class AcceptCharset(object):
+class AcceptCharset:
"""
Represent an ``Accept-Charset`` header.
@@ -2142,7 +2141,7 @@ class AcceptCharsetValidHeader(AcceptCharset):
filtered_offers = []
for index, offer in enumerate(lowercased_offers):
# If offer matches a non-* charset with q=0, it is filtered out
- if any(((offer == charset) for charset in not_acceptable_charsets)):
+ if any((offer == charset) for charset in not_acceptable_charsets):
continue
matched_charset_qvalue = None
@@ -2784,7 +2783,7 @@ def accept_charset_property():
return property(fget, fset, fdel, textwrap.dedent(doc))
-class AcceptEncoding(object):
+class AcceptEncoding:
"""
Represent an ``Accept-Encoding`` header.
@@ -3171,7 +3170,7 @@ class AcceptEncodingValidHeader(AcceptEncoding):
filtered_offers = []
for index, offer in enumerate(lowercased_offers):
# If offer matches a non-* codings with q=0, it is filtered out
- if any(((offer == codings) for codings in not_acceptable_codingss)):
+ if any((offer == codings) for codings in not_acceptable_codingss):
continue
matched_codings_qvalue = None
@@ -3827,7 +3826,7 @@ def accept_encoding_property():
return property(fget, fset, fdel, textwrap.dedent(doc))
-class AcceptLanguage(object):
+class AcceptLanguage:
"""
Represent an ``Accept-Language`` header.
@@ -4308,7 +4307,7 @@ class AcceptLanguageValidHeader(AcceptLanguage):
filtered_tags = []
for index, tag in enumerate(lowercased_tags):
# If tag matches a non-* range with q=0, it is filtered out
- if any((match(tag=tag, range_=range_) for range_ in not_acceptable_ranges)):
+ if any(match(tag=tag, range_=range_) for range_ in not_acceptable_ranges):
continue
matched_range_qvalue = None
diff --git a/src/webob/byterange.py b/src/webob/byterange.py
index 435282b..44db453 100644
--- a/src/webob/byterange.py
+++ b/src/webob/byterange.py
@@ -6,7 +6,7 @@ _rx_range = re.compile(r"bytes *= *(\d*) *- *(\d*)", flags=re.I)
_rx_content_range = re.compile(r"bytes (?:(\d+)-(\d+)|[*])/(?:(\d+)|[*])")
-class Range(object):
+class Range:
"""
Represents the Range header.
"""
@@ -85,7 +85,7 @@ class Range(object):
return cls(start, end)
-class ContentRange(object):
+class ContentRange:
"""
Represents the Content-Range header
diff --git a/src/webob/cachecontrol.py b/src/webob/cachecontrol.py
index 1278329..1029344 100644
--- a/src/webob/cachecontrol.py
+++ b/src/webob/cachecontrol.py
@@ -63,7 +63,7 @@ token_re = re.compile(r'([a-zA-Z][a-zA-Z_-]*)\s*(?:=(?:"([^"]*)"|([^ \t",;]*)))?
need_quote_re = re.compile(r"[^a-zA-Z0-9._-]")
-class exists_property(object):
+class exists_property:
"""
Represents a property that either is listed in the Cache-Control
header, or is not listed (has no value)
@@ -95,7 +95,7 @@ class exists_property(object):
self.__set__(obj, False)
-class value_property(object):
+class value_property:
"""
Represents a property that has a value in the Cache-Control header.
@@ -139,7 +139,7 @@ class value_property(object):
del obj.properties[self.prop]
-class CacheControl(object):
+class CacheControl:
"""
Represents the Cache-Control header.
diff --git a/src/webob/client.py b/src/webob/client.py
index 0ba9654..980137a 100644
--- a/src/webob/client.py
+++ b/src/webob/client.py
@@ -1,15 +1,16 @@
import errno
-import sys
import re
+import sys
try:
import httplib
except ImportError:
import http.client as httplib
-from webob.compat import url_quote
+
import socket
+from urllib.parse import quote as url_quote
+
from webob import exc
-from webob.compat import PY2
__all__ = ["send_request_app", "SendRequest"]
@@ -43,16 +44,20 @@ class SendRequest:
def __call__(self, environ, start_response):
scheme = environ["wsgi.url_scheme"]
+
if scheme == "http":
ConnClass = self.HTTPConnection
elif scheme == "https":
ConnClass = self.HTTPSConnection
else:
raise ValueError("Unknown scheme: %r" % scheme)
+
if "SERVER_NAME" not in environ:
host = environ.get("HTTP_HOST")
+
if not host:
raise ValueError("environ contains neither SERVER_NAME nor HTTP_HOST")
+
if ":" in host:
host, port = host.split(":", 1)
else:
@@ -63,10 +68,12 @@ class SendRequest:
environ["SERVER_NAME"] = host
environ["SERVER_PORT"] = port
kw = {}
+
if "webob.client.timeout" in environ and self._timeout_supported(ConnClass):
kw["timeout"] = environ["webob.client.timeout"]
conn = ConnClass("%(SERVER_NAME)s:%(SERVER_PORT)s" % environ, **kw)
headers = {}
+
for key, value in environ.items():
if key.startswith("HTTP_"):
key = key[5:].replace("_", "-").title()
@@ -74,6 +81,7 @@ class SendRequest:
path = url_quote(environ.get("SCRIPT_NAME", "")) + url_quote(
environ.get("PATH_INFO", "")
)
+
if environ.get("QUERY_STRING"):
path += "?" + environ["QUERY_STRING"]
try:
@@ -82,13 +90,16 @@ class SendRequest:
content_length = 0
# FIXME: there is no streaming of the body, and that might be useful
# in some cases
+
if content_length:
body = environ["wsgi.input"].read(content_length)
else:
body = ""
headers["Content-Length"] = content_length
+
if environ.get("CONTENT_TYPE"):
headers["Content-Type"] = environ["CONTENT_TYPE"]
+
if not path.startswith("/"):
path = "/" + path
try:
@@ -96,8 +107,9 @@ class SendRequest:
res = conn.getresponse()
except socket.timeout:
resp = exc.HTTPGatewayTimeout()
+
return resp(environ, start_response)
- except (socket.error, socket.gaierror) as e:
+ except (OSError, socket.gaierror) as e:
if (isinstance(e, socket.error) and e.args[0] == -2) or (
isinstance(e, socket.gaierror) and e.args[0] == 8
):
@@ -106,10 +118,12 @@ class SendRequest:
"Name or service not known (bad domain name: %s)"
% environ["SERVER_NAME"]
)
+
return resp(environ, start_response)
elif e.args[0] in _e_refused: # pragma: no cover
# Connection refused
resp = exc.HTTPBadGateway("Connection refused")
+
return resp(environ, start_response)
raise
headers_out = self.parse_headers(res.msg)
@@ -117,11 +131,13 @@ class SendRequest:
start_response(status, headers_out)
length = res.getheader("content-length")
# FIXME: This shouldn't really read in all the content at once
+
if length is not None:
body = res.read(int(length))
else:
body = res.read()
conn.close()
+
return [body]
# Remove these headers from response (specify lower case header
@@ -135,16 +151,17 @@ class SendRequest:
Turn a Message object into a list of WSGI-style headers.
"""
headers_out = []
- if not PY2:
- headers = message._headers
- else:
- headers = message.headers
+ headers = message._headers
+
for full_header in headers:
if not full_header: # pragma: no cover
# Shouldn't happen, but we'll just ignore
+
continue
+
if full_header[0].isspace(): # pragma: no cover
# Continuation line, add to the last header
+
if not headers_out:
raise ValueError(
"First header starts with a space (%r)" % full_header
@@ -152,7 +169,9 @@ class SendRequest:
last_header, last_value = headers_out.pop()
value = last_value + ", " + full_header.strip()
headers_out.append((last_header, value))
+
continue
+
if isinstance(full_header, tuple): # pragma: no cover
header, value = full_header
else: # pragma: no cover
@@ -161,12 +180,15 @@ class SendRequest:
except Exception:
raise ValueError("Invalid header: %r" % (full_header,))
value = value.strip()
+
if "\n" in value or "\r\n" in value: # pragma: no cover
# Python 3 has multiline values for continuations, Python 2
# has two items in headers
value = self.MULTILINE_RE.sub(", ", value)
+
if header.lower() not in self.filtered_headers:
headers_out.append((header, value))
+
return headers_out
def _timeout_supported(self, ConnClass):
@@ -175,11 +197,13 @@ class SendRequest:
httplib.HTTPSConnection,
): # pragma: no cover
return False
+
return True
send_request_app = SendRequest()
_e_refused = (errno.ECONNREFUSED,)
+
if hasattr(errno, "ENODATA"): # pragma: no cover
_e_refused += (errno.ENODATA,)
diff --git a/src/webob/compat.py b/src/webob/compat.py
index 9d343a0..735d510 100644
--- a/src/webob/compat.py
+++ b/src/webob/compat.py
@@ -1,253 +1,117 @@
-# code stolen from "six"
-
# flake8: noqa
+import cgi
+from cgi import FieldStorage as _cgi_FieldStorage, parse_header
+from html import escape
+from queue import Empty, Queue
import sys
+import tempfile
import types
-from cgi import parse_header
-
-# True if we are running on Python 3.
-PY3 = sys.version_info[0] == 3
-PY2 = sys.version_info[0] == 2
-
-if PY3:
- string_types = (str,)
- integer_types = (int,)
- class_types = (type,)
- text_type = str
- long = int
-else:
- string_types = (basestring,)
- integer_types = (int, long)
- class_types = (type, types.ClassType)
- text_type = unicode
- long = long
-
-# TODO check if errors is ever used
-
-def text_(s, encoding="latin-1", errors="strict"):
- if isinstance(s, bytes):
- return s.decode(encoding, errors)
- return s
+# Various different FieldStorage work-arounds required on Python 3.x
+class cgi_FieldStorage(_cgi_FieldStorage): # pragma: no cover
+ def __repr__(self):
+ """monkey patch for FieldStorage.__repr__
-def bytes_(s, encoding="latin-1", errors="strict"):
- if isinstance(s, text_type):
- return s.encode(encoding, errors)
- return s
+ Unbelievably, the default __repr__ on FieldStorage reads
+ the entire file content instead of being sane about it.
+ This is a simple replacement that doesn't do that
+ """
+ if self.file:
+ return "FieldStorage(%r, %r)" % (self.name, self.filename)
-if PY3:
+ return "FieldStorage(%r, %r, %r)" % (self.name, self.filename, self.value)
- def native_(s, encoding="latin-1", errors="strict"):
- if isinstance(s, text_type):
- return s
- return str(s, encoding, errors)
+ # Work around https://bugs.python.org/issue27777
+ def make_file(self):
+ if self._binary_file or self.length >= 0:
+ return tempfile.TemporaryFile("wb+")
+ else:
+ return tempfile.TemporaryFile("w+", encoding=self.encoding, newline="\n")
+ # Work around http://bugs.python.org/issue23801
+ # This is taken exactly from Python 3.5's cgi.py module
+ def read_multi(self, environ, keep_blank_values, strict_parsing):
+ """Internal: read a part that is itself multipart."""
+ ib = self.innerboundary
-else:
+ if not cgi.valid_boundary(ib):
+ raise ValueError("Invalid boundary in multipart form: %r" % (ib,))
+ self.list = []
- def native_(s, encoding="latin-1", errors="strict"):
- if isinstance(s, text_type):
- return s.encode(encoding, errors)
- return str(s)
+ if self.qs_on_post:
+ query = cgi.urllib.parse.parse_qsl(
+ self.qs_on_post,
+ self.keep_blank_values,
+ self.strict_parsing,
+ encoding=self.encoding,
+ errors=self.errors,
+ )
+ for key, value in query:
+ self.list.append(cgi.MiniFieldStorage(key, value))
-try:
- from queue import Queue, Empty
-except ImportError:
- from Queue import Queue, Empty
+ klass = self.FieldStorageClass or self.__class__
+ first_line = self.fp.readline() # bytes
-try:
- from collections.abc import MutableMapping
- from collections.abc import Iterable
-except ImportError:
- from collections import MutableMapping
- from collections import Iterable
+ if not isinstance(first_line, bytes):
+ raise ValueError(
+ "%s should return bytes, got %s" % (self.fp, type(first_line).__name__)
+ )
+ self.bytes_read += len(first_line)
-if PY3:
- from urllib import parse
+ # Ensure that we consume the file until we've hit our innerboundary
- urlparse = parse
- from urllib.parse import quote as url_quote
- from urllib.parse import urlencode as url_encode, quote_plus
- from urllib.request import urlopen as url_open
-else:
- import urlparse
- from urllib import quote_plus
- from urllib import quote as url_quote
- from urllib import unquote as url_unquote
- from urllib import urlencode as url_encode
- from urllib2 import urlopen as url_open
-
-if PY3: # pragma: no cover
-
- def reraise(exc_info):
- etype, exc, tb = exc_info
- if exc.__traceback__ is not tb:
- raise exc.with_traceback(tb)
- raise exc
-
-
-else:
- exec("def reraise(exc): raise exc[0], exc[1], exc[2]")
-
-
-if PY3:
-
- def iteritems_(d):
- return d.items()
-
- def itervalues_(d):
- return d.values()
-
-
-else:
-
- def iteritems_(d):
- return d.iteritems() # noqa: B301
-
- def itervalues_(d):
- return d.itervalues() # noqa: B301
-
-
-if PY3: # pragma: no cover
-
- def unquote(string):
- if not string:
- return b""
- res = string.split(b"%")
- if len(res) != 1:
- string = res[0]
- for item in res[1:]:
- try:
- string += bytes([int(item[:2], 16)]) + item[2:]
- except ValueError:
- string += b"%" + item
- return string
-
- def url_unquote(s):
- return unquote(s.encode("ascii")).decode("latin-1")
-
- def parse_qsl_text(qs, encoding="utf-8"):
- qs = qs.encode("latin-1")
- qs = qs.replace(b"+", b" ")
- pairs = [s2 for s1 in qs.split(b"&") for s2 in s1.split(b";") if s2]
- for name_value in pairs:
- nv = name_value.split(b"=", 1)
- if len(nv) != 2:
- nv.append("")
- name = unquote(nv[0])
- value = unquote(nv[1])
- yield (name.decode(encoding), value.decode(encoding))
-
-
-else:
- from urlparse import parse_qsl
-
- def parse_qsl_text(qs, encoding="utf-8"):
- qsl = parse_qsl(qs, keep_blank_values=True, strict_parsing=False)
- for (x, y) in qsl:
- yield (x.decode(encoding), y.decode(encoding))
-
-
-if PY3:
- from html import escape
-else:
- from cgi import escape
-
-
-if PY3:
- import cgi
- import tempfile
- from cgi import FieldStorage as _cgi_FieldStorage
-
- # Various different FieldStorage work-arounds required on Python 3.x
- class cgi_FieldStorage(_cgi_FieldStorage): # pragma: no cover
-
- # Work around https://bugs.python.org/issue27777
- def make_file(self):
- if self._binary_file or self.length >= 0:
- return tempfile.TemporaryFile("wb+")
- else:
- return tempfile.TemporaryFile(
- "w+", encoding=self.encoding, newline="\n"
- )
-
- # Work around http://bugs.python.org/issue23801
- # This is taken exactly from Python 3.5's cgi.py module
- def read_multi(self, environ, keep_blank_values, strict_parsing):
- """Internal: read a part that is itself multipart."""
- ib = self.innerboundary
- if not cgi.valid_boundary(ib):
- raise ValueError("Invalid boundary in multipart form: %r" % (ib,))
- self.list = []
- if self.qs_on_post:
- query = cgi.urllib.parse.parse_qsl(
- self.qs_on_post,
- self.keep_blank_values,
- self.strict_parsing,
- encoding=self.encoding,
- errors=self.errors,
- )
- for key, value in query:
- self.list.append(cgi.MiniFieldStorage(key, value))
-
- klass = self.FieldStorageClass or self.__class__
- first_line = self.fp.readline() # bytes
- if not isinstance(first_line, bytes):
- raise ValueError(
- "%s should return bytes, got %s"
- % (self.fp, type(first_line).__name__)
- )
+ while first_line.strip() != (b"--" + self.innerboundary) and first_line:
+ first_line = self.fp.readline()
self.bytes_read += len(first_line)
- # Ensure that we consume the file until we've hit our innerboundary
- while first_line.strip() != (b"--" + self.innerboundary) and first_line:
- first_line = self.fp.readline()
- self.bytes_read += len(first_line)
+ while True:
+ parser = cgi.FeedParser()
+ hdr_text = b""
while True:
- parser = cgi.FeedParser()
- hdr_text = b""
- while True:
- data = self.fp.readline()
- hdr_text += data
- if not data.strip():
- break
- if not hdr_text:
- break
- # parser takes strings, not bytes
- self.bytes_read += len(hdr_text)
- parser.feed(hdr_text.decode(self.encoding, self.errors))
- headers = parser.close()
- # Some clients add Content-Length for part headers, ignore them
- if "content-length" in headers:
- filename = None
- if "content-disposition" in self.headers:
- cdisp, pdict = parse_header(self.headers["content-disposition"])
- if "filename" in pdict:
- filename = pdict["filename"]
- if filename is None:
- del headers["content-length"]
- part = klass(
- self.fp,
- headers,
- ib,
- environ,
- keep_blank_values,
- strict_parsing,
- self.limit - self.bytes_read,
- self.encoding,
- self.errors,
- )
- self.bytes_read += part.bytes_read
- self.list.append(part)
- if part.done or self.bytes_read >= self.length > 0:
- break
- self.skip_lines()
+ data = self.fp.readline()
+ hdr_text += data
+ if not data.strip():
+ break
-else:
- from cgi import FieldStorage as cgi_FieldStorage
+ if not hdr_text:
+ break
+ # parser takes strings, not bytes
+ self.bytes_read += len(hdr_text)
+ parser.feed(hdr_text.decode(self.encoding, self.errors))
+ headers = parser.close()
+ # Some clients add Content-Length for part headers, ignore them
+
+ if "content-length" in headers:
+ filename = None
+
+ if "content-disposition" in self.headers:
+ cdisp, pdict = parse_header(self.headers["content-disposition"])
+
+ if "filename" in pdict:
+ filename = pdict["filename"]
+
+ if filename is None:
+ del headers["content-length"]
+ part = klass(
+ self.fp,
+ headers,
+ ib,
+ environ,
+ keep_blank_values,
+ strict_parsing,
+ self.limit - self.bytes_read,
+ self.encoding,
+ self.errors,
+ )
+ self.bytes_read += part.bytes_read
+ self.list.append(part)
+
+ if part.done or self.bytes_read >= self.length > 0:
+ break
+ self.skip_lines()
diff --git a/src/webob/cookies.py b/src/webob/cookies.py
index 81bf7a3..8244846 100644
--- a/src/webob/cookies.py
+++ b/src/webob/cookies.py
@@ -1,25 +1,16 @@
import base64
import binascii
+from collections.abc import MutableMapping
+from datetime import date, datetime, timedelta
import hashlib
import hmac
import json
-from datetime import date, datetime, timedelta
import re
import string
import time
import warnings
-from webob.compat import (
- MutableMapping,
- PY2,
- text_type,
- bytes_,
- text_,
- native_,
- string_types,
-)
-
-from webob.util import strings_differ
+from webob.util import bytes_, text_
__all__ = [
"Cookie",
@@ -57,7 +48,7 @@ class RequestCookies(MutableMapping):
def d(b):
return b.decode("utf8")
- cache = dict((d(k), d(v)) for k, v in parse_cookie(header))
+ cache = {d(k): d(v) for k, v in parse_cookie(header)}
env[self._cache_key] = (cache, header)
return cache
@@ -67,8 +58,7 @@ class RequestCookies(MutableMapping):
had_header = header is not None
header = header or ""
- if not PY2:
- header = header.encode("latin-1")
+ header = header.encode("latin-1")
bytes_name = bytes_(name, "ascii")
if value is None:
@@ -100,18 +90,16 @@ class RequestCookies(MutableMapping):
header = replacement
if header:
- self._environ["HTTP_COOKIE"] = native_(header, "latin-1")
+ self._environ["HTTP_COOKIE"] = text_(header, "latin-1")
elif had_header:
self._environ["HTTP_COOKIE"] = ""
return found
def _valid_cookie_name(self, name):
- if not isinstance(name, string_types):
+ if not isinstance(name, str):
raise TypeError(name, "cookie name must be a string")
- if not isinstance(name, text_type):
- name = text_(name, "utf-8")
try:
bytes_cookie_name = bytes_(name, "ascii")
except UnicodeEncodeError:
@@ -125,14 +113,9 @@ class RequestCookies(MutableMapping):
def __setitem__(self, name, value):
name = self._valid_cookie_name(name)
- if not isinstance(value, string_types):
+ if not isinstance(value, str):
raise ValueError(value, "cookie value must be a string")
- if not isinstance(value, text_type):
- try:
- value = text_(value, "utf-8")
- except UnicodeDecodeError:
- raise ValueError(value, "cookie value must be utf-8 binary or unicode")
self._mutate_header(name, value)
def __getitem__(self, name):
@@ -157,17 +140,6 @@ class RequestCookies(MutableMapping):
def items(self):
return self._cache.items()
- if PY2:
-
- def iterkeys(self):
- return self._cache.iterkeys() # noqa: B301
-
- def itervalues(self):
- return self._cache.itervalues() # noqa: B301
-
- def iteritems(self):
- return self._cache.iteritems() # noqa: B301
-
def __contains__(self, name):
return name in self._cache
@@ -227,8 +199,7 @@ class Cookie(dict):
def _parse_cookie(data):
- if not PY2:
- data = data.encode("latin-1")
+ data = data.encode("latin-1")
for key, val in _rx_cookie.findall(data):
yield key, _unquote(val)
@@ -263,7 +234,7 @@ def serialize_cookie_date(v):
return None
elif isinstance(v, bytes):
return v
- elif isinstance(v, text_type):
+ elif isinstance(v, str):
return v.encode("ascii")
elif isinstance(v, int):
v = timedelta(seconds=v)
@@ -345,15 +316,15 @@ class Morsel(dict):
)
add(b"SameSite=" + self.samesite)
- return native_(b"; ".join(result), "ascii")
+ return text_(b"; ".join(result), "ascii")
__str__ = serialize
def __repr__(self):
return "<%s: %s=%r>" % (
self.__class__.__name__,
- native_(self.name),
- native_(self.value),
+ text_(self.name),
+ text_(self.value),
)
@@ -375,12 +346,16 @@ _re_cookie_str = _re_cookie_str_key + _re_cookie_str_equal + _re_cookie_str_val
_rx_cookie = re.compile(bytes_(_re_cookie_str, "ascii"))
_rx_unquote = re.compile(bytes_(r"\\([0-3][0-7][0-7]|.)", "ascii"))
-_bchr = chr if PY2 else (lambda i: bytes([i]))
-_ch_unquote_map = dict((bytes_("%03o" % i), _bchr(i)) for i in range(256))
+
+def _bchr(i):
+ return bytes([i])
+
+
+_ch_unquote_map = {bytes_("%03o" % i): _bchr(i) for i in range(256)}
_ch_unquote_map.update((v, v) for v in list(_ch_unquote_map.values()))
-_b_dollar_sign = "$" if PY2 else ord("$")
-_b_quote_mark = '"' if PY2 else ord('"')
+_b_dollar_sign = ord("$")
+_b_quote_mark = ord('"')
def _unquote(v):
@@ -429,12 +404,10 @@ _valid_token_bytes = bytes_(_valid_token_chars)
# this is a map used to escape the values
_escape_noop_chars = _allowed_cookie_chars + " "
-_escape_map = dict((chr(i), "\\%03o" % i) for i in range(256))
+_escape_map = {chr(i): "\\%03o" % i for i in range(256)}
_escape_map.update(zip(_escape_noop_chars, _escape_noop_chars))
-if not PY2:
- # convert to {int -> bytes}
- _escape_map = dict((ord(k), bytes_(v, "ascii")) for k, v in _escape_map.items())
+_escape_map = {ord(k): bytes_(v, "ascii") for k, v in _escape_map.items()}
_escape_char = _escape_map.__getitem__
weekdays = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
@@ -640,7 +613,7 @@ def make_cookie(
return morsel.serialize()
-class JSONSerializer(object):
+class JSONSerializer:
""" A serializer which uses `json.dumps`` and ``json.loads``"""
def dumps(self, appstruct):
@@ -653,7 +626,7 @@ class JSONSerializer(object):
return json.loads(text_(bstruct, encoding="utf-8"))
-class Base64Serializer(object):
+class Base64Serializer:
""" A serializer which uses base64 to encode/decode data"""
def __init__(self, serializer=None):
@@ -687,7 +660,7 @@ class Base64Serializer(object):
return self.serializer.loads(cstruct)
-class SignedSerializer(object):
+class SignedSerializer:
"""
A helper to cryptographically sign arbitrary content using HMAC.
@@ -764,7 +737,7 @@ class SignedSerializer(object):
sig = hmac.new(self.salted_secret, bytes_(cstruct), self.digestmod).digest()
- if strings_differ(sig, expected_sig):
+ if not hmac.compare_digest(sig, expected_sig):
raise ValueError("Invalid signature")
return self.serializer.loads(cstruct)
@@ -773,7 +746,7 @@ class SignedSerializer(object):
_default = object()
-class CookieProfile(object):
+class CookieProfile:
"""
A helper class that helps bring some sanity to the insanity that is cookie
handling.
diff --git a/src/webob/datetime_utils.py b/src/webob/datetime_utils.py
index 8cbedbe..b9bbb43 100644
--- a/src/webob/datetime_utils.py
+++ b/src/webob/datetime_utils.py
@@ -1,12 +1,9 @@
import calendar
-
from datetime import date, datetime, timedelta, tzinfo
-
from email.utils import formatdate, mktime_tz, parsedate_tz
-
import time
-from webob.compat import integer_types, long, native_, text_type
+from webob.util import text_
__all__ = [
"UTC",
@@ -48,6 +45,7 @@ def timedelta_to_seconds(td):
"""
Converts a timedelta instance to seconds.
"""
+
return td.seconds + (td.days * 24 * 60 * 60)
@@ -65,34 +63,41 @@ def parse_date(value):
if not value:
return None
try:
- value = native_(value)
+ if not isinstance(value, str):
+ value = str(value, "latin-1")
except Exception:
return None
t = parsedate_tz(value)
+
if t is None:
# Could not parse
+
return None
- if t[-1] is None:
- # No timezone given. None would mean local time, but we'll force UTC
- t = t[:9] + (0,)
+
t = mktime_tz(t)
+
return datetime.fromtimestamp(t, UTC)
def serialize_date(dt):
- if isinstance(dt, (bytes, text_type)):
- return native_(dt)
+ if isinstance(dt, (bytes, str)):
+ return text_(dt)
+
if isinstance(dt, timedelta):
dt = _now() + dt
+
if isinstance(dt, (datetime, date)):
dt = dt.timetuple()
+
if isinstance(dt, (tuple, time.struct_time)):
dt = calendar.timegm(dt)
- if not (isinstance(dt, float) or isinstance(dt, integer_types)):
+
+ if not (isinstance(dt, float) or isinstance(dt, int)):
raise ValueError(
"You must pass in a datetime, date, time tuple, or integer object, "
"not %r" % dt
)
+
return formatdate(dt, usegmt=True)
@@ -100,6 +105,7 @@ def parse_date_delta(value):
"""
like parse_date, but also handle delta seconds
"""
+
if not value:
return None
try:
@@ -111,7 +117,7 @@ def parse_date_delta(value):
def serialize_date_delta(value):
- if isinstance(value, (float, int, long)):
+ if isinstance(value, (float, int, int)):
return str(int(value))
else:
return serialize_date(value)
diff --git a/src/webob/dec.py b/src/webob/dec.py
index def23db..02ec255 100644
--- a/src/webob/dec.py
+++ b/src/webob/dec.py
@@ -6,15 +6,14 @@ application (while also allowing normal calling of the method with an
instantiated request).
"""
-from webob.compat import bytes_, text_type
-
-from webob.request import Request
from webob.exc import HTTPException
+from webob.request import Request
+from webob.util import bytes_
__all__ = ["wsgify"]
-class wsgify(object):
+class wsgify:
"""Turns a request-taking, response-returning function into a WSGI
app
@@ -84,9 +83,11 @@ class wsgify(object):
self, func=None, RequestClass=None, args=(), kwargs=None, middleware_wraps=None
):
self.func = func
+
if RequestClass is not None and RequestClass is not self.RequestClass:
self.RequestClass = RequestClass
self.args = tuple(args)
+
if kwargs is None:
kwargs = {}
self.kwargs = kwargs
@@ -97,6 +98,7 @@ class wsgify(object):
def __get__(self, obj, type=None):
# This handles wrapping methods
+
if hasattr(self.func, "__get__"):
return self.clone(self.func.__get__(obj, type))
else:
@@ -105,6 +107,7 @@ class wsgify(object):
def __call__(self, req, *args, **kw):
"""Call this as a WSGI application or with a request"""
func = self.func
+
if func is None:
if args or kw:
raise TypeError(
@@ -112,7 +115,9 @@ class wsgify(object):
"will wrap" % self.__class__.__name__
)
func = req
+
return self.clone(func)
+
if isinstance(req, dict):
if len(args) != 1 or kw:
raise TypeError(
@@ -127,20 +132,26 @@ class wsgify(object):
resp = self.call_func(req, *args, **kw)
except HTTPException as exc:
resp = exc
+
if resp is None:
# FIXME: I'm not sure what this should be?
resp = req.response
- if isinstance(resp, text_type):
+
+ if isinstance(resp, str):
resp = bytes_(resp, req.charset)
+
if isinstance(resp, bytes):
body = resp
resp = req.response
resp.write(body)
+
if resp is not req.response:
resp = req.response.merge_cookies(resp)
+
return resp(environ, start_response)
else:
args, kw = self._prepare_args(args, kw)
+
return self.call_func(req, *args, **kw)
def get(self, url, **kw):
@@ -156,6 +167,7 @@ class wsgify(object):
"""
kw.setdefault("method", "GET")
req = self.RequestClass.blank(url, **kw)
+
return self(req)
def post(self, url, POST=None, **kw):
@@ -173,6 +185,7 @@ class wsgify(object):
"""
kw.setdefault("method", "POST")
req = self.RequestClass.blank(url, POST=POST, **kw)
+
return self(req)
def request(self, url, **kw):
@@ -183,11 +196,13 @@ class wsgify(object):
resp = myapp.request('/article/1', method='PUT', body='New article')
"""
req = self.RequestClass.blank(url, **kw)
+
return self(req)
def call_func(self, req, *args, **kwargs):
"""Call the wrapped function; override this in a subclass to
change how the function is called."""
+
return self.func(req, *args, **kwargs)
def clone(self, func=None, **kw):
@@ -195,15 +210,20 @@ class wsgify(object):
parameters rebound
"""
kwargs = {}
+
if func is not None:
kwargs["func"] = func
+
if self.RequestClass is not self.__class__.RequestClass:
kwargs["RequestClass"] = self.RequestClass
+
if self.args:
kwargs["args"] = self.args
+
if self.kwargs:
kwargs["kwargs"] = self.kwargs
kwargs.update(kw)
+
return self.__class__(**kwargs)
# To match @decorator:
@@ -256,21 +276,26 @@ class wsgify(object):
binding the application).
"""
+
if middle_func is None:
return _UnboundMiddleware(cls, app, kw)
+
if app is None:
return _MiddlewareFactory(cls, middle_func, kw)
+
return cls(middle_func, middleware_wraps=app, kwargs=kw)
def _prepare_args(self, args, kwargs):
args = args or self.args
kwargs = kwargs or self.kwargs
+
if self.middleware_wraps:
args = (self.middleware_wraps,) + args
+
return args, kwargs
-class _UnboundMiddleware(object):
+class _UnboundMiddleware:
"""A `wsgify.middleware` invocation that has not yet wrapped a
middleware function; the intermediate object when you do
something like ``@wsgify.middleware(RequestClass=Foo)``
@@ -287,10 +312,11 @@ class _UnboundMiddleware(object):
def __call__(self, func, app=None):
if app is None:
app = self.app
+
return self.wrapper_class.middleware(func, app=app, **self.kw)
-class _MiddlewareFactory(object):
+class _MiddlewareFactory:
"""A middleware that has not yet been bound to an application or
configured.
"""
@@ -310,4 +336,5 @@ class _MiddlewareFactory(object):
def __call__(self, app=None, **config):
kw = self.kw.copy()
kw.update(config)
+
return self.wrapper_class.middleware(self.middleware, app, **kw)
diff --git a/src/webob/descriptors.py b/src/webob/descriptors.py
index ede7436..e24b4f4 100644
--- a/src/webob/descriptors.py
+++ b/src/webob/descriptors.py
@@ -1,18 +1,11 @@
-import re
-
-from datetime import date, datetime
-
from collections import namedtuple
+from datetime import date, datetime
+import re
from webob.byterange import ContentRange, Range
-
-from webob.compat import PY2, text_type
-
from webob.datetime_utils import parse_date, serialize_date
-
from webob.util import header_docstring, warn_deprecation
-
CHARSET_RE = re.compile(r";\s*charset=([^;]*)", re.I)
SCHEME_RE = re.compile(r"^[a-z]+:", re.I)
@@ -85,27 +78,13 @@ def environ_decoder(key, default=_not_given, rfc_section=None, encattr=None):
def upath_property(key):
- if PY2:
+ def fget(req):
+ encoding = req.url_encoding
+ return req.environ.get(key, "").encode("latin-1").decode(encoding)
- def fget(req):
- encoding = req.url_encoding
- return req.environ.get(key, "").decode(encoding)
-
- def fset(req, val):
- encoding = req.url_encoding
- if isinstance(val, text_type):
- val = val.encode(encoding)
- req.environ[key] = val
-
- else:
-
- def fget(req):
- encoding = req.url_encoding
- return req.environ.get(key, "").encode("latin-1").decode(encoding)
-
- def fset(req, val):
- encoding = req.url_encoding
- req.environ[key] = val.encode(encoding).decode("latin-1")
+ def fset(req, val):
+ encoding = req.url_encoding
+ req.environ[key] = val.encode(encoding).decode("latin-1")
return property(fget, fset, doc="upath_property(%r)" % key)
@@ -147,11 +126,11 @@ def header_getter(header, rfc_section):
def fset(r, value):
fdel(r)
if value is not None:
+ if not isinstance(value, str):
+ raise ValueError("Value must be text_type")
if "\n" in value or "\r" in value:
raise ValueError("Header value may not contain control characters")
- if isinstance(value, text_type) and PY2:
- value = value.encode("latin-1")
r._headerlist.append((header, value))
def fdel(r):
@@ -193,7 +172,7 @@ def parse_list(value):
def serialize_list(value):
- if isinstance(value, (text_type, bytes)):
+ if isinstance(value, (str, bytes)):
return str(value)
else:
return ", ".join(map(str, value))
diff --git a/src/webob/etag.py b/src/webob/etag.py
index e0ee8d5..6e91d45 100644
--- a/src/webob/etag.py
+++ b/src/webob/etag.py
@@ -34,7 +34,7 @@ def etag_property(key, default, rfc_section, strong=True):
return property(fget, fset, fdel, doc=doc)
-class _AnyETag(object):
+class _AnyETag:
"""
Represents an ETag of *, or a missing ETag when matching is 'safe'
"""
@@ -57,7 +57,7 @@ class _AnyETag(object):
AnyETag = _AnyETag()
-class _NoETag(object):
+class _NoETag:
"""
Represents a missing ETag when matching is unsafe
"""
@@ -83,7 +83,7 @@ NoETag = _NoETag()
# TODO: convert into a simple tuple
-class ETagMatcher(object):
+class ETagMatcher:
def __init__(self, etags):
self.etags = etags
@@ -114,7 +114,7 @@ class ETagMatcher(object):
return ", ".join(map('"%s"'.__mod__, self.etags))
-class IfRange(object):
+class IfRange:
def __init__(self, etag):
self.etag = etag
@@ -149,7 +149,7 @@ class IfRange(object):
__bool__ = __nonzero__ # python 3
-class IfRangeDate(object):
+class IfRangeDate:
def __init__(self, date):
self.date = date
diff --git a/src/webob/exc.py b/src/webob/exc.py
index d6ae1ec..0893223 100644
--- a/src/webob/exc.py
+++ b/src/webob/exc.py
@@ -166,22 +166,22 @@ References:
"""
import json
-from string import Template
import re
+from string import Template
import sys
+from urllib import parse as urlparse
from webob.acceptparse import create_accept_header
-from webob.compat import class_types, text_, text_type, urlparse
from webob.request import Request
from webob.response import Response
-from webob.util import html_escape
+from webob.util import html_escape, text_
tag_re = re.compile(r"<.*?>", re.S)
br_re = re.compile(r"<br.*?>", re.I | re.S)
comment_re = re.compile(r"<!--|-->")
-class _lazified(object):
+class _lazified:
def __init__(self, func, value):
self.func = func
self.value = value
@@ -201,14 +201,11 @@ def no_escape(value):
if value is None:
return ""
- if not isinstance(value, text_type):
- if hasattr(value, "__unicode__"):
- value = value.__unicode__()
-
+ if not isinstance(value, str):
if isinstance(value, bytes):
value = text_(value, "utf-8")
else:
- value = text_type(value)
+ value = str(value)
return value
@@ -280,7 +277,7 @@ ${body}"""
comment=None,
body_template=None,
json_formatter=None,
- **kw
+ **kw,
):
Response.__init__(self, status="%s %s" % (self.code, self.title), **kw)
Exception.__init__(self, detail)
@@ -564,7 +561,7 @@ ${html_comment}"""
location=None,
add_slash=False,
):
- super(_HTTPMove, self).__init__(
+ super().__init__(
detail=detail, headers=headers, comment=comment, body_template=body_template
)
@@ -593,7 +590,7 @@ ${html_comment}"""
self.location = url
self.location = urlparse.urljoin(req.path_url, self.location)
- return super(_HTTPMove, self).__call__(environ, start_response)
+ return super().__call__(environ, start_response)
class HTTPMultipleChoices(_HTTPMove):
@@ -1306,7 +1303,7 @@ class HTTPNetworkAuthenticationRequired(HTTPServerError):
explanation = "Network authentication is required"
-class HTTPExceptionMiddleware(object):
+class HTTPExceptionMiddleware:
"""
Middleware that catches exceptions in the sub-application. This
does not catch exceptions in the app_iter; only during the initial
@@ -1335,31 +1332,12 @@ class HTTPExceptionMiddleware(object):
return parent_exc_info[1](environ, repl_start_response)
-try:
- from paste import httpexceptions
-except ImportError: # pragma: no cover
- # Without Paste we don't need to do this fixup
- pass
-else: # pragma: no cover
- for name in dir(httpexceptions):
- obj = globals().get(name)
-
- if (
- obj
- and isinstance(obj, type)
- and issubclass(obj, HTTPException)
- and obj is not HTTPException
- and obj is not WSGIHTTPException
- ):
- obj.__bases__ = obj.__bases__ + (getattr(httpexceptions, name),)
- del name, obj, httpexceptions
-
__all__ = ["HTTPExceptionMiddleware", "status_map"]
status_map = {}
for name, value in list(globals().items()):
if (
- isinstance(value, (type, class_types))
+ isinstance(value, (type, type))
and issubclass(value, HTTPException)
and not name.startswith("_")
):
diff --git a/src/webob/headers.py b/src/webob/headers.py
index f68a6ad..0061486 100644
--- a/src/webob/headers.py
+++ b/src/webob/headers.py
@@ -1,4 +1,5 @@
-from webob.compat import MutableMapping, iteritems_, string_types
+from collections.abc import MutableMapping
+
from webob.multidict import MultiDict
__all__ = ["ResponseHeaders", "EnvironHeaders"]
@@ -12,6 +13,7 @@ class ResponseHeaders(MultiDict):
def __getitem__(self, key):
key = key.lower()
+
for k, v in reversed(self._items):
if k.lower() == key:
return v
@@ -19,19 +21,24 @@ class ResponseHeaders(MultiDict):
def getall(self, key):
key = key.lower()
+
return [v for (k, v) in self._items if k.lower() == key]
def mixed(self):
r = self.dict_of_lists()
- for key, val in iteritems_(r):
+
+ for key, val in r.items():
if len(val) == 1:
r[key] = val[0]
+
return r
def dict_of_lists(self):
r = {}
- for key, val in iteritems_(self):
+
+ for key, val in self.items():
r.setdefault(key.lower(), []).append(val)
+
return r
def __setitem__(self, key, value):
@@ -43,28 +50,34 @@ class ResponseHeaders(MultiDict):
key = key.lower()
items = self._items
found = False
+
for i in range(len(items) - 1, -1, -1):
if items[i][0].lower() == key:
del items[i]
found = True
+
if not found:
raise KeyError(key)
def __contains__(self, key):
key = key.lower()
+
for k, _ in self._items:
if k.lower() == key:
return True
+
return False
has_key = __contains__
def setdefault(self, key, default=None):
c_key = key.lower()
+
for k, v in self._items:
if k.lower() == c_key:
return v
self._items.append((key, default))
+
return default
def pop(self, key, *args):
@@ -73,11 +86,14 @@ class ResponseHeaders(MultiDict):
"pop expected at most 2 arguments, got %s" % repr(1 + len(args))
)
key = key.lower()
+
for i in range(len(self._items)):
if self._items[i][0].lower() == key:
v = self._items[i][1]
del self._items[i]
+
return v
+
if args:
return args[0]
else:
@@ -91,11 +107,11 @@ key2header = {
"HTTP_CONTENT_LENGTH": "Content_Length",
}
-header2key = dict([(v.upper(), k) for (k, v) in key2header.items()])
+header2key = {v.upper(): k for (k, v) in key2header.items()}
def _trans_key(key):
- if not isinstance(key, string_types):
+ if not isinstance(key, str):
return None
elif key in key2header:
return key2header[key]
@@ -107,8 +123,10 @@ def _trans_key(key):
def _trans_name(name):
name = name.upper()
+
if name in header2key:
return header2key[name]
+
return "HTTP_" + name.replace("-", "_")
@@ -145,5 +163,4 @@ class EnvironHeaders(MutableMapping):
return len(list(self.keys()))
def __iter__(self):
- for k in self.keys():
- yield k
+ yield from self.keys()
diff --git a/src/webob/multidict.py b/src/webob/multidict.py
index abe69ee..e11bcb7 100644
--- a/src/webob/multidict.py
+++ b/src/webob/multidict.py
@@ -5,10 +5,10 @@
Gives a multi-value dictionary object (MultiDict) plus several wrappers
"""
import binascii
+from collections.abc import MutableMapping
+from urllib.parse import urlencode as url_encode
import warnings
-from webob.compat import MutableMapping, PY2, iteritems_, itervalues_, url_encode
-
__all__ = ["MultiDict", "NestedMultiDict", "NoVars", "GetDict"]
@@ -26,9 +26,7 @@ class MultiDict(MutableMapping):
)
if args:
- if hasattr(args[0], "iteritems"):
- items = list(args[0].iteritems()) # noqa: B301
- elif hasattr(args[0], "items"):
+ if hasattr(args[0], "items"):
items = list(args[0].items())
else:
items = list(args[0])
@@ -42,7 +40,7 @@ class MultiDict(MutableMapping):
@classmethod
def view_list(cls, lst):
"""
- Create a dict that is a view on the given list
+ Create a multidict that is a view on the given list
"""
if not isinstance(lst, list):
@@ -58,7 +56,7 @@ class MultiDict(MutableMapping):
@classmethod
def from_fieldstorage(cls, fs):
"""
- Create a dict from a cgi.FieldStorage instance
+ Create a multidict from a cgi.FieldStorage instance
"""
obj = cls()
# fs.list can be None when there's nothing to parse
@@ -71,21 +69,15 @@ class MultiDict(MutableMapping):
"quoted-printable": binascii.a2b_qp,
}
- if not PY2:
- if charset == "utf8":
-
- def decode(b):
- return b
-
- else:
+ if charset == "utf8":
- def decode(b):
- return b.encode("utf8").decode(charset)
+ def decode(b):
+ return b
else:
def decode(b):
- return b.decode(charset)
+ return b.encode("utf8").decode(charset)
if field.filename:
field.filename = decode(field.filename)
@@ -94,14 +86,12 @@ class MultiDict(MutableMapping):
value = field.value
if transfer_encoding in supported_transfer_encoding:
- if not PY2:
- # binascii accepts bytes
- value = value.encode("utf8")
+ # binascii accepts bytes
+ value = value.encode("utf8")
value = supported_transfer_encoding[transfer_encoding](value)
- if not PY2:
- # binascii returns bytes
- value = value.decode("utf8")
+ # binascii returns bytes
+ value = value.decode("utf8")
obj.add(field.name, decode(value))
return obj
@@ -225,17 +215,15 @@ class MultiDict(MutableMapping):
"pop expected at most 2 arguments, got %s" % repr(1 + len(args))
)
- for i in range(len(self._items)):
- if self._items[i][0] == key:
- v = self._items[i][1]
+ for i, (k, v) in enumerate(self._items):
+ if k == key:
del self._items[i]
return v
if args:
return args[0]
- else:
- raise KeyError(key)
+ raise KeyError(key)
def popitem(self):
return self._items.pop()
@@ -281,43 +269,19 @@ class MultiDict(MutableMapping):
# All the iteration:
#
- def iterkeys(self):
+ def keys(self):
for k, _ in self._items:
yield k
- if PY2:
-
- def keys(self):
- return [k for k, v in self._items]
-
- else:
- keys = iterkeys
-
- __iter__ = iterkeys
+ __iter__ = keys
- def iteritems(self):
+ def items(self):
return iter(self._items)
- if PY2:
-
- def items(self):
- return self._items[:]
-
- else:
- items = iteritems
-
- def itervalues(self):
+ def values(self):
for _, v in self._items:
yield v
- if PY2:
-
- def values(self):
- return [v for k, v in self._items]
-
- else:
- values = itervalues
-
_dummy = object()
@@ -461,49 +425,22 @@ class NestedMultiDict(MultiDict):
return False
- def iteritems(self):
+ def items(self):
for d in self.dicts:
- for item in iteritems_(d):
- yield item
-
- if PY2:
-
- def items(self):
- return list(self.iteritems()) # noqa: B301
+ yield from d.items()
- else:
- items = iteritems
-
- def itervalues(self):
+ def values(self):
for d in self.dicts:
- for value in itervalues_(d):
- yield value
-
- if PY2:
-
- def values(self):
- return list(self.itervalues()) # noqa: B301
-
- else:
- values = itervalues
+ yield from d.values()
- def __iter__(self):
+ def keys(self):
for d in self.dicts:
- for key in d:
- yield key
+ yield from d
- iterkeys = __iter__
+ __iter__ = keys
- if PY2:
- def keys(self):
- return list(self.iterkeys()) # noqa: B301
-
- else:
- keys = iterkeys
-
-
-class NoVars(object):
+class NoVars:
"""
Represents no variables; used when no variables
are applicable.
@@ -559,27 +496,12 @@ class NoVars(object):
def __len__(self):
return 0
- def iterkeys(self):
+ def keys(self):
return iter([])
- if PY2:
-
- def __cmp__(self, other):
- return cmp({}, other) # noqa: F821
-
- def keys(self):
- return []
-
- items = keys
- values = keys
- itervalues = iterkeys
- iteritems = iterkeys
- else:
- keys = iterkeys
- items = iterkeys
- values = iterkeys
-
- __iter__ = iterkeys
+ items = keys
+ values = keys
+ __iter__ = keys
def _hide_passwd(items):
diff --git a/src/webob/request.py b/src/webob/request.py
index 2684ba1..bd6a293 100644
--- a/src/webob/request.py
+++ b/src/webob/request.py
@@ -1,15 +1,12 @@
import binascii
import io
+import mimetypes
import os
import re
import sys
import tempfile
-import mimetypes
-
-try:
- import simplejson as json
-except ImportError:
- import json
+from urllib import parse as urlparse
+from urllib.parse import quote as url_quote, quote_plus, urlencode as url_encode
import warnings
from webob.acceptparse import (
@@ -18,33 +15,16 @@ from webob.acceptparse import (
accept_language_property,
accept_property,
)
-
from webob.cachecontrol import CacheControl, serialize_cache_control
-
-from webob.compat import (
- PY2,
- bytes_,
- native_,
- parse_qsl_text,
- reraise,
- text_type,
- url_encode,
- url_quote,
- url_unquote,
- quote_plus,
- urlparse,
- cgi_FieldStorage,
-)
-
+from webob.compat import cgi_FieldStorage
from webob.cookies import RequestCookies
-
from webob.descriptors import (
CHARSET_RE,
SCHEME_RE,
converter,
converter_date,
- environ_getter,
environ_decoder,
+ environ_getter,
parse_auth,
parse_int,
parse_int_safe,
@@ -55,14 +35,18 @@ from webob.descriptors import (
serialize_range,
upath_property,
)
-
-from webob.etag import IfRange, AnyETag, NoETag, etag_property
-
+from webob.etag import AnyETag, IfRange, NoETag, etag_property
from webob.headers import EnvironHeaders
+from webob.multidict import GetDict, MultiDict, NestedMultiDict, NoVars
+from webob.util import bytes_, parse_qsl_text, text_, url_unquote
+
+try:
+ import simplejson as json
+except ImportError:
+ import json
-from webob.multidict import NestedMultiDict, MultiDict, NoVars, GetDict
-__all__ = ["BaseRequest", "Request", "LegacyRequest"]
+__all__ = ["BaseRequest", "Request"]
class _NoDefault:
@@ -90,7 +74,7 @@ _LATIN_ENCODINGS = (
)
-class BaseRequest(object):
+class BaseRequest:
# The limit after which request bodies should be stored on disk
# if they are read in (under this, and the request body is stored
# in memory):
@@ -98,42 +82,12 @@ class BaseRequest(object):
_charset = None
- def __init__(
- self, environ, charset=None, unicode_errors=None, decode_param_names=None, **kw
- ):
+ def __init__(self, environ, **kw):
if type(environ) is not dict:
raise TypeError("WSGI environ must be a dict; you passed %r" % (environ,))
- if unicode_errors is not None:
- warnings.warn(
- "You unicode_errors=%r to the Request constructor. Passing a "
- "``unicode_errors`` value to the Request is no longer "
- "supported in WebOb 1.2+. This value has been ignored "
- % (unicode_errors,),
- DeprecationWarning,
- )
-
- if decode_param_names is not None:
- warnings.warn(
- "You passed decode_param_names=%r to the Request constructor. "
- "Passing a ``decode_param_names`` value to the Request "
- "is no longer supported in WebOb 1.2+. This value has "
- "been ignored " % (decode_param_names,),
- DeprecationWarning,
- )
-
- if not _is_utf8(charset):
- raise DeprecationWarning(
- "You passed charset=%r to the Request constructor. As of "
- "WebOb 1.2, if your application needs a non-UTF-8 request "
- "charset, please construct the request without a charset or "
- "with a charset of 'None', then use ``req = "
- "req.decode(charset)``" % charset
- )
-
- d = self.__dict__
- d["environ"] = environ
+ self.__dict__["environ"] = environ
if kw:
cls = self.__class__
@@ -150,19 +104,20 @@ class BaseRequest(object):
def encget(self, key, default=NoDefault, encattr=None):
val = self.environ.get(key, default)
+
if val is NoDefault:
raise KeyError(key)
+
if val is default:
return default
+
if not encattr:
return val
encoding = getattr(self, encattr)
- if PY2:
- return val.decode(encoding)
-
if encoding in _LATIN_ENCODINGS: # shortcut
return val
+
return bytes_(val, "latin-1").decode(encoding)
def encset(self, key, val, encattr=None):
@@ -170,29 +125,30 @@ class BaseRequest(object):
encoding = getattr(self, encattr)
else:
encoding = "ascii"
- if PY2: # pragma: no cover
- self.environ[key] = bytes_(val, encoding)
- else:
- self.environ[key] = bytes_(val, encoding).decode("latin-1")
+ self.environ[key] = bytes_(val, encoding).decode("latin-1")
@property
def charset(self):
if self._charset is None:
charset = detect_charset(self._content_type_raw)
+
if _is_utf8(charset):
charset = "UTF-8"
self._charset = charset
+
return self._charset
@charset.setter
def charset(self, charset):
if _is_utf8(charset):
charset = "UTF-8"
+
if charset != self.charset:
raise DeprecationWarning("Use req = req.decode(%r)" % charset)
def decode(self, charset=None, errors="strict"):
charset = charset or self.charset
+
if charset == "UTF-8":
return self
# cookies and path are always utf-8
@@ -207,7 +163,8 @@ class BaseRequest(object):
)
if content_type == "application/x-www-form-urlencoded":
- r.body = bytes_(t.transcode_query(native_(self.body)))
+ r.body = bytes_(t.transcode_query(text_(self.body)))
+
return r
elif content_type != "multipart/form-data":
return r
@@ -215,18 +172,13 @@ class BaseRequest(object):
fs_environ = self.environ.copy()
fs_environ.setdefault("CONTENT_LENGTH", "0")
fs_environ["QUERY_STRING"] = ""
- if PY2:
- fs = cgi_FieldStorage(
- fp=self.body_file, environ=fs_environ, keep_blank_values=True
- )
- else:
- fs = cgi_FieldStorage(
- fp=self.body_file,
- environ=fs_environ,
- keep_blank_values=True,
- encoding=charset,
- errors=errors,
- )
+ fs = cgi_FieldStorage(
+ fp=self.body_file,
+ environ=fs_environ,
+ keep_blank_values=True,
+ encoding=charset,
+ errors=errors,
+ )
fout = t.transcode_fs(fs, r._content_type_raw)
@@ -235,6 +187,7 @@ class BaseRequest(object):
r.body_file = fout
r.content_length = fout.tell()
fout.seek(0)
+
return r
# this is necessary for correct warnings depth for both
@@ -298,8 +251,10 @@ class BaseRequest(object):
If you access this value, CONTENT_LENGTH will also be updated.
"""
+
if not self.is_body_seekable:
self.make_body_seekable()
+
return self.body_file_raw
url_encoding = environ_getter("webob.url_encoding", "UTF-8")
@@ -339,13 +294,16 @@ class BaseRequest(object):
you don't include any parameters in the value then existing
parameters will be preserved.
"""
+
return self._content_type_raw.split(";", 1)[0]
def _content_type__set(self, value=None):
if value is not None:
value = str(value)
+
if ";" not in value:
content_type = self._content_type_raw
+
if ";" in content_type:
value += ";" + content_type.split(";", 1)[1]
self._content_type_raw = value
@@ -364,8 +322,10 @@ class BaseRequest(object):
All the request headers as a case-insensitive dictionary-like
object.
"""
+
if self._headers is None:
self._headers = EnvironHeaders(self.environ)
+
return self._headers
def _headers__set(self, value):
@@ -398,10 +358,12 @@ class BaseRequest(object):
"""
e = self.environ
xff = e.get("HTTP_X_FORWARDED_FOR")
+
if xff is not None:
addr = xff.split(",")[0].strip()
else:
addr = e.get("REMOTE_ADDR")
+
return addr
@property
@@ -418,17 +380,20 @@ class BaseRequest(object):
"""
e = self.environ
host = e.get("HTTP_HOST")
+
if host is not None:
if ":" in host and host[-1] != "]":
host, port = host.rsplit(":", 1)
else:
url_scheme = e["wsgi.url_scheme"]
+
if url_scheme == "https":
port = "443"
else:
port = "80"
else:
port = e["SERVER_PORT"]
+
return port
@property
@@ -440,6 +405,7 @@ class BaseRequest(object):
scheme = e.get("wsgi.url_scheme")
url = scheme + "://"
host = e.get("HTTP_HOST")
+
if host is not None:
if ":" in host and host[-1] != "]":
host, port = host.rsplit(":", 1)
@@ -448,6 +414,7 @@ class BaseRequest(object):
else:
host = e.get("SERVER_NAME")
port = e.get("SERVER_PORT")
+
if scheme == "https":
if port == "443":
port = None
@@ -455,8 +422,10 @@ class BaseRequest(object):
if port == "80":
port = None
url += host
+
if port:
url += ":%s" % port
+
return url
@property
@@ -465,6 +434,7 @@ class BaseRequest(object):
The URL including SCRIPT_NAME (no PATH_INFO or query string)
"""
bscript_name = bytes_(self.script_name, self.url_encoding)
+
return self.host_url + url_quote(bscript_name, PATH_SAFE)
@property
@@ -473,6 +443,7 @@ class BaseRequest(object):
The URL including SCRIPT_NAME and PATH_INFO, but not QUERY_STRING
"""
bpath_info = bytes_(self.path_info, self.url_encoding)
+
return self.application_url + url_quote(bpath_info, PATH_SAFE)
@property
@@ -482,6 +453,7 @@ class BaseRequest(object):
"""
bscript = bytes_(self.script_name, self.url_encoding)
bpath = bytes_(self.path_info, self.url_encoding)
+
return url_quote(bscript, PATH_SAFE) + url_quote(bpath, PATH_SAFE)
@property
@@ -491,8 +463,10 @@ class BaseRequest(object):
"""
path = self.path
qs = self.environ.get("QUERY_STRING")
+
if qs:
path += "?" + qs
+
return path
@property
@@ -502,8 +476,10 @@ class BaseRequest(object):
"""
url = self.path_url
qs = self.environ.get("QUERY_STRING")
+
if qs:
url += "?" + qs
+
return url
def relative_url(self, other_url, to_application=False):
@@ -513,12 +489,15 @@ class BaseRequest(object):
If ``to_application`` is True, then resolve it relative to the
URL with only SCRIPT_NAME
"""
+
if to_application:
url = self.application_url
+
if not url.endswith("/"):
url += "/"
else:
url = self.path_url
+
return urlparse.urljoin(url, other_url)
def path_info_pop(self, pattern=None):
@@ -535,19 +514,24 @@ class BaseRequest(object):
request and None is returned.
"""
path = self.path_info
+
if not path:
return None
slashes = ""
+
while path.startswith("/"):
slashes += "/"
path = path[1:]
idx = path.find("/")
+
if idx == -1:
idx = len(path)
r = path[:idx]
+
if pattern is None or re.match(pattern, r):
self.script_name += slashes + r
self.path_info = path[idx:]
+
return r
def path_info_peek(self):
@@ -556,9 +540,11 @@ class BaseRequest(object):
next segment. Doesn't modify the environment.
"""
path = self.path_info
+
if not path:
return None
path = path.lstrip("/")
+
return path.split("/", 1)[0]
def _urlvars__get(self):
@@ -568,6 +554,7 @@ class BaseRequest(object):
Takes values from ``environ['wsgiorg.routing_args']``.
Systems like ``routes`` set this value.
"""
+
if "paste.urlvars" in self.environ:
return self.environ["paste.urlvars"]
elif "wsgiorg.routing_args" in self.environ:
@@ -575,15 +562,18 @@ class BaseRequest(object):
else:
result = {}
self.environ["wsgiorg.routing_args"] = ((), result)
+
return result
def _urlvars__set(self, value):
environ = self.environ
+
if "wsgiorg.routing_args" in environ:
environ["wsgiorg.routing_args"] = (
environ["wsgiorg.routing_args"][0],
value,
)
+
if "paste.urlvars" in environ:
del environ["paste.urlvars"]
elif "paste.urlvars" in environ:
@@ -594,6 +584,7 @@ class BaseRequest(object):
def _urlvars__del(self):
if "paste.urlvars" in self.environ:
del self.environ["paste.urlvars"]
+
if "wsgiorg.routing_args" in self.environ:
if not self.environ["wsgiorg.routing_args"][0]:
del self.environ["wsgiorg.routing_args"]
@@ -614,15 +605,18 @@ class BaseRequest(object):
Takes values from ``environ['wsgiorg.routing_args']``.
Systems like ``routes`` set this value.
"""
+
if "wsgiorg.routing_args" in self.environ:
return self.environ["wsgiorg.routing_args"][0]
else:
# Since you can't update this value in-place, we don't need
# to set the key in the environment
+
return ()
def _urlargs__set(self, value):
environ = self.environ
+
if "paste.urlvars" in environ:
# Some overlap between this and wsgiorg.routing_args; we need
# wsgiorg.routing_args to make this work
@@ -655,10 +649,12 @@ class BaseRequest(object):
only set if you are using a Javascript library that sets it
(or you set the header yourself manually). Currently
Prototype and jQuery are known to set this header."""
+
return self.environ.get("HTTP_X_REQUESTED_WITH", "") == "XMLHttpRequest"
def _host__get(self):
"""Host name provided in HTTP_HOST, with fall-back to SERVER_NAME"""
+
if "HTTP_HOST" in self.environ:
return self.environ["HTTP_HOST"]
else:
@@ -693,8 +689,10 @@ class BaseRequest(object):
value use :meth:`webob.request.Request.host` instead.
"""
domain = self.host
+
if ":" in domain and domain[-1] != "]":
domain = domain.rsplit(":", 1)[0]
+
return domain
@property
@@ -702,18 +700,21 @@ class BaseRequest(object):
"""
Return the content of the request body.
"""
+
if not self.is_body_readable:
return b""
self.make_body_seekable() # we need this to have content_length
r = self.body_file.read(self.content_length)
self.body_file_raw.seek(0)
+
return r
@body.setter
def body(self, value):
if value is None:
value = b""
+
if not isinstance(value, bytes):
raise TypeError(
"You can only set Request.body to bytes (not %r)" % type(value)
@@ -728,6 +729,7 @@ class BaseRequest(object):
def _json_body__get(self):
"""Access the body of the request as JSON"""
+
return json.loads(self.body.decode(self.charset))
def _json_body__set(self, value):
@@ -742,9 +744,11 @@ class BaseRequest(object):
"""
Get/set the text value of the body
"""
+
if not self.charset:
raise AttributeError("You cannot access Request.text unless charset is set")
body = self.body
+
return body.decode(self.charset)
def _text__set(self, value):
@@ -752,7 +756,8 @@ class BaseRequest(object):
raise AttributeError(
"You cannot access Response.text unless charset is set"
)
- if not isinstance(value, text_type):
+
+ if not isinstance(value, str):
raise TypeError(
"You can only set Request.text to a unicode string "
"(not %s)" % type(value)
@@ -774,17 +779,21 @@ class BaseRequest(object):
requests with an appropriate Content-Type are also supported.
"""
env = self.environ
+
if "webob._parsed_post_vars" in env:
vars, body_file = env["webob._parsed_post_vars"]
+
if body_file is self.body_file_raw:
return vars
content_type = self.content_type
+
if (self.method != "POST" and not content_type) or content_type not in (
"",
"application/x-www-form-urlencoded",
"multipart/form-data",
):
# Not an HTML form submission
+
return NoVars(
"Not an HTML form submission (Content-Type: %s)" % content_type
)
@@ -798,21 +807,17 @@ class BaseRequest(object):
# default of 0 is better:
fs_environ.setdefault("CONTENT_LENGTH", "0")
fs_environ["QUERY_STRING"] = ""
- if PY2:
- fs = cgi_FieldStorage(
- fp=self.body_file, environ=fs_environ, keep_blank_values=True
- )
- else:
- fs = cgi_FieldStorage(
- fp=self.body_file,
- environ=fs_environ,
- keep_blank_values=True,
- encoding="utf8",
- )
+ fs = cgi_FieldStorage(
+ fp=self.body_file,
+ environ=fs_environ,
+ keep_blank_values=True,
+ encoding="utf8",
+ )
self.body_file_raw.seek(0)
vars = MultiDict.from_fieldstorage(fs)
env["webob._parsed_post_vars"] = (vars, self.body_file_raw)
+
return vars
@property
@@ -823,12 +828,15 @@ class BaseRequest(object):
"""
env = self.environ
source = env.get("QUERY_STRING", "")
+
if "webob._parsed_query_vars" in env:
vars, qs = env["webob._parsed_query_vars"]
+
if qs == source:
return vars
data = []
+
if source:
# this is disabled because we want to access req.GET
# for text/plain; charset=ascii uploads for example
@@ -838,6 +846,7 @@ class BaseRequest(object):
# data = [(d(k), d(v)) for k,v in data]
vars = GetDict(data, env)
env["webob._parsed_query_vars"] = (vars, source)
+
return vars
def _check_charset(self):
@@ -855,6 +864,7 @@ class BaseRequest(object):
the query string and request body.
"""
params = NestedMultiDict(self.GET, self.POST)
+
return params
@property
@@ -862,6 +872,7 @@ class BaseRequest(object):
"""
Return a dictionary of cookies as found in the request.
"""
+
return RequestCookies(self.environ)
@cookies.setter
@@ -880,6 +891,7 @@ class BaseRequest(object):
env = self.environ.copy()
new_req = self.__class__(env)
new_req.copy_body()
+
return new_req
def copy_get(self):
@@ -889,6 +901,7 @@ class BaseRequest(object):
verb) then it becomes GET, and the request body is thrown away.
"""
env = self.environ.copy()
+
return self.__class__(env, method="GET", content_type=None, body=b"")
# webob.is_body_seekable marks input streams that are seekable
@@ -911,6 +924,7 @@ class BaseRequest(object):
# Encoding is allowed (and works) or we have replaced
# self.body_file with something that is readable and EOF's
# correctly.
+
return self.environ.get(
"wsgi.input_terminated",
# For backwards compatibility, we fall back to checking if
@@ -937,6 +951,7 @@ class BaseRequest(object):
The choice to copy to BytesIO is made from
``self.request_body_tempfile_limit``
"""
+
if self.is_body_seekable:
self.body_file_raw.seek(0)
else:
@@ -953,6 +968,7 @@ class BaseRequest(object):
if self.is_body_readable:
# Before we copy, if we can, rewind the body file
+
if self.is_body_seekable:
self.body_file_raw.seek(0)
@@ -970,6 +986,7 @@ class BaseRequest(object):
# We attempted to read more data, but got none, break.
# This can happen if for instance we are reading as much as
# we can because we don't have a Content-Length...
+
break
elif not data:
# We have a Content-Length and we attempted to read, but
@@ -989,12 +1006,14 @@ class BaseRequest(object):
# When we have enough data that we need a tempfile, let's
# create one, then clear the temporary variable we were
# using
+
if len(newbody) > tempfile_limit:
fileobj = self.make_tempfile()
fileobj.write(newbody)
newbody = b""
# Only decrement todo if Content-Length is set
+
if self.content_length is not None:
todo -= len(data)
@@ -1030,6 +1049,7 @@ class BaseRequest(object):
Create a tempfile to store big request body.
This API is not stable yet. A 'size' argument might be added.
"""
+
return tempfile.TemporaryFile()
def remove_conditional_headers(
@@ -1049,12 +1069,16 @@ class BaseRequest(object):
conflict detection.
"""
check_keys = []
+
if remove_range:
check_keys += ["HTTP_IF_RANGE", "HTTP_RANGE"]
+
if remove_match:
check_keys.append("HTTP_IF_NONE_MATCH")
+
if remove_modified:
check_keys.append("HTTP_IF_MODIFIED_SINCE")
+
if remove_encoding:
check_keys.append("HTTP_ACCEPT_ENCODING")
@@ -1079,19 +1103,23 @@ class BaseRequest(object):
env = self.environ
value = env.get("HTTP_CACHE_CONTROL", "")
cache_header, cache_obj = env.get("webob._cache_control", (None, None))
+
if cache_obj is not None and cache_header == value:
return cache_obj
cache_obj = CacheControl.parse(
value, updates_to=self._update_cache_control, type="request"
)
env["webob._cache_control"] = (value, cache_obj)
+
return cache_obj
def _cache_control__set(self, value):
env = self.environ
value = value or ""
+
if isinstance(value, dict):
value = CacheControl(value, type="request")
+
if isinstance(value, CacheControl):
str_value = str(value)
env["HTTP_CACHE_CONTROL"] = str_value
@@ -1102,8 +1130,10 @@ class BaseRequest(object):
def _cache_control__del(self):
env = self.environ
+
if "HTTP_CACHE_CONTROL" in env:
del env["HTTP_CACHE_CONTROL"]
+
if "webob._cache_control" in env:
del env["webob._cache_control"]
@@ -1161,6 +1191,7 @@ class BaseRequest(object):
except KeyError:
name = "(invalid WSGI environ)"
msg = "<%s at 0x%x %s>" % (self.__class__.__name__, abs(id(self)), name)
+
return msg
def as_bytes(self, skip_body=False):
@@ -1179,12 +1210,14 @@ class BaseRequest(object):
# acquire body before we handle headers so that
# content-length will be set
body = None
+
if self.is_body_readable:
if skip_body > 1:
if len(self.body) > skip_body:
body = bytes_("<body skipped (len=%s)>" % len(self.body))
else:
skip_body = False
+
if not skip_body:
body = self.body
@@ -1195,10 +1228,12 @@ class BaseRequest(object):
if body:
parts.extend([b"", body])
# HTTP clearly specifies CRLF
+
return b"\r\n".join(parts)
def as_text(self, skip_body=False):
bytes = self.as_bytes(skip_body)
+
return bytes.decode(self.charset)
__str__ = as_text
@@ -1211,13 +1246,16 @@ class BaseRequest(object):
"""
f = io.BytesIO(b)
r = cls.from_file(f)
+
if f.tell() != len(b):
raise ValueError("The string contains more data than expected")
+
return r
@classmethod
def from_text(cls, s):
b = bytes_(s, "utf-8")
+
return cls.from_bytes(b)
@classmethod
@@ -1233,7 +1271,8 @@ class BaseRequest(object):
not read every valid HTTP request properly.
"""
start_line = fp.readline()
- is_text = isinstance(start_line, text_type)
+ is_text = isinstance(start_line, str)
+
if is_text:
crlf = "\r\n"
colon = ":"
@@ -1243,32 +1282,38 @@ class BaseRequest(object):
try:
header = start_line.rstrip(crlf)
method, resource, http_version = header.split(None, 2)
- method = native_(method, "utf-8")
- resource = native_(resource, "utf-8")
- http_version = native_(http_version, "utf-8")
+ method = text_(method, "utf-8")
+ resource = text_(resource, "utf-8")
+ http_version = text_(http_version, "utf-8")
except ValueError:
raise ValueError("Bad HTTP request line: %r" % start_line)
r = cls(
environ_from_url(resource), http_version=http_version, method=method.upper()
)
del r.environ["HTTP_HOST"]
+
while 1:
line = fp.readline()
+
if not line.strip():
# end of headers
+
break
hname, hval = line.split(colon, 1)
- hname = native_(hname, "utf-8")
- hval = native_(hval, "utf-8").strip()
+ hname = text_(hname, "utf-8")
+ hval = text_(hval, "utf-8").strip()
+
if hname in r.headers:
hval = r.headers[hname] + ", " + hval
r.headers[hname] = hval
clen = r.content_length
+
if clen is None:
body = fp.read()
else:
body = fp.read(clen)
+
if is_text:
body = bytes_(body, "utf-8")
r.body = body
@@ -1288,6 +1333,7 @@ class BaseRequest(object):
do this and there was an exception, the exception will be
raised directly.
"""
+
if self.is_body_seekable:
self.body_file_raw.seek(0)
captured = []
@@ -1295,11 +1341,14 @@ class BaseRequest(object):
def start_response(status, headers, exc_info=None):
if exc_info is not None and not catch_exc_info:
- reraise(exc_info)
+ etype, exc, tb = exc_info
+ raise etype(exc).with_traceback(tb)
captured[:] = [status, headers, exc_info]
+
return output.append
app_iter = application(self.environ, start_response)
+
if output or not captured:
try:
output.extend(app_iter)
@@ -1307,6 +1356,7 @@ class BaseRequest(object):
if hasattr(app_iter, "close"):
app_iter.close()
app_iter = output
+
if catch_exc_info:
return (captured[0], captured[1], app_iter, captured[2])
else:
@@ -1327,8 +1377,10 @@ class BaseRequest(object):
If ``application`` is not given, this will send the request to
``self.make_default_send_app()``
"""
+
if application is None:
application = self.make_default_send_app()
+
if catch_exc_info:
status, headers, app_iter, exc_info = self.call_application(
application, catch_exc_info=True
@@ -1338,6 +1390,7 @@ class BaseRequest(object):
status, headers, app_iter = self.call_application(
application, catch_exc_info=False
)
+
return self.ResponseClass(
status=status, headerlist=list(headers), app_iter=app_iter
)
@@ -1352,6 +1405,7 @@ class BaseRequest(object):
from webob import client
_client = client
+
return client.send_request_app
@classmethod
@@ -1372,14 +1426,18 @@ class BaseRequest(object):
Any extra keyword will be passed to ``__init__``.
"""
env = environ_from_url(path)
+
if base_url:
scheme, netloc, path, query, fragment = urlparse.urlsplit(base_url)
+
if query or fragment:
raise ValueError(
"base_url (%r) cannot have a query or fragment" % base_url
)
+
if scheme:
env["wsgi.url_scheme"] = scheme
+
if netloc:
if ":" not in netloc:
if scheme == "http":
@@ -1392,36 +1450,29 @@ class BaseRequest(object):
env["SERVER_PORT"] = port
env["SERVER_NAME"] = host
env["HTTP_HOST"] = netloc
+
if path:
env["SCRIPT_NAME"] = url_unquote(path)
+
if environ:
env.update(environ)
content_type = kw.get("content_type", env.get("CONTENT_TYPE"))
+
if headers and "Content-Type" in headers:
content_type = headers["Content-Type"]
+
if content_type is not None:
kw["content_type"] = content_type
environ_add_POST(env, POST, content_type=content_type)
obj = cls(env, **kw)
+
if headers is not None:
obj.headers.update(headers)
- return obj
-
-
-class LegacyRequest(BaseRequest):
- uscript_name = upath_property("SCRIPT_NAME")
- upath_info = upath_property("PATH_INFO")
- def encget(self, key, default=NoDefault, encattr=None):
- val = self.environ.get(key, default)
- if val is NoDefault:
- raise KeyError(key)
- if val is default:
- return default
- return val
+ return obj
-class AdhocAttrMixin(object):
+class AdhocAttrMixin:
_setattr_stacklevel = 3
def __setattr__(self, attr, value, DEFAULT=DEFAULT):
@@ -1454,10 +1505,13 @@ class Request(AdhocAttrMixin, BaseRequest):
def environ_from_url(path):
if SCHEME_RE.search(path):
scheme, netloc, path, qs, fragment = urlparse.urlsplit(path)
+
if fragment:
raise TypeError("Path cannot contain a fragment (%r)" % fragment)
+
if qs:
path += "?" + qs
+
if ":" not in netloc:
if scheme == "http":
netloc += ":80"
@@ -1468,6 +1522,7 @@ def environ_from_url(path):
else:
scheme = "http"
netloc = "localhost:80"
+
if path and "?" in path:
path_info, query_string = path.split("?", 1)
path_info = url_unquote(path_info)
@@ -1492,28 +1547,35 @@ def environ_from_url(path):
"wsgi.run_once": False,
"webob.is_body_seekable": True,
}
+
return env
def environ_add_POST(env, data, content_type=None):
if data is None:
return
- elif isinstance(data, text_type):
+ elif isinstance(data, str):
data = data.encode("ascii")
+
if env["REQUEST_METHOD"] not in ("POST", "PUT"):
env["REQUEST_METHOD"] = "POST"
has_files = False
+
if hasattr(data, "items"):
data = list(data.items())
+
for _, v in data:
- if isinstance(v, (tuple, list)):
+ if isinstance(v, (tuple, list)) or hasattr(v, "filename"):
has_files = True
+
break
+
if content_type is None:
if has_files:
content_type = "multipart/form-data"
else:
content_type = "application/x-www-form-urlencoded"
+
if content_type.startswith("multipart/form-data"):
if not isinstance(data, bytes):
content_type, data = _encode_multipart(data, content_type)
@@ -1522,6 +1584,7 @@ def environ_add_POST(env, data, content_type=None):
raise ValueError(
"Submiting files is not allowed for" " content type `%s`" % content_type
)
+
if not isinstance(data, bytes):
data = url_encode(data)
else:
@@ -1569,88 +1632,22 @@ class LimitedLengthFile(io.RawIOBase):
data = self.file.read(sz0)
sz = len(data)
self.remaining -= sz
+
if sz < sz0 and self.remaining:
raise DisconnectionError(
"The client disconnected while sending the body "
"(%d more bytes were expected)" % (self.remaining,)
)
buff[:sz] = data
- return sz
-
-
-def _cgi_FieldStorage__repr__patch(self):
- """monkey patch for FieldStorage.__repr__
-
- Unbelievably, the default __repr__ on FieldStorage reads
- the entire file content instead of being sane about it.
- This is a simple replacement that doesn't do that
- """
- if self.file:
- return "FieldStorage(%r, %r)" % (self.name, self.filename)
- return "FieldStorage(%r, %r, %r)" % (self.name, self.filename, self.value)
-
-
-cgi_FieldStorage.__repr__ = _cgi_FieldStorage__repr__patch
-
-
-class FakeCGIBody(io.RawIOBase):
- def __init__(self, vars, content_type):
- warnings.warn(
- "FakeCGIBody is no longer used by WebOb and will be removed from a future "
- "version of WebOb. If you require FakeCGIBody please make a copy into "
- "you own project",
- DeprecationWarning,
- )
-
- if content_type.startswith("multipart/form-data"):
- if not _get_multipart_boundary(content_type):
- raise ValueError(
- "Content-type: %r does not contain boundary" % content_type
- )
- self.vars = vars
- self.content_type = content_type
- self.file = None
-
- def __repr__(self):
- inner = repr(self.vars)
- if len(inner) > 20:
- inner = inner[:15] + "..." + inner[-5:]
- return "<%s at 0x%x viewing %s>" % (
- self.__class__.__name__,
- abs(id(self)),
- inner,
- )
-
- def fileno(self):
- return None
-
- @staticmethod
- def readable():
- return True
- def readinto(self, buff):
- if self.file is None:
- if self.content_type.startswith("application/x-www-form-urlencoded"):
- data = "&".join(
- "%s=%s"
- % (quote_plus(bytes_(k, "utf8")), quote_plus(bytes_(v, "utf8")))
- for k, v in self.vars.items()
- )
- self.file = io.BytesIO(bytes_(data))
- elif self.content_type.startswith("multipart/form-data"):
- self.file = _encode_multipart(
- self.vars.items(), self.content_type, fout=io.BytesIO()
- )[1]
- self.file.seek(0)
- else:
- assert 0, "Bad content type: %r" % self.content_type
- return self.file.readinto(buff)
+ return sz
def _get_multipart_boundary(ctype):
m = re.search(r"boundary=([^ ]+)", ctype, re.I)
+
if m:
- return native_(m.group(1).strip('"'))
+ return text_(m.group(1).strip('"'))
def _encode_multipart(vars, content_type, fout=None):
@@ -1663,21 +1660,26 @@ def _encode_multipart(vars, content_type, fout=None):
CRLF = b"\r\n"
boundary = _get_multipart_boundary(content_type)
+
if not boundary:
- boundary = native_(binascii.hexlify(os.urandom(10)))
+ boundary = text_(binascii.hexlify(os.urandom(10)))
content_type += "; boundary=%s" % boundary
+
for name, value in vars:
w(b"--")
wt(boundary)
w(CRLF)
wt("Content-Disposition: form-data")
+
if name is not None:
wt('; name="%s"' % name)
filename = None
+
if getattr(value, "filename", None):
filename = value.filename
elif isinstance(value, (list, tuple)):
filename, value = value
+
if hasattr(value, "read"):
value = value.read()
@@ -1690,8 +1692,10 @@ def _encode_multipart(vars, content_type, fout=None):
w(CRLF)
# TODO: should handle value.disposition_options
+
if getattr(value, "type", None):
wt("Content-type: %s" % value.type)
+
if value.type_options:
for ct_name, ct_value in sorted(value.type_options.items()):
wt('; %s="%s"' % (ct_name, ct_value))
@@ -1700,14 +1704,17 @@ def _encode_multipart(vars, content_type, fout=None):
wt("Content-type: %s" % mime_type)
w(CRLF)
w(CRLF)
+
if hasattr(value, "value"):
value = value.value
+
if isinstance(value, bytes):
w(value)
else:
wt(value)
w(CRLF)
wt("--%s--" % boundary)
+
if fout:
return content_type, fout
else:
@@ -1716,6 +1723,7 @@ def _encode_multipart(vars, content_type, fout=None):
def detect_charset(ctype):
m = CHARSET_RE.search(ctype)
+
if m:
return m.group(1).strip('"').strip()
@@ -1727,7 +1735,7 @@ def _is_utf8(charset):
return charset.lower().replace("-", "") == "utf8"
-class Transcoder(object):
+class Transcoder:
def __init__(self, charset, errors="strict"):
self.charset = charset # source charset
self.errors = errors # unicode errors
@@ -1735,37 +1743,26 @@ class Transcoder(object):
def transcode_query(self, q):
q_orig = q
+
if "=" not in q:
# this doesn't look like a form submission
+
return q_orig
- if PY2:
- q = urlparse.parse_qsl(q, self.charset)
- t = self._trans
- q = [(t(k), t(v)) for k, v in q]
- else:
- q = list(parse_qsl_text(q, self.charset))
+ q = list(parse_qsl_text(q, self.charset))
return url_encode(q)
def transcode_fs(self, fs, content_type):
# transcode FieldStorage
- if PY2:
-
- def decode(b):
- if b is not None:
- return b.decode(self.charset, self.errors)
- else:
- return b
-
- else:
-
- def decode(b):
- return b
+ def decode(b):
+ return b
data = []
+
for field in fs.list or ():
field.name = decode(field.name)
+
if field.filename:
field.filename = decode(field.filename)
data.append((field.name, field))
@@ -1774,4 +1771,5 @@ class Transcoder(object):
# TODO: transcode big requests to temp file
content_type, fout = _encode_multipart(data, content_type, fout=io.BytesIO())
+
return fout
diff --git a/src/webob/response.py b/src/webob/response.py
index 58195ac..e02de87 100644
--- a/src/webob/response.py
+++ b/src/webob/response.py
@@ -1,21 +1,14 @@
-import re
-import struct
-import zlib
from base64 import b64encode
from datetime import datetime, timedelta
from hashlib import md5
+import re
+import struct
+from urllib import parse as urlparse
+from urllib.parse import quote as url_quote
+import zlib
from webob.byterange import ContentRange
from webob.cachecontrol import CacheControl, serialize_cache_control
-from webob.compat import (
- PY2,
- bytes_,
- native_,
- string_types,
- text_type,
- url_quote,
- urlparse,
-)
from webob.cookies import Cookie, make_cookie
from webob.datetime_utils import (
parse_date_delta,
@@ -41,7 +34,13 @@ from webob.descriptors import (
)
from webob.headers import ResponseHeaders
from webob.request import BaseRequest
-from webob.util import status_generic_reasons, status_reasons, warn_deprecation
+from webob.util import (
+ bytes_,
+ status_generic_reasons,
+ status_reasons,
+ text_,
+ warn_deprecation,
+)
try:
import simplejson as json
@@ -59,7 +58,7 @@ _gzip_header = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff"
_marker = object()
-class Response(object):
+class Response:
"""
Represents a WSGI response.
@@ -175,9 +174,10 @@ class Response(object):
content_type=None,
conditional_response=None,
charset=_marker,
- **kw
+ **kw,
):
# Do some sanity checking, and turn json_body into an actual body
+
if app_iter is None and body is None and ("json_body" in kw or "json" in kw):
if "json_body" in kw:
json_body = kw.pop("json_body")
@@ -195,6 +195,7 @@ class Response(object):
raise TypeError("You may only give one of the body and app_iter arguments")
# Set up Response.status
+
if status is None:
self._status = "200 OK"
else:
@@ -202,6 +203,7 @@ class Response(object):
# Initialize headers
self._headers = None
+
if headerlist is None:
self._headerlist = []
else:
@@ -218,6 +220,7 @@ class Response(object):
# Content-Type: application/foo with no charset on it.
encoding = None
+
if charset is not _marker:
encoding = charset
@@ -253,6 +256,7 @@ class Response(object):
# If the Content-Type already has a charset, we don't set the user
# provided charset on the Content-Type, so we shouldn't use it as
# the encoding for text_type based body's.
+
if has_charset:
encoding = None
@@ -283,17 +287,20 @@ class Response(object):
self._headerlist.append(("Content-Type", content_type))
# Set up conditional response
+
if conditional_response is None:
self.conditional_response = self.default_conditional_response
else:
self.conditional_response = bool(conditional_response)
# Set up app_iter if the HTTP Status code has a body
+
if app_iter is None and code_has_body:
- if isinstance(body, text_type):
+ if isinstance(body, str):
# Fall back to trying self.charset if encoding is not set. In
# most cases encoding will be set to the default value.
encoding = encoding or self.charset
+
if encoding is None:
raise TypeError(
"You cannot set the body to a text value without a " "charset"
@@ -314,6 +321,7 @@ class Response(object):
self._app_iter = app_iter
# Loop through all the remaining keyword arguments
+
for name, value in kw.items():
if not hasattr(self.__class__, name):
# Not a basic attribute
@@ -333,7 +341,7 @@ class Response(object):
must have a ``Content-Length``."""
headerlist = []
status = fp.readline().strip()
- is_text = isinstance(status, text_type)
+ is_text = isinstance(status, str)
if is_text:
_colon = ":"
@@ -344,27 +352,29 @@ class Response(object):
if status.startswith(_http):
(http_ver, status_num, status_text) = status.split(None, 2)
- status = "%s %s" % (native_(status_num), native_(status_text))
+ status = "%s %s" % (text_(status_num), text_(status_text))
while 1:
line = fp.readline().strip()
+
if not line:
# end of headers
+
break
try:
header_name, value = line.split(_colon, 1)
except ValueError:
raise ValueError("Bad header line: %r" % line)
value = value.strip()
- headerlist.append(
- (native_(header_name, "latin-1"), native_(value, "latin-1"))
- )
+ headerlist.append((text_(header_name, "latin-1"), text_(value, "latin-1")))
r = cls(status=status, headerlist=headerlist, app_iter=())
body = fp.read(r.content_length or 0)
+
if is_text:
r.text = body
else:
r.body = body
+
return r
def copy(self):
@@ -374,6 +384,7 @@ class Response(object):
iter_close(self._app_iter)
# and this to make sure app_iter instances are different
self._app_iter = list(app_iter)
+
return self.__class__(
status=self._status,
headerlist=self._headerlist[:],
@@ -390,12 +401,15 @@ class Response(object):
def __str__(self, skip_body=False):
parts = [self.status]
+
if not skip_body:
# Force enumeration of the body (to set content-length)
self.body
parts += map("%s: %s".__mod__, self.headerlist)
+
if not skip_body and self.body:
- parts += ["", self.body if PY2 else self.text]
+ parts += ["", self.text]
+
return "\r\n".join(parts)
#
@@ -406,6 +420,7 @@ class Response(object):
"""
The status string.
"""
+
return self._status
def _status__set(self, value):
@@ -415,12 +430,12 @@ class Response(object):
pass
else:
self.status_code = code
+
return
- if not PY2:
- if isinstance(value, bytes):
- value = value.decode("ascii")
- elif isinstance(value, text_type):
- value = value.encode("ascii")
+
+ if isinstance(value, bytes):
+ value = value.decode("ascii")
+
if not isinstance(value, str):
raise TypeError(
"You must set status to a string or integer (not %s)" % type(value)
@@ -442,6 +457,7 @@ class Response(object):
"""
The status as an integer.
"""
+
return int(self._status.split()[0])
def _status_code__set(self, code):
@@ -462,10 +478,12 @@ class Response(object):
"""
The list of response headers.
"""
+
return self._headerlist
def _headerlist__set(self, value):
self._headers = None
+
if not isinstance(value, list):
if hasattr(value, "items"):
value = value.items()
@@ -486,8 +504,10 @@ class Response(object):
"""
The headers in a dictionary-like object.
"""
+
if self._headers is None:
self._headers = ResponseHeaders.view_list(self._headerlist)
+
return self._headers
def _headers__set(self, value):
@@ -513,17 +533,19 @@ class Response(object):
# return app_iter[0]
# except:
# pass
+
if isinstance(app_iter, list) and len(app_iter) == 1:
return app_iter[0]
+
if app_iter is None:
raise AttributeError("No body has been set")
try:
body = b"".join(app_iter)
finally:
iter_close(app_iter)
- if isinstance(body, text_type):
- raise _error_unicode_in_app_iter(app_iter, body)
+
self._app_iter = [body]
+
if len(body) == 0:
# if body-length is zero, we assume it's a HEAD response and
# leave content_length alone
@@ -535,11 +557,12 @@ class Response(object):
"Content-Length is different from actual app_iter length "
"(%r!=%r)" % (self.content_length, len(body))
)
+
return body
def _body__set(self, value=b""):
if not isinstance(value, bytes):
- if isinstance(value, text_type):
+ if isinstance(value, str):
msg = (
"You cannot set Response.body to a text object "
"(use Response.text)"
@@ -549,6 +572,7 @@ class Response(object):
value
)
raise TypeError(msg)
+
if self._app_iter is not None:
self.content_md5 = None
self._app_iter = [value]
@@ -573,6 +597,7 @@ class Response(object):
"""
# Note: UTF-8 is a content-type specific default for JSON
+
return json.loads(self.body.decode("UTF-8"))
def _json_body__set(self, value):
@@ -614,6 +639,7 @@ class Response(object):
Get/set the text value of the body using the ``charset`` of the
``Content-Type`` or the ``default_body_encoding``.
"""
+
if not self.charset and not self.default_body_encoding:
raise AttributeError(
"You cannot access Response.text unless charset or "
@@ -621,6 +647,7 @@ class Response(object):
)
decoding = self.charset or self.default_body_encoding
body = self.body
+
return body.decode(decoding, self.unicode_errors)
def _text__set(self, value):
@@ -629,7 +656,8 @@ class Response(object):
"You cannot access Response.text unless charset or "
"default_body_encoding is set"
)
- if not isinstance(value, text_type):
+
+ if not isinstance(value, str):
raise TypeError(
"You can only set Response.text to a unicode string "
"(not %s)" % type(value)
@@ -656,6 +684,7 @@ class Response(object):
body. If you passed in a list ``app_iter``, that ``app_iter`` will be
modified by writes.
"""
+
return ResponseBodyFile(self)
def _body_file__set(self, file):
@@ -670,15 +699,17 @@ class Response(object):
def write(self, text):
if not isinstance(text, bytes):
- if not isinstance(text, text_type):
+ if not isinstance(text, str):
msg = "You can only write str to a Response.body_file, not %s"
raise TypeError(msg % type(text))
+
if not self.charset:
msg = "You can only write text to Response if charset has " "been set"
raise TypeError(msg)
text = text.encode(self.charset)
text_len = len(text)
app_iter = self._app_iter
+
if not isinstance(app_iter, list):
try:
new_app_iter = self._app_iter = list(app_iter)
@@ -687,8 +718,10 @@ class Response(object):
app_iter = new_app_iter
self.content_length = sum(len(chunk) for chunk in app_iter)
app_iter.append(text)
+
if self.content_length is not None:
self.content_length += text_len
+
return text_len
#
@@ -702,6 +735,7 @@ class Response(object):
If ``body`` was set, this will create an ``app_iter`` from that
``body`` (a single-item list).
"""
+
return self._app_iter
def _app_iter__set(self, value):
@@ -788,23 +822,29 @@ class Response(object):
allows for a ``charset`` parameter.
"""
header = self.headers.get("Content-Type")
+
if not header:
return None
match = CHARSET_RE.search(header)
+
if match:
return match.group(1)
+
return None
def _charset__set(self, charset):
if charset is None:
self._charset__del()
+
return
header = self.headers.get("Content-Type", None)
+
if header is None:
raise AttributeError(
"You cannot set the charset when no " "content-type is defined"
)
match = CHARSET_RE.search(header)
+
if match:
header = header[: match.start()] + header[match.end() :]
header += "; charset=%s" % charset
@@ -812,10 +852,13 @@ class Response(object):
def _charset__del(self):
header = self.headers.pop("Content-Type", None)
+
if header is None:
# Don't need to remove anything
+
return
match = CHARSET_RE.search(header)
+
if match:
header = header[: match.start()] + header[match.end() :]
self.headers["Content-Type"] = header
@@ -851,19 +894,19 @@ class Response(object):
resp.content_type_params = params
"""
header = self.headers.get("Content-Type")
+
if not header:
return None
+
return header.split(";", 1)[0]
def _content_type__set(self, value):
if not value:
self._content_type__del()
+
return
else:
- if PY2 and isinstance(value, text_type):
- value = value.encode("latin-1")
-
- if not isinstance(value, string_types):
+ if not isinstance(value, str):
raise TypeError("content_type requires value to be of string_types")
content_type = value
@@ -882,6 +925,7 @@ class Response(object):
# otherwise add a charset if the content_type has a charset.
#
# We add the default charset if the content-type is "texty".
+
if new_charset and (
content_type == "text/html" or _content_type_has_charset(content_type)
):
@@ -911,20 +955,25 @@ class Response(object):
be applied otherwise.)
"""
params = self.headers.get("Content-Type", "")
+
if ";" not in params:
return {}
params = params.split(";", 1)[1]
result = {}
+
for match in _PARAM_RE.finditer(params):
result[match.group(1)] = match.group(2) or match.group(3) or ""
+
return result
def _content_type_params__set(self, value_dict):
if not value_dict:
self._content_type_params__del()
+
return
params = []
+
for k, v in sorted(value_dict.items()):
if not _OK_PARAM_RE.search(v):
v = '"%s"' % v.replace('"', '\\"')
@@ -959,7 +1008,6 @@ class Response(object):
secure=False,
httponly=False,
comment=None,
- expires=None,
overwrite=False,
samesite=None,
):
@@ -1025,25 +1073,6 @@ class Response(object):
If ``comment`` is ``None``, no ``Comment`` value will be sent in
the cookie.
- ``expires``
-
- A ``datetime.timedelta`` object representing an amount of time,
- ``datetime.datetime`` or ``None``. A non-``None`` value is used to
- generate the ``Expires`` value of the generated cookie. If
- ``max_age`` is not passed, but this value is not ``None``, it will
- influence the ``Max-Age`` header. If this value is ``None``, the
- ``Expires`` cookie value will be unset (unless ``max_age`` is set).
- If ``max_age`` is set, it will be used to generate the ``expires``
- and this value is ignored.
-
- If a ``datetime.datetime`` is provided it has to either be timezone
- aware or be based on UTC. ``datetime.datetime`` objects that are
- local time are not supported. Timezone aware ``datetime.datetime``
- objects are converted to UTC.
-
- This argument will be removed in future versions of WebOb (version
- 1.9).
-
``overwrite``
If this key is ``True``, before setting the cookie, unset any
@@ -1051,33 +1080,10 @@ class Response(object):
"""
- # Remove in WebOb 1.10
- if expires:
- warn_deprecation(
- 'Argument "expires" will be removed in a future '
- 'version of WebOb, please use "max_age".',
- 1.10,
- 1,
- )
-
if overwrite:
self.unset_cookie(name, strict=False)
- # If expires is set, but not max_age we set max_age to expires
- if not max_age and isinstance(expires, timedelta):
- max_age = expires
-
- # expires can also be a datetime
- if not max_age and isinstance(expires, datetime):
-
- # If expires has a timezone attached, convert it to UTC
- if expires.tzinfo and expires.utcoffset():
- expires = (expires - expires.utcoffset()).replace(tzinfo=None)
-
- max_age = expires - datetime.utcnow()
-
value = bytes_(value, "utf-8")
-
cookie = make_cookie(
name,
value,
@@ -1106,16 +1112,21 @@ class Response(object):
Unset a cookie with the given name (remove it from the response).
"""
existing = self.headers.getall("Set-Cookie")
+
if not existing and not strict:
return
cookies = Cookie()
+
for header in existing:
cookies.load(header)
- if isinstance(name, text_type):
+
+ if isinstance(name, str):
name = name.encode("utf8")
+
if name in cookies:
del cookies[name]
del self.headers["Set-Cookie"]
+
for m in cookies.values():
self.headerlist.append(("Set-Cookie", m.serialize()))
elif strict:
@@ -1128,11 +1139,14 @@ class Response(object):
If the ``resp`` is a :class:`webob.Response` object, then the
other object will be modified in-place.
"""
+
if not self.headers.get("Set-Cookie"):
return resp
+
if isinstance(resp, Response):
for header in self.headers.getall("Set-Cookie"):
resp.headers.add("Set-Cookie", header)
+
return resp
else:
c_headers = [h for h in self.headerlist if h[0].lower() == "set-cookie"]
@@ -1159,29 +1173,37 @@ class Response(object):
<http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9>`_).
"""
value = self.headers.get("cache-control", "")
+
if self._cache_control_obj is None:
self._cache_control_obj = CacheControl.parse(
value, updates_to=self._update_cache_control, type="response"
)
self._cache_control_obj.header_value = value
+
if self._cache_control_obj.header_value != value:
new_obj = CacheControl.parse(value, type="response")
self._cache_control_obj.properties.clear()
self._cache_control_obj.properties.update(new_obj.properties)
self._cache_control_obj.header_value = value
+
return self._cache_control_obj
def _cache_control__set(self, value):
# This actually becomes a copy
+
if not value:
value = ""
+
if isinstance(value, dict):
value = CacheControl(value, "response")
- if isinstance(value, text_type):
+
+ if isinstance(value, str):
value = str(value)
+
if isinstance(value, str):
if self._cache_control_obj is None:
self.headers["Cache-Control"] = value
+
return
value = CacheControl.parse(value, "response")
cache = self.cache_control
@@ -1193,6 +1215,7 @@ class Response(object):
def _update_cache_control(self, prop_dict):
value = serialize_cache_control(prop_dict)
+
if not value:
if "Cache-Control" in self.headers:
del self.headers["Cache-Control"]
@@ -1216,11 +1239,13 @@ class Response(object):
expire in the given seconds, and any other attributes are used
for ``cache_control`` (e.g., ``private=True``).
"""
+
if seconds is True:
seconds = 0
elif isinstance(seconds, timedelta):
seconds = timedelta_to_seconds(seconds)
cache_control = self.cache_control
+
if seconds is None:
pass
elif not seconds:
@@ -1235,6 +1260,7 @@ class Response(object):
cache_control.post_check = 0
cache_control.pre_check = 0
self.expires = datetime.utcnow()
+
if "last-modified" not in self.headers:
self.last_modified = datetime.utcnow()
self.pragma = "no-cache"
@@ -1243,6 +1269,7 @@ class Response(object):
cache_control.max_age = seconds
self.expires = datetime.utcnow() + timedelta(seconds=seconds)
self.pragma = None
+
for name, value in kw.items():
setattr(cache_control, name, value)
@@ -1258,11 +1285,15 @@ class Response(object):
``identity`` are supported).
"""
assert encoding in ("identity", "gzip"), "Unknown encoding: %r" % encoding
+
if encoding == "identity":
self.decode_content()
+
return
+
if self.content_encoding == "gzip":
return
+
if lazy:
self.app_iter = gzip_app_iter(self._app_iter)
self.content_length = None
@@ -1273,12 +1304,15 @@ class Response(object):
def decode_content(self):
content_encoding = self.content_encoding or "identity"
+
if content_encoding == "identity":
return
+
if content_encoding not in ("gzip", "deflate"):
raise ValueError(
"I don't know how to decode the content %s" % content_encoding
)
+
if content_encoding == "gzip":
from gzip import GzipFile
from io import BytesIO
@@ -1301,13 +1335,15 @@ class Response(object):
If ``set_content_md5`` is ``True``, sets ``self.content_md5`` as well.
"""
+
if body is None:
body = self.body
md5_digest = md5(body).digest()
md5_digest = b64encode(md5_digest)
md5_digest = md5_digest.replace(b"\n", b"")
- md5_digest = native_(md5_digest)
+ md5_digest = text_(md5_digest)
self.etag = md5_digest.strip("=")
+
if set_content_md5:
self.content_md5 = md5_digest
@@ -1317,10 +1353,12 @@ class Response(object):
return value
new_location = urlparse.urljoin(_request_uri(environ), value)
+
return new_location
def _abs_headerlist(self, environ):
# Build the headerlist, if we have a Location header, make it absolute
+
return [
(k, v)
if k.lower() != "location"
@@ -1336,15 +1374,19 @@ class Response(object):
"""
WSGI application interface
"""
+
if self.conditional_response:
return self.conditional_response_app(environ, start_response)
headerlist = self._abs_headerlist(environ)
start_response(self.status, headerlist)
+
if environ["REQUEST_METHOD"] == "HEAD":
# Special case here...
+
return EmptyResponse(self._app_iter)
+
return self._app_iter
_safe_methods = ("GET", "HEAD")
@@ -1365,15 +1407,20 @@ class Response(object):
headerlist = self._abs_headerlist(environ)
method = environ.get("REQUEST_METHOD", "GET")
+
if method in self._safe_methods:
status304 = False
+
if req.if_none_match and self.etag:
status304 = self.etag in req.if_none_match
elif req.if_modified_since and self.last_modified:
status304 = self.last_modified <= req.if_modified_since
+
if status304:
start_response("304 Not Modified", filter_headers(headerlist))
+
return EmptyResponse(self._app_iter)
+
if (
req.range
and self in req.if_range
@@ -1383,6 +1430,7 @@ class Response(object):
and self.content_length is not None
):
content_range = req.range.content_range(self.content_length)
+
if content_range is None:
iter_close(self._app_iter)
body = bytes_("Requested range not satisfiable: %s" % req.range)
@@ -1395,11 +1443,14 @@ class Response(object):
("Content-Type", "text/plain"),
] + filter_headers(headerlist)
start_response("416 Requested Range Not Satisfiable", headerlist)
+
if method == "HEAD":
return ()
+
return [body]
else:
app_iter = self.app_iter_range(content_range.start, content_range.stop)
+
if app_iter is not None:
# the following should be guaranteed by
# Range.range_for_length(length)
@@ -1412,13 +1463,17 @@ class Response(object):
("Content-Range", str(content_range)),
] + filter_headers(headerlist, ("content-length",))
start_response("206 Partial Content", headerlist)
+
if method == "HEAD":
return EmptyResponse(app_iter)
+
return app_iter
start_response(self.status, headerlist)
+
if method == "HEAD":
return EmptyResponse(self._app_iter)
+
return self._app_iter
def app_iter_range(self, start, stop):
@@ -1427,8 +1482,10 @@ class Response(object):
serves up only the given ``start:stop`` range.
"""
app_iter = self._app_iter
+
if hasattr(app_iter, "app_iter_range"):
return app_iter.app_iter_range(start, stop)
+
return AppIterRange(app_iter, start, stop)
@@ -1439,12 +1496,13 @@ def filter_headers(hlist, remove_headers=("content-length", "content-type")):
def iter_file(file, block_size=1 << 18): # 256Kb
while True:
data = file.read(block_size)
+
if not data:
break
yield data
-class ResponseBodyFile(object):
+class ResponseBodyFile:
mode = "wb"
closed = False
@@ -1467,6 +1525,7 @@ class ResponseBodyFile(object):
"""
Write a sequence of lines to the response.
"""
+
for item in seq:
self.write(item)
@@ -1480,13 +1539,14 @@ class ResponseBodyFile(object):
"""
Provide the current location where we are going to start writing.
"""
+
if not self.response.has_body:
return 0
return sum([len(chunk) for chunk in self.response.app_iter])
-class AppIterRange(object):
+class AppIterRange:
"""
Wraps an ``app_iter``, returning just a range of bytes.
"""
@@ -1504,17 +1564,21 @@ class AppIterRange(object):
def _skip_start(self):
start, stop = self.start, self.stop
+
for chunk in self.app_iter:
self._pos += len(chunk)
+
if self._pos < start:
continue
elif self._pos == start:
return b""
else:
chunk = chunk[start - self._pos :]
+
if stop is not None and self._pos > stop:
chunk = chunk[: stop - self._pos]
assert len(chunk) == stop - start
+
return chunk
else:
raise StopIteration()
@@ -1522,8 +1586,10 @@ class AppIterRange(object):
def next(self):
if self._pos < self.start:
# need to skip some leading bytes
+
return self._skip_start()
stop = self.stop
+
if stop is not None and self._pos >= stop:
raise StopIteration
@@ -1541,7 +1607,7 @@ class AppIterRange(object):
iter_close(self.app_iter)
-class EmptyResponse(object):
+class EmptyResponse:
"""
An empty WSGI response.
@@ -1587,24 +1653,23 @@ def _request_uri(environ):
url += environ["HTTP_HOST"]
else:
url += environ["SERVER_NAME"] + ":" + environ["SERVER_PORT"]
+
if url.endswith(":80") and environ["wsgi.url_scheme"] == "http":
url = url[:-3]
elif url.endswith(":443") and environ["wsgi.url_scheme"] == "https":
url = url[:-4]
- if PY2:
- script_name = environ.get("SCRIPT_NAME", "/")
- path_info = environ.get("PATH_INFO", "")
- else:
- script_name = bytes_(environ.get("SCRIPT_NAME", "/"), "latin-1")
- path_info = bytes_(environ.get("PATH_INFO", ""), "latin-1")
+ script_name = bytes_(environ.get("SCRIPT_NAME", "/"), "latin-1")
+ path_info = bytes_(environ.get("PATH_INFO", ""), "latin-1")
url += url_quote(script_name)
qpath_info = url_quote(path_info)
+
if "SCRIPT_NAME" not in environ:
url += qpath_info[1:]
else:
url += qpath_info
+
return url
@@ -1621,6 +1686,7 @@ def gzip_app_iter(app_iter):
)
yield _gzip_header
+
for item in app_iter:
size += len(item)
crc = zlib.crc32(item, crc) & 0xFFFFFFFF
@@ -1629,21 +1695,13 @@ def gzip_app_iter(app_iter):
# small enough; it buffers the input for the next iteration or for a
# flush.
result = compress.compress(item)
+
if result:
yield result
# Similarly, flush may also not yield a value.
result = compress.flush()
+
if result:
yield result
yield struct.pack("<2L", crc, size & 0xFFFFFFFF)
-
-
-def _error_unicode_in_app_iter(app_iter, body):
- app_iter_repr = repr(app_iter)
- if len(app_iter_repr) > 50:
- app_iter_repr = app_iter_repr[:30] + "..." + app_iter_repr[-10:]
- raise TypeError(
- "An item of the app_iter (%s) was text, causing a "
- "text body: %r" % (app_iter_repr, body)
- )
diff --git a/src/webob/static.py b/src/webob/static.py
index 52ad566..114bfd0 100644
--- a/src/webob/static.py
+++ b/src/webob/static.py
@@ -16,7 +16,7 @@ mimetypes.add_type("image/x-icon", ".ico") # not among defaults
BLOCK_SIZE = 1 << 16
-class FileApp(object):
+class FileApp:
"""An application that will send the file at the given filename.
Adds a mime type based on `mimetypes.guess_type()`.
@@ -38,13 +38,13 @@ class FileApp(object):
return exc.HTTPMethodNotAllowed("You cannot %s a file" % req.method)
try:
stat = os.stat(self.filename)
- except (IOError, OSError) as e:
+ except OSError as e:
msg = "Can't open %r: %s" % (self.filename, e)
return exc.HTTPNotFound(comment=msg)
try:
file = self._open(self.filename, "rb")
- except (IOError, OSError) as e:
+ except OSError as e:
msg = "You are not permitted to view this file (%s)" % e
return exc.HTTPForbidden(msg)
@@ -58,11 +58,11 @@ class FileApp(object):
content_length=stat.st_size,
last_modified=stat.st_mtime,
# @@ etag
- **self.kw
+ **self.kw,
).conditional_response_app
-class FileIter(object):
+class FileIter:
def __init__(self, file):
self.file = file
@@ -104,7 +104,7 @@ class FileIter(object):
__iter__ = app_iter_range
-class DirectoryApp(object):
+class DirectoryApp:
"""An application that serves up the files in a given directory.
This will serve index files (by default ``index.html``), or set
@@ -123,7 +123,7 @@ class DirectoryApp(object):
if not self.path.endswith(os.path.sep):
self.path += os.path.sep
if not os.path.isdir(self.path):
- raise IOError("Path does not exist or is not directory: %r" % self.path)
+ raise OSError("Path does not exist or is not directory: %r" % self.path)
self.index_page = index_page
self.hide_index_with_redirect = hide_index_with_redirect
self.fileapp_kw = kw
diff --git a/src/webob/util.py b/src/webob/util.py
index 806ac63..ccabe5b 100644
--- a/src/webob/util.py
+++ b/src/webob/util.py
@@ -1,10 +1,56 @@
import warnings
-from webob.compat import escape, string_types, text_, text_type
-
+from webob.compat import escape
from webob.headers import _trans_key
+def unquote(string):
+ if not string:
+ return b""
+ res = string.split(b"%")
+
+ if len(res) != 1:
+ string = res[0]
+
+ for item in res[1:]:
+ string += bytes([int(item[:2], 16)]) + item[2:]
+
+ return string
+
+
+def url_unquote(s):
+ return unquote(s.encode("ascii")).decode("latin-1")
+
+
+def parse_qsl_text(qs, encoding="utf-8"):
+ qs = qs.encode("latin-1")
+ qs = qs.replace(b"+", b" ")
+ pairs = [s2 for s1 in qs.split(b"&") for s2 in s1.split(b";") if s2]
+
+ for name_value in pairs:
+ nv = name_value.split(b"=", 1)
+
+ if len(nv) != 2:
+ nv.append("")
+ name = unquote(nv[0])
+ value = unquote(nv[1])
+ yield (name.decode(encoding), value.decode(encoding))
+
+
+def text_(s, encoding="latin-1", errors="strict"):
+ if isinstance(s, bytes):
+ return str(s, encoding, errors)
+
+ return s
+
+
+def bytes_(s, encoding="latin-1", errors="strict"):
+ if isinstance(s, str):
+ return s.encode(encoding, errors)
+
+ return s
+
+
def html_escape(s):
"""HTML-escape a string or object
@@ -15,20 +61,21 @@ def html_escape(s):
None is treated specially, and returns the empty string.
"""
+
if s is None:
return ""
__html__ = getattr(s, "__html__", None)
+
if __html__ is not None and callable(__html__):
return s.__html__()
- if not isinstance(s, string_types):
- __unicode__ = getattr(s, "__unicode__", None)
- if __unicode__ is not None and callable(__unicode__):
- s = s.__unicode__()
- else:
- s = str(s)
+
+ if not isinstance(s, str):
+ s = str(s)
s = escape(s, True)
- if isinstance(s, text_type):
+
+ if isinstance(s, str):
s = s.encode("ascii", "xmlcharrefreplace")
+
return text_(s)
@@ -40,6 +87,7 @@ def header_docstring(header, rfc_section):
major_section,
rfc_section,
)
+
return "Gets and sets the ``%s`` header (`HTTP spec section %s <%s>`_)." % (
header,
rfc_section,
@@ -49,6 +97,7 @@ def header_docstring(header, rfc_section):
def warn_deprecation(text, version, stacklevel):
# version specifies when to start raising exceptions instead of warnings
+
if version in ("1.2", "1.3", "1.4", "1.5", "1.6", "1.7"):
raise DeprecationWarning(text)
else:
@@ -129,41 +178,3 @@ status_generic_reasons = {
4: "Unknown Client Error",
5: "Unknown Server Error",
}
-
-try:
- # py3.3+ have native comparison support
- from hmac import compare_digest
-except ImportError: # pragma: nocover (Python 2.7.7 backported this)
- compare_digest = None
-
-
-def strings_differ(string1, string2, compare_digest=compare_digest):
- """Check whether two strings differ while avoiding timing attacks.
-
- This function returns True if the given strings differ and False
- if they are equal. It's careful not to leak information about *where*
- they differ as a result of its running time, which can be very important
- to avoid certain timing-related crypto attacks:
-
- http://seb.dbzteam.org/crypto/python-oauth-timing-hmac.pdf
-
- .. versionchanged:: 1.5
- Support :func:`hmac.compare_digest` if it is available (Python 2.7.7+
- and Python 3.3+).
-
- """
- len_eq = len(string1) == len(string2)
- if len_eq:
- invalid_bits = 0
- left = string1
- else:
- invalid_bits = 1
- left = string2
- right = string2
-
- if compare_digest is not None:
- invalid_bits += not compare_digest(left, right)
- else:
- for a, b in zip(left, right):
- invalid_bits += a != b
- return invalid_bits != 0
diff --git a/tests/conftest.py b/tests/conftest.py
index e84fabe..f8a1073 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,13 +1,15 @@
-import pytest
-import threading
-import random
-import logging
from contextlib import contextmanager
+import logging
+import random
+import threading
+from wsgiref.simple_server import (
+ ServerHandler,
+ WSGIRequestHandler,
+ WSGIServer,
+ make_server,
+)
-from wsgiref.simple_server import make_server
-from wsgiref.simple_server import WSGIRequestHandler
-from wsgiref.simple_server import WSGIServer
-from wsgiref.simple_server import ServerHandler
+import pytest
log = logging.getLogger(__name__)
ServerHandler.handle_error = lambda: None
@@ -62,7 +64,7 @@ def serve():
log.debug("shutting server down")
server.shutdown()
worker.join(1)
- if worker.isAlive():
+ if worker.is_alive():
log.warning("worker is hanged")
else:
log.debug("server stopped")
diff --git a/tests/performance_test.py b/tests/performance_test.py
index 4c4917f..0d5f932 100644
--- a/tests/performance_test.py
+++ b/tests/performance_test.py
@@ -1,6 +1,5 @@
#!/usr/bin/env python
from webob.response import Response
-from webob.compat import PY2
def make_middleware(app):
@@ -21,9 +20,9 @@ def simple_app(environ, start_response):
if __name__ == "__main__":
- import sys
import os
import signal
+ import sys
if sys.argv[1:]:
arg = sys.argv[1]
@@ -31,8 +30,8 @@ if __name__ == "__main__":
arg = None
if arg in ["open", "run"]:
import subprocess
- import webbrowser
import time
+ import webbrowser
os.environ["SHOW_OUTPUT"] = "0"
proc = subprocess.Popen([sys.executable, __file__])
@@ -43,10 +42,7 @@ if __name__ == "__main__":
print("Hit ^C to end")
try:
while 1:
- if PY2:
- raw_input() # noqa: F821
- else:
- input()
+ input()
finally:
os.kill(proc.pid, signal.SIGKILL)
else:
diff --git a/tests/test_acceptparse.py b/tests/test_acceptparse.py
index e5ff5ea..736a682 100644
--- a/tests/test_acceptparse.py
+++ b/tests/test_acceptparse.py
@@ -2,6 +2,7 @@ import re
import warnings
import pytest
+
from webob.acceptparse import (
Accept,
AcceptCharset,
@@ -40,7 +41,7 @@ IGNORE_ITER = "ignore:.*__iter__.*"
IGNORE_MIMEACCEPT = "ignore:.*MIMEAccept.*"
-class Test_ItemNWeightRe(object):
+class Test_ItemNWeightRe:
@pytest.mark.parametrize(
"header_value",
[
@@ -95,7 +96,7 @@ class Test_ItemNWeightRe(object):
assert re.match("^" + regex + "$", header_value, re.VERBOSE).groups() == groups
-class Test_List1OrMoreCompiledRe(object):
+class Test_List1OrMoreCompiledRe:
@pytest.mark.parametrize(
"header_value",
[
@@ -138,7 +139,7 @@ class Test_List1OrMoreCompiledRe(object):
assert regex.match(header_value)
-class TestAccept(object):
+class TestAccept:
@pytest.mark.parametrize(
"value",
[
@@ -465,7 +466,7 @@ class TestAccept(object):
Accept.parse_offer(offer)
-class TestAcceptValidHeader(object):
+class TestAcceptValidHeader:
def test_parse__inherited(self):
returned = AcceptValidHeader.parse(
value=(
@@ -541,7 +542,7 @@ class TestAcceptValidHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptValidHeader(header_value="text/html")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -562,7 +563,7 @@ class TestAcceptValidHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptValidHeader(header_value=",\t ,i/j, k/l;q=0.333,")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -655,7 +656,7 @@ class TestAcceptValidHeader(object):
def test___add___other_type_with_valid___str___not_empty(self):
header = ",\t ,i/j, k/l;q=0.333,"
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -759,7 +760,7 @@ class TestAcceptValidHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptValidHeader(header_value="a/b")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -779,7 +780,7 @@ class TestAcceptValidHeader(object):
def test___radd___other_type_with_valid___str___empty(self):
right_operand = AcceptValidHeader(header_value=",\t ,i/j, k/l;q=0.333,")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -872,7 +873,7 @@ class TestAcceptValidHeader(object):
def test___radd___other_type_with_valid___str___not_empty(self):
header = ",\t ,i/j, k/l;q=0.333,"
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -1238,7 +1239,7 @@ class TestAcceptValidHeader(object):
assert accept.quality("foo/bar") is None
-class TestAcceptNoHeader(object):
+class TestAcceptNoHeader:
def test_parse__inherited(self):
returned = AcceptNoHeader.parse(
value=(
@@ -1290,7 +1291,7 @@ class TestAcceptNoHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -1308,7 +1309,7 @@ class TestAcceptNoHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -1397,7 +1398,7 @@ class TestAcceptNoHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -1482,7 +1483,7 @@ class TestAcceptNoHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -1497,7 +1498,7 @@ class TestAcceptNoHeader(object):
assert result.header_value == ""
def test___radd___other_type_with_valid___str___empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -1586,7 +1587,7 @@ class TestAcceptNoHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -1656,7 +1657,7 @@ class TestAcceptNoHeader(object):
assert returned == 1.0
-class TestAcceptInvalidHeader(object):
+class TestAcceptInvalidHeader:
def test_parse__inherited(self):
returned = AcceptInvalidHeader.parse(
value=(
@@ -1709,7 +1710,7 @@ class TestAcceptInvalidHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptInvalidHeader(header_value="invalid header")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -1727,7 +1728,7 @@ class TestAcceptInvalidHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptInvalidHeader(header_value=", ")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -1816,7 +1817,7 @@ class TestAcceptInvalidHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -1900,7 +1901,7 @@ class TestAcceptInvalidHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptInvalidHeader(header_value=", ")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -1917,7 +1918,7 @@ class TestAcceptInvalidHeader(object):
def test___radd___other_type_with_valid___str___empty(self):
right_operand = AcceptInvalidHeader(header_value="invalid header")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -2006,7 +2007,7 @@ class TestAcceptInvalidHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "a/b;q=0.5, c/d;p1=1;q=0, e/f, g/h;p1=1;q=1;e1=1"
@@ -2079,7 +2080,7 @@ class TestAcceptInvalidHeader(object):
assert returned == 1.0
-class TestCreateAcceptHeader(object):
+class TestCreateAcceptHeader:
def test_header_value_is_None(self):
header_value = None
returned = create_accept_header(header_value=header_value)
@@ -2108,7 +2109,7 @@ class TestCreateAcceptHeader(object):
assert returned2._header_value == returned._header_value
-class TestAcceptProperty(object):
+class TestAcceptProperty:
def test_fget_header_is_valid(self):
header_value = 'text/html;p1="1";p2=v2;q=0.9;e1="1";e2, audio/basic'
request = Request.blank("/", environ={"HTTP_ACCEPT": header_value})
@@ -2243,7 +2244,7 @@ class TestAcceptProperty(object):
request = Request.blank("/", environ={"HTTP_ACCEPT": "text/html"})
property_ = accept_property()
- class Other(object):
+ class Other:
def __str__(self):
return header_value
@@ -2287,7 +2288,7 @@ class TestAcceptProperty(object):
assert "HTTP_ACCEPT" not in request.environ
-class TestAcceptCharset(object):
+class TestAcceptCharset:
@pytest.mark.parametrize(
"value",
[
@@ -2348,7 +2349,7 @@ class TestAcceptCharset(object):
assert list_of_returned == expected_list
-class TestAcceptCharsetValidHeader(object):
+class TestAcceptCharsetValidHeader:
def test_parse__inherited(self):
returned = AcceptCharsetValidHeader.parse(
value=",iso-8859-5 ; q=0.333 , ,utf-8,unicode-1-1 ;q=0.90,"
@@ -2404,7 +2405,7 @@ class TestAcceptCharsetValidHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptCharsetValidHeader(header_value="iso-8859-5")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -2444,7 +2445,7 @@ class TestAcceptCharsetValidHeader(object):
def test___add___other_type_with_valid___str__(self):
left_operand = AcceptCharsetValidHeader(header_value=",\t ,iso-8859-5;q=0.333,")
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -2534,7 +2535,7 @@ class TestAcceptCharsetValidHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptCharsetValidHeader(header_value="iso-8859-5")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -2579,7 +2580,7 @@ class TestAcceptCharsetValidHeader(object):
header_value=",\t ,iso-8859-5;q=0.333,"
)
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -2704,7 +2705,7 @@ class TestAcceptCharsetValidHeader(object):
assert accept.quality("iso-8859-5") is None
-class TestAcceptCharsetNoHeader(object):
+class TestAcceptCharsetNoHeader:
def test_parse__inherited(self):
returned = AcceptCharsetNoHeader.parse(
value=",iso-8859-5 ; q=0.333 , ,utf-8,unicode-1-1 ;q=0.90,"
@@ -2743,7 +2744,7 @@ class TestAcceptCharsetNoHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptCharsetNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -2778,7 +2779,7 @@ class TestAcceptCharsetNoHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str__(self):
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -2847,7 +2848,7 @@ class TestAcceptCharsetNoHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptCharsetNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -2882,7 +2883,7 @@ class TestAcceptCharsetNoHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str__(self):
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -2928,7 +2929,7 @@ class TestAcceptCharsetNoHeader(object):
assert returned == 1.0
-class TestAcceptCharsetInvalidHeader(object):
+class TestAcceptCharsetInvalidHeader:
def test_parse__inherited(self):
returned = AcceptCharsetInvalidHeader.parse(
value=",iso-8859-5 ; q=0.333 , ,utf-8,unicode-1-1 ;q=0.90,"
@@ -2963,7 +2964,7 @@ class TestAcceptCharsetInvalidHeader(object):
@pytest.mark.parametrize("str_", ["", "UTF/8"])
def test___add___other_type_with_invalid___str__(self, str_):
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -2997,7 +2998,7 @@ class TestAcceptCharsetInvalidHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_valid_header_value(self):
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -3057,7 +3058,7 @@ class TestAcceptCharsetInvalidHeader(object):
@pytest.mark.parametrize("str_", ["", "UTF/8"])
def test___radd___other_type_with_invalid___str__(self, str_):
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -3091,7 +3092,7 @@ class TestAcceptCharsetInvalidHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_valid_header_value(self):
- class Other(object):
+ class Other:
def __str__(self):
return "UTF-7;q=0.5, unicode-1-1;q=0, UTF-8"
@@ -3137,7 +3138,7 @@ class TestAcceptCharsetInvalidHeader(object):
assert returned == 1.0
-class TestCreateAcceptCharsetHeader(object):
+class TestCreateAcceptCharsetHeader:
def test_header_value_is_valid(self):
header_value = "iso-8859-5, unicode-1-1;q=0.8"
returned = create_accept_charset_header(header_value=header_value)
@@ -3166,7 +3167,7 @@ class TestCreateAcceptCharsetHeader(object):
assert returned2._header_value == returned._header_value
-class TestAcceptCharsetProperty(object):
+class TestAcceptCharsetProperty:
def test_fget_header_is_None(self):
request = Request.blank("/", environ={"HTTP_ACCEPT_CHARSET": None})
property_ = accept_charset_property()
@@ -3238,7 +3239,7 @@ class TestAcceptCharsetProperty(object):
request = Request.blank("/", environ={"HTTP_ACCEPT_CHARSET": ""})
property_ = accept_charset_property()
- class Other(object):
+ class Other:
def __str__(self):
return "utf-8;q=0.5, iso-8859-5;q=0, utf-7"
@@ -3286,7 +3287,7 @@ class TestAcceptCharsetProperty(object):
assert "HTTP_ACCEPT_CHARSET" not in request.environ
-class TestAcceptEncoding(object):
+class TestAcceptEncoding:
@pytest.mark.parametrize(
"value",
[
@@ -3344,7 +3345,7 @@ class TestAcceptEncoding(object):
assert list_of_returned == expected_list
-class TestAcceptEncodingValidHeader(object):
+class TestAcceptEncodingValidHeader:
def test_parse__inherited(self):
returned = AcceptEncodingValidHeader.parse(
value=",,\t gzip;q=1.0, identity; q=0.5, *;q=0 \t ,"
@@ -3385,7 +3386,7 @@ class TestAcceptEncodingValidHeader(object):
def test___add___other_type_with_invalid___str__(self):
left_operand = AcceptEncodingValidHeader(header_value="gzip")
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -3406,7 +3407,7 @@ class TestAcceptEncodingValidHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptEncodingValidHeader(header_value="gzip")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -3444,7 +3445,7 @@ class TestAcceptEncodingValidHeader(object):
def test___add___other_type_with_valid___str___not_empty(self):
header = ",\t ,gzip, identity;q=0.333,"
- class Other(object):
+ class Other:
def __str__(self):
return "compress;q=0.5, deflate;q=0, *"
@@ -3532,7 +3533,7 @@ class TestAcceptEncodingValidHeader(object):
def test___radd___other_type_with_invalid___str__(self):
right_operand = AcceptEncodingValidHeader(header_value="gzip")
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -3552,7 +3553,7 @@ class TestAcceptEncodingValidHeader(object):
def test___radd___other_type_with_valid___str___empty(self):
right_operand = AcceptEncodingValidHeader(header_value="gzip")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -3590,7 +3591,7 @@ class TestAcceptEncodingValidHeader(object):
def test___radd___other_type_with_valid___str___not_empty(self):
header = ",\t ,gzip, identity;q=0.333,"
- class Other(object):
+ class Other:
def __str__(self):
return "compress;q=0.5, deflate;q=0, *"
@@ -3714,7 +3715,7 @@ class TestAcceptEncodingValidHeader(object):
assert accept.quality("compress") is None
-class TestAcceptEncodingNoHeader(object):
+class TestAcceptEncodingNoHeader:
def test_parse__inherited(self):
returned = AcceptEncodingNoHeader.parse(
value=",,\t gzip;q=1.0, identity; q=0.5, *;q=0 \t ,"
@@ -3743,7 +3744,7 @@ class TestAcceptEncodingNoHeader(object):
def test___add___other_type_with_invalid___str__(self):
left_operand = AcceptEncodingNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -3761,7 +3762,7 @@ class TestAcceptEncodingNoHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptEncodingNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -3795,7 +3796,7 @@ class TestAcceptEncodingNoHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "compress;q=0.5, deflate;q=0, *"
@@ -3867,7 +3868,7 @@ class TestAcceptEncodingNoHeader(object):
def test___radd___other_type_with_invalid___str__(self):
right_operand = AcceptEncodingNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -3882,7 +3883,7 @@ class TestAcceptEncodingNoHeader(object):
assert result.header_value == ""
def test___radd___other_type_with_valid___str___empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -3916,7 +3917,7 @@ class TestAcceptEncodingNoHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "compress;q=0.5, deflate;q=0, *"
@@ -3962,7 +3963,7 @@ class TestAcceptEncodingNoHeader(object):
assert returned == 1.0
-class TestAcceptEncodingInvalidHeader(object):
+class TestAcceptEncodingInvalidHeader:
def test_parse__inherited(self):
returned = AcceptEncodingInvalidHeader.parse(
value=",,\t gzip;q=1.0, identity; q=0.5, *;q=0 \t ,"
@@ -3992,7 +3993,7 @@ class TestAcceptEncodingInvalidHeader(object):
def test___add___other_type_with_invalid___str__(self):
left_operand = AcceptEncodingInvalidHeader(header_value="invalid header")
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -4010,7 +4011,7 @@ class TestAcceptEncodingInvalidHeader(object):
def test___add___other_type_with_valid___str___empty(self):
left_operand = AcceptEncodingInvalidHeader(header_value=", ")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -4044,7 +4045,7 @@ class TestAcceptEncodingInvalidHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "*, compress;q=0.5, deflate;q=0"
@@ -4115,7 +4116,7 @@ class TestAcceptEncodingInvalidHeader(object):
def test___radd___other_type_with_invalid___str__(self):
right_operand = AcceptEncodingInvalidHeader(header_value="gzip;q= 1")
- class Other(object):
+ class Other:
def __str__(self):
return ", "
@@ -4132,7 +4133,7 @@ class TestAcceptEncodingInvalidHeader(object):
def test___radd___other_type_with_valid___str___empty(self):
right_operand = AcceptEncodingInvalidHeader(header_value=", ")
- class Other(object):
+ class Other:
def __str__(self):
return ""
@@ -4166,7 +4167,7 @@ class TestAcceptEncodingInvalidHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str___not_empty(self):
- class Other(object):
+ class Other:
def __str__(self):
return "compress;q=0.5, deflate;q=0, *"
@@ -4212,7 +4213,7 @@ class TestAcceptEncodingInvalidHeader(object):
assert returned == 1.0
-class TestCreateAcceptEncodingHeader(object):
+class TestCreateAcceptEncodingHeader:
def test_header_value_is_None(self):
header_value = None
returned = create_accept_encoding_header(header_value=header_value)
@@ -4241,7 +4242,7 @@ class TestCreateAcceptEncodingHeader(object):
assert returned2._header_value == returned._header_value
-class TestAcceptEncodingProperty(object):
+class TestAcceptEncodingProperty:
def test_fget_header_is_None(self):
request = Request.blank("/", environ={"HTTP_ACCEPT_ENCODING": None})
property_ = accept_encoding_property()
@@ -4310,7 +4311,7 @@ class TestAcceptEncodingProperty(object):
request = Request.blank("/", environ={"HTTP_ACCEPT_ENCODING": ""})
property_ = accept_encoding_property()
- class Other(object):
+ class Other:
def __str__(self):
return "gzip;q=0.5, compress;q=0, deflate"
@@ -4358,7 +4359,7 @@ class TestAcceptEncodingProperty(object):
assert "HTTP_ACCEPT_ENCODING" not in request.environ
-class TestAcceptLanguage(object):
+class TestAcceptLanguage:
@pytest.mark.parametrize(
"value",
[
@@ -4427,7 +4428,7 @@ class TestAcceptLanguage(object):
assert list_of_returned == expected_list
-class TestAcceptLanguageValidHeader(object):
+class TestAcceptLanguageValidHeader:
@pytest.mark.parametrize("header_value", ["", ", da;q=0.2, en-gb;q=0.3 "])
def test___init___invalid_header(self, header_value):
with pytest.raises(ValueError):
@@ -4485,7 +4486,7 @@ class TestAcceptLanguageValidHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptLanguageValidHeader(header_value="en")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -4513,7 +4514,7 @@ class TestAcceptLanguageValidHeader(object):
def test___add___other_type_with_valid___str__(self):
header = ",\t ,de, zh-Hans;q=0.333,"
- class Other(object):
+ class Other:
def __str__(self):
return "en-gb;q=0.5, fr;q=0, es"
@@ -4635,7 +4636,7 @@ class TestAcceptLanguageValidHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptLanguageValidHeader(header_value="en")
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -4668,7 +4669,7 @@ class TestAcceptLanguageValidHeader(object):
header_value=",\t ,de, zh-Hans;q=0.333,"
)
- class Other(object):
+ class Other:
def __str__(self):
return "en-gb;q=0.5, fr;q=0, es"
@@ -5245,7 +5246,7 @@ class TestAcceptLanguageValidHeader(object):
assert returned == expected_returned
-class TestAcceptLanguageNoHeader(object):
+class TestAcceptLanguageNoHeader:
def test___init__(self):
instance = AcceptLanguageNoHeader()
assert instance.header_value is None
@@ -5273,7 +5274,7 @@ class TestAcceptLanguageNoHeader(object):
def test___add___other_type_with_invalid___str__(self, str_):
left_operand = AcceptLanguageNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -5296,7 +5297,7 @@ class TestAcceptLanguageNoHeader(object):
assert result.header_value == value_as_header
def test___add___other_type_with_valid___str__(self):
- class Other(object):
+ class Other:
def __str__(self):
return "en-gb;q=0.5, fr;q=0, es"
@@ -5364,7 +5365,7 @@ class TestAcceptLanguageNoHeader(object):
def test___radd___other_type_with_invalid___str__(self, str_):
right_operand = AcceptLanguageNoHeader()
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -5387,7 +5388,7 @@ class TestAcceptLanguageNoHeader(object):
assert result.header_value == value_as_header
def test___radd___other_type_with_valid___str__(self):
- class Other(object):
+ class Other:
def __str__(self):
return "en-gb;q=0.5, fr;q=0, es"
@@ -5456,7 +5457,7 @@ class TestAcceptLanguageNoHeader(object):
assert returned == 1.0
-class TestAcceptLanguageInvalidHeader(object):
+class TestAcceptLanguageInvalidHeader:
def test___init__(self):
header_value = "invalid header"
instance = AcceptLanguageInvalidHeader(header_value=header_value)
@@ -5480,7 +5481,7 @@ class TestAcceptLanguageInvalidHeader(object):
@pytest.mark.parametrize("str_", ["", "en_gb"])
def test___add___other_type_with_invalid___str__(self, str_):
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -5494,7 +5495,7 @@ class TestAcceptLanguageInvalidHeader(object):
assert result.header_value == "en"
def test___add___other_type_valid_header_value(self):
- class Other(object):
+ class Other:
def __str__(self):
return "en"
@@ -5552,7 +5553,7 @@ class TestAcceptLanguageInvalidHeader(object):
@pytest.mark.parametrize("str_", ["", "en_gb"])
def test___radd___other_type_with_invalid___str__(self, str_):
- class Other(object):
+ class Other:
def __str__(self):
return str_
@@ -5566,7 +5567,7 @@ class TestAcceptLanguageInvalidHeader(object):
assert result.header_value == "en"
def test___radd___other_type_valid_header_value(self):
- class Other(object):
+ class Other:
def __str__(self):
return "en"
@@ -5634,7 +5635,7 @@ class TestAcceptLanguageInvalidHeader(object):
assert returned == 1.0
-class TestCreateAcceptLanguageHeader(object):
+class TestCreateAcceptLanguageHeader:
def test_header_value_is_None(self):
header_value = None
returned = create_accept_language_header(header_value=header_value)
@@ -5663,7 +5664,7 @@ class TestCreateAcceptLanguageHeader(object):
assert returned2._header_value == returned._header_value
-class TestAcceptLanguageProperty(object):
+class TestAcceptLanguageProperty:
def test_fget_header_is_None(self):
request = Request.blank("/", environ={"HTTP_ACCEPT_LANGUAGE": None})
property_ = accept_language_property()
@@ -5723,7 +5724,7 @@ class TestAcceptLanguageProperty(object):
request = Request.blank("/", environ={"HTTP_ACCEPT_LANGUAGE": ""})
property_ = accept_language_property()
- class Other(object):
+ class Other:
def __str__(self):
return "en-gb;q=0.5, fr;q=0, es"
diff --git a/tests/test_byterange.py b/tests/test_byterange.py
index a3222c6..a10a9f6 100644
--- a/tests/test_byterange.py
+++ b/tests/test_byterange.py
@@ -1,9 +1,8 @@
+from collections.abc import Iterable
+
import pytest
-from webob.byterange import Range
-from webob.byterange import ContentRange
-from webob.byterange import _is_content_range_valid
-from webob.compat import Iterable
+from webob.byterange import ContentRange, Range, _is_content_range_valid
# Range class
diff --git a/tests/test_cachecontrol.py b/tests/test_cachecontrol.py
index dd7c283..5162858 100644
--- a/tests/test_cachecontrol.py
+++ b/tests/test_cachecontrol.py
@@ -9,7 +9,7 @@ def test_cache_control_object_max_age_None():
assert cc.max_age == -1
-class TestUpdateDict(object):
+class TestUpdateDict:
def setup_method(self, method):
self.call_queue = []
@@ -73,7 +73,7 @@ class TestUpdateDict(object):
assert self.call_queue[-1] == "Called with: {}", self.call_queue[-1]
-class TestExistProp(object):
+class TestExistProp:
"""
Test webob.cachecontrol.exists_property
"""
@@ -84,7 +84,7 @@ class TestExistProp(object):
def make_one(self):
from webob.cachecontrol import exists_property
- class Dummy(object):
+ class Dummy:
properties = dict(prop=1)
type = "dummy"
prop = exists_property("prop", "dummy")
@@ -119,7 +119,7 @@ class TestExistProp(object):
assert "prop" not in obj.properties
-class TestValueProp(object):
+class TestValueProp:
"""
Test webob.cachecontrol.exists_property
"""
@@ -130,7 +130,7 @@ class TestValueProp(object):
def make_one(self):
from webob.cachecontrol import value_property
- class Dummy(object):
+ class Dummy:
properties = dict(prop=1)
type = "dummy"
prop = value_property("prop", "dummy")
@@ -163,7 +163,7 @@ class TestValueProp(object):
def test_set_wrong_type(self):
from webob.cachecontrol import value_property
- class Dummy(object):
+ class Dummy:
properties = dict(prop=1, type="fail")
type = "dummy"
prop = value_property("prop", "dummy", type="failingtype")
@@ -211,34 +211,34 @@ def test_serialize_cache_control_emptydict():
def test_serialize_cache_control_cache_control_object():
- from webob.cachecontrol import serialize_cache_control, CacheControl
+ from webob.cachecontrol import CacheControl, serialize_cache_control
result = serialize_cache_control(CacheControl({}, "request"))
assert result == ""
def test_serialize_cache_control_object_with_headers():
- from webob.cachecontrol import serialize_cache_control, CacheControl
+ from webob.cachecontrol import CacheControl, serialize_cache_control
result = serialize_cache_control(CacheControl({"header": "a"}, "request"))
assert result == "header=a"
def test_serialize_cache_control_value_is_None():
- from webob.cachecontrol import serialize_cache_control, CacheControl
+ from webob.cachecontrol import CacheControl, serialize_cache_control
result = serialize_cache_control(CacheControl({"header": None}, "request"))
assert result == "header"
def test_serialize_cache_control_value_needs_quote():
- from webob.cachecontrol import serialize_cache_control, CacheControl
+ from webob.cachecontrol import CacheControl, serialize_cache_control
result = serialize_cache_control(CacheControl({"header": '""'}, "request"))
assert result == 'header=""""'
-class TestCacheControl(object):
+class TestCacheControl:
def make_one(self, props, typ):
from webob.cachecontrol import CacheControl
diff --git a/tests/test_client.py b/tests/test_client.py
index 4dac6bb..de9635e 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,6 +1,6 @@
-import unittest
import io
import socket
+import unittest
class TestSendRequest(unittest.TestCase):
@@ -220,13 +220,13 @@ class TestSendRequest(unittest.TestCase):
self.assertEqual(response.length, None)
-class DummyMessage(object):
+class DummyMessage:
def __init__(self, msg):
self.msg = msg
self.headers = self._headers = {}
-class DummyResponse(object):
+class DummyResponse:
def __init__(self, msg, headerval="10"):
self.msg = DummyMessage(msg)
self.status = "200"
@@ -241,7 +241,7 @@ class DummyResponse(object):
return b"foo"
-class DummyConnectionFactory(object):
+class DummyConnectionFactory:
def __init__(self, result=None):
self.result = result
self.closed = False
@@ -261,7 +261,7 @@ class DummyConnectionFactory(object):
self.closed = True
-class DummyRequestFactory(object):
+class DummyRequestFactory:
def __init__(self, hostport, **kw):
self.hostport = hostport
self.kw = kw
diff --git a/tests/test_client_functional.py b/tests/test_client_functional.py
index d84ef86..32f884e 100644
--- a/tests/test_client_functional.py
+++ b/tests/test_client_functional.py
@@ -3,8 +3,8 @@ import time
import pytest
from webob import Request, Response
-from webob.dec import wsgify
from webob.client import SendRequest
+from webob.dec import wsgify
@wsgify
diff --git a/tests/test_compat.py b/tests/test_compat.py
index c381fd8..44554e5 100644
--- a/tests/test_compat.py
+++ b/tests/test_compat.py
@@ -1,40 +1,38 @@
-import unittest
from io import BytesIO
-
-import pytest
import sys
+import unittest
-from webob.compat import text_type
+import pytest
class text_Tests(unittest.TestCase):
def _callFUT(self, *arg, **kw):
- from webob.compat import text_
+ from webob.util import text_
return text_(*arg, **kw)
def test_binary(self):
result = self._callFUT(b"123")
- self.assertTrue(isinstance(result, text_type))
- self.assertEqual(result, text_type(b"123", "ascii"))
+ self.assertTrue(isinstance(result, str))
+ self.assertEqual(result, str(b"123", "ascii"))
def test_binary_alternate_decoding(self):
result = self._callFUT(b"La Pe\xc3\xb1a", "utf-8")
- self.assertTrue(isinstance(result, text_type))
- self.assertEqual(result, text_type(b"La Pe\xc3\xb1a", "utf-8"))
+ self.assertTrue(isinstance(result, str))
+ self.assertEqual(result, str(b"La Pe\xc3\xb1a", "utf-8"))
def test_binary_decoding_error(self):
self.assertRaises(UnicodeDecodeError, self._callFUT, b"\xff", "utf-8")
def test_text(self):
- result = self._callFUT(text_type(b"123", "ascii"))
- self.assertTrue(isinstance(result, text_type))
- self.assertEqual(result, text_type(b"123", "ascii"))
+ result = self._callFUT(str(b"123", "ascii"))
+ self.assertTrue(isinstance(result, str))
+ self.assertEqual(result, str(b"123", "ascii"))
class bytes_Tests(unittest.TestCase):
def _callFUT(self, *arg, **kw):
- from webob.compat import bytes_
+ from webob.util import bytes_
return bytes_(*arg, **kw)
@@ -44,19 +42,19 @@ class bytes_Tests(unittest.TestCase):
self.assertEqual(result, b"123")
def test_text(self):
- val = text_type(b"123", "ascii")
+ val = str(b"123", "ascii")
result = self._callFUT(val)
self.assertTrue(isinstance(result, bytes))
self.assertEqual(result, b"123")
def test_text_alternate_encoding(self):
- val = text_type(b"La Pe\xc3\xb1a", "utf-8")
+ val = str(b"La Pe\xc3\xb1a", "utf-8")
result = self._callFUT(val, "utf-8")
self.assertTrue(isinstance(result, bytes))
self.assertEqual(result, b"La Pe\xc3\xb1a")
-class Test_cgi_FieldStorage_Py3_tests(object):
+class Test_cgi_FieldStorage_Py3_tests:
def test_fieldstorage_not_multipart(self):
from webob.compat import cgi_FieldStorage
diff --git a/tests/test_cookies.py b/tests/test_cookies.py
index 578bfb8..f8e9e36 100644
--- a/tests/test_cookies.py
+++ b/tests/test_cookies.py
@@ -1,13 +1,9 @@
-import pytest
-
from datetime import timedelta
-from webob import cookies
-from webob.compat import text_
-from webob.compat import native_
+import pytest
-py2only = pytest.mark.skipif("sys.version_info >= (3, 0)")
-py3only = pytest.mark.skipif("sys.version_info < (3, 0)")
+from webob import cookies
+from webob.util import text_
def setup_module(module):
@@ -53,7 +49,7 @@ def test_cookie_complex():
def d(v):
return v.decode("ascii")
- c_dict = dict((d(k), d(v.value)) for k, v in c.items())
+ c_dict = {d(k): d(v.value) for k, v in c.items()}
assert c_dict == {
"a": "42,",
"CP": "null*",
@@ -273,7 +269,7 @@ def test_morsel_repr():
assert result == "<Morsel: a='b'>"
-class TestRequestCookies(object):
+class TestRequestCookies:
def _makeOne(self, environ):
from webob.cookies import RequestCookies
@@ -324,7 +320,7 @@ class TestRequestCookies(object):
inst.__setitem__(None, 1)
def test___setitem__name_not_encodeable_to_ascii(self):
- name = native_(b"La Pe\xc3\xb1a", "utf-8")
+ name = str(b"La Pe\xc3\xb1a", "utf-8")
inst = self._makeOne({})
with pytest.raises(TypeError):
inst.__setitem__(name, "abc")
@@ -348,14 +344,14 @@ class TestRequestCookies(object):
inst.__setitem__("a", value)
def test__setitem__success_no_existing_headers(self):
- value = native_(b"test_cookie", "utf-8")
+ value = str(b"test_cookie", "utf-8")
environ = {}
inst = self._makeOne(environ)
inst["a"] = value
assert environ["HTTP_COOKIE"] == "a=test_cookie"
def test__setitem__success_append(self):
- value = native_(b"test_cookie", "utf-8")
+ value = str(b"test_cookie", "utf-8")
environ = {"HTTP_COOKIE": "a=1; b=2"}
inst = self._makeOne(environ)
inst["c"] = value
@@ -407,40 +403,17 @@ class TestRequestCookies(object):
inst = self._makeOne(environ)
assert sorted(list(inst.items())) == [("a", "1"), ("b", val), ("c", "3")]
- @py2only
- def test_iterkeys(self):
- environ = {"HTTP_COOKIE": 'a=1; b="La Pe\\303\\261a"; c=3'}
- inst = self._makeOne(environ)
- assert sorted(list(inst.iterkeys())) == ["a", "b", "c"]
-
- @py3only
def test_iterkeys_py3(self):
environ = {"HTTP_COOKIE": b'a=1; b="La Pe\\303\\261a"; c=3'.decode("utf-8")}
inst = self._makeOne(environ)
assert sorted(list(inst.keys())) == ["a", "b", "c"]
- @py2only
- def test_itervalues(self):
- val = text_(b"La Pe\xc3\xb1a", "utf-8")
- environ = {"HTTP_COOKIE": 'a=1; b="La Pe\\303\\261a"; c=3'}
- inst = self._makeOne(environ)
- sorted(list(inst.itervalues())) == ["1", "3", val]
-
- @py3only
def test_itervalues_py3(self):
val = text_(b"La Pe\xc3\xb1a", "utf-8")
environ = {"HTTP_COOKIE": b'a=1; b="La Pe\\303\\261a"; c=3'.decode("utf-8")}
inst = self._makeOne(environ)
sorted(list(inst.values())) == ["1", "3", val]
- @py2only
- def test_iteritems(self):
- val = text_(b"La Pe\xc3\xb1a", "utf-8")
- environ = {"HTTP_COOKIE": 'a=1; b="La Pe\\303\\261a"; c=3'}
- inst = self._makeOne(environ)
- assert sorted(list(inst.iteritems())) == [("a", "1"), ("b", val), ("c", "3")]
-
- @py3only
def test_iteritems_py3(self):
val = text_(b"La Pe\xc3\xb1a", "utf-8")
environ = {"HTTP_COOKIE": b'a=1; b="La Pe\\303\\261a"; c=3'.decode("utf-8")}
@@ -480,7 +453,7 @@ class TestRequestCookies(object):
assert r.endswith(">")
-class TestCookieMakeCookie(object):
+class TestCookieMakeCookie:
def makeOne(self, name, value, **kw):
from webob.cookies import make_cookie
@@ -535,9 +508,9 @@ class TestCookieMakeCookie(object):
assert "SameSite=" + samesite in cookie
-class CommonCookieProfile(object):
+class CommonCookieProfile:
def makeDummyRequest(self, **kw):
- class Dummy(object):
+ class Dummy:
def __init__(self, **kwargs):
self.__dict__.update(**kwargs)
@@ -591,7 +564,7 @@ class TestCookieProfile(CommonCookieProfile):
cookie.get_value()
def test_get_value_serializer_raises_value_error(self):
- class RaisingSerializer(object):
+ class RaisingSerializer:
def loads(self, val):
raise ValueError("foo")
@@ -783,11 +756,12 @@ class TestSignedCookieProfile(CommonCookieProfile):
def serialize(secret, salt, data):
- import hmac
import base64
- import json
from hashlib import sha1
- from webob.compat import bytes_
+ import hmac
+ import json
+
+ from webob.util import bytes_
salted_secret = bytes_(salt or "", "utf-8") + bytes_(secret, "utf-8")
cstruct = bytes_(json.dumps(data))
@@ -795,7 +769,7 @@ def serialize(secret, salt, data):
return base64.urlsafe_b64encode(sig + cstruct).rstrip(b"=")
-class TestSignedSerializer(object):
+class TestSignedSerializer:
def makeOne(self, secret, salt, hashalg="sha1", **kw):
from webob.cookies import SignedSerializer
diff --git a/tests/test_datetime_utils.py b/tests/test_datetime_utils.py
index 466493d..56ffff2 100644
--- a/tests/test_datetime_utils.py
+++ b/tests/test_datetime_utils.py
@@ -1,8 +1,9 @@
-import pytest
-
-import datetime
import calendar
+import datetime
from email.utils import formatdate
+
+import pytest
+
from webob import datetime_utils
@@ -15,52 +16,38 @@ def test_UTC():
assert repr(x) == "UTC"
-def test_parse_date():
- """Testing datetime_utils.parse_date.
- We need to verify the following scenarios:
- * a nil submitted value
- * a submitted value that cannot be parse into a date
- * a valid RFC2822 date with and without timezone
- """
-
- ret = datetime_utils.parse_date(None)
- assert ret is None, (
- "We passed a None value to parse_date. We should get"
- " None but instead we got %s" % ret
- )
-
- ret = datetime_utils.parse_date("Hi There")
- assert ret is None, (
- "We passed an invalid value to parse_date. We should"
- " get None but instead we got %s" % ret
- )
-
- ret = datetime_utils.parse_date(1)
- assert ret is None, (
- "We passed an invalid value to parse_date. We should"
- " get None but instead we got %s" % ret
- )
+# Testing datetime_utils.parse_date.
+# We need to verify the following scenarios:
+# * a nil submitted value
+# * a submitted value that cannot be parse into a date
+# * a valid RFC2822 date with and without timezone
- ret = datetime_utils.parse_date("\xc3")
- assert ret is None, (
- "We passed an invalid value to parse_date. We should"
- " get None but instead we got %s" % ret
- )
- ret = datetime_utils.parse_date("Mon, 20 Nov 1995 19:12:08 -0500")
- assert ret == datetime.datetime(1995, 11, 21, 0, 12, 8, tzinfo=datetime_utils.UTC)
-
- ret = datetime_utils.parse_date("Mon, 20 Nov 1995 19:12:08")
+class Uncooperative:
+ def __str__(self):
+ raise NotImplementedError
- assert ret == datetime.datetime(1995, 11, 20, 19, 12, 8, tzinfo=datetime_utils.UTC)
- ret = datetime_utils.parse_date(Uncooperative())
- assert ret is None
+@pytest.mark.parametrize("invalid_date", [None, "Hi there", 1, "\xc3", Uncooperative()])
+def test_parse_date_invalid(invalid_date):
+ assert datetime_utils.parse_date(invalid_date) is None
-class Uncooperative(object):
- def __str__(self):
- raise NotImplementedError
+@pytest.mark.parametrize(
+ "valid_date, parsed_datetime",
+ [
+ (
+ "Mon, 20 Nov 1995 19:12:08 -0500",
+ datetime.datetime(1995, 11, 21, 0, 12, 8, tzinfo=datetime_utils.UTC),
+ ),
+ (
+ "Mon, 20 Nov 1995 19:12:08",
+ datetime.datetime(1995, 11, 20, 19, 12, 8, tzinfo=datetime_utils.UTC),
+ ),
+ ],
+)
+def test_parse_date_valid(valid_date, parsed_datetime):
+ assert datetime_utils.parse_date(valid_date) == parsed_datetime
def test_serialize_date():
@@ -71,7 +58,7 @@ def test_serialize_date():
* passing a timedelta, return now plus the delta
* passing an invalid object, should raise ValueError
"""
- from webob.compat import text_
+ from webob.util import text_
ret = datetime_utils.serialize_date("Mon, 20 Nov 1995 19:12:08 GMT")
assert isinstance(ret, str)
@@ -125,7 +112,7 @@ def test_timedelta_to_seconds():
assert result == 7464960000
-class _NowRestorer(object):
+class _NowRestorer:
def __init__(self, new_now):
self._new_now = new_now
self._old_now = None
diff --git a/tests/test_dec.py b/tests/test_dec.py
index 3c4c186..1c0200f 100644
--- a/tests/test_dec.py
+++ b/tests/test_dec.py
@@ -1,9 +1,9 @@
import unittest
+
+from webob.dec import wsgify
from webob.request import Request
from webob.response import Response
-from webob.dec import wsgify
-from webob.compat import bytes_
-from webob.compat import text_
+from webob.util import bytes_, text_
class DecoratorTests(unittest.TestCase):
@@ -11,6 +11,7 @@ class DecoratorTests(unittest.TestCase):
if isinstance(req, str):
req = Request.blank(req)
resp = req.get_response(app)
+
return resp
def test_wsgify(self):
@@ -70,7 +71,7 @@ class DecoratorTests(unittest.TestCase):
def test_wsgify_no___get__(self):
# use a class instance instead of a fn so we wrap something w/
# no __get__
- class TestApp(object):
+ class TestApp:
def __call__(self, req):
return "nothing to see here"
@@ -162,6 +163,7 @@ class DecoratorTests(unittest.TestCase):
@wsgify
def test_app(req):
self.assertEqual(req.method, "PUT")
+
return Response(req.body)
resp = test_app.request("/url/path", method="PUT", body=resp_str)
@@ -198,6 +200,7 @@ class DecoratorTests(unittest.TestCase):
@wsgify.middleware
def set_urlvar(req, app, **vars):
req.urlvars.update(vars)
+
return app(req)
from webob.dec import _MiddlewareFactory
@@ -223,6 +226,7 @@ class DecoratorTests(unittest.TestCase):
@wsgify.middleware
def set_args(req, app, **kwargs):
req.urlvars = kwargs
+
return req.get_response(app)
@wsgify
@@ -240,6 +244,7 @@ class DecoratorTests(unittest.TestCase):
@wsgify.middleware
def set_args(req, app, **kwargs):
req.urlvars = kwargs
+
return req.get_response(app)
@wsgify
@@ -257,6 +262,7 @@ class DecoratorTests(unittest.TestCase):
@wsgify.middleware
def set_urlvar(req, app, **vars):
req.urlvars.update(vars)
+
return app(req)
@set_urlvar(a=1, b=2)
diff --git a/tests/test_descriptors.py b/tests/test_descriptors.py
index 9d3eecc..a214094 100644
--- a/tests/test_descriptors.py
+++ b/tests/test_descriptors.py
@@ -1,11 +1,9 @@
-from datetime import tzinfo
-from datetime import timedelta
+from datetime import timedelta, tzinfo
import pytest
-from webob.compat import native_, text_
-
from webob.request import Request
+from webob.util import text_
class GMT(tzinfo):
@@ -145,8 +143,8 @@ def test_header_getter_doc():
def test_header_getter_fget():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -154,8 +152,8 @@ def test_header_getter_fget():
def test_header_getter_fset():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -164,8 +162,8 @@ def test_header_getter_fset():
def test_header_getter_fset_none():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -175,9 +173,8 @@ def test_header_getter_fset_none():
def test_header_getter_fset_text():
- from webob.compat import text_
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -186,9 +183,8 @@ def test_header_getter_fset_text():
def test_header_getter_fset_text_control_chars():
- from webob.compat import text_
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -196,9 +192,19 @@ def test_header_getter_fset_text_control_chars():
desc.fset(resp, text_("\n"))
-def test_header_getter_fdel():
+def test_header_getter_fset_non_text():
+ from webob import Response
from webob.descriptors import header_getter
+
+ resp = Response("aresp")
+ desc = header_getter("AHEADER", "14.3")
+ with pytest.raises(ValueError):
+ desc.fset(resp, b"test")
+
+
+def test_header_getter_fdel():
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -208,8 +214,8 @@ def test_header_getter_fdel():
def test_header_getter_unicode_fget_none():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -217,8 +223,8 @@ def test_header_getter_unicode_fget_none():
def test_header_getter_unicode_fget():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -227,8 +233,8 @@ def test_header_getter_unicode_fget():
def test_header_getter_unicode_fset_none():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -237,8 +243,8 @@ def test_header_getter_unicode_fset_none():
def test_header_getter_unicode_fset():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -247,8 +253,8 @@ def test_header_getter_unicode_fset():
def test_header_getter_unicode_fdel():
- from webob.descriptors import header_getter
from webob import Response
+ from webob.descriptors import header_getter
resp = Response("aresp")
desc = header_getter("AHEADER", "14.3")
@@ -258,9 +264,7 @@ def test_header_getter_unicode_fdel():
def test_converter_not_prop():
- from webob.descriptors import converter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import converter, parse_int_safe, serialize_int
with pytest.raises(AssertionError):
converter(
@@ -269,10 +273,12 @@ def test_converter_not_prop():
def test_converter_with_name_docstring():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
desc = converter(
environ_getter("CONTENT_LENGTH", "666", "14.13"),
@@ -289,10 +295,12 @@ def test_converter_with_name_docstring():
def test_converter_with_name_fget():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -305,10 +313,12 @@ def test_converter_with_name_fget():
def test_converter_with_name_fset():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -322,10 +332,12 @@ def test_converter_with_name_fset():
def test_converter_without_name_fget():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -335,10 +347,12 @@ def test_converter_without_name_fget():
def test_converter_without_name_fset():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -349,10 +363,12 @@ def test_converter_without_name_fset():
def test_converter_none_for_wrong_type():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -365,10 +381,12 @@ def test_converter_none_for_wrong_type():
def test_converter_delete():
- from webob.descriptors import converter
- from webob.descriptors import environ_getter
- from webob.descriptors import parse_int_safe
- from webob.descriptors import serialize_int
+ from webob.descriptors import (
+ converter,
+ environ_getter,
+ parse_int_safe,
+ serialize_int,
+ )
req = Request.blank("/")
desc = converter(
@@ -446,8 +464,8 @@ def test_serialize_list_unicode():
def test_converter_date():
import datetime
- from webob.descriptors import converter_date
- from webob.descriptors import environ_getter
+
+ from webob.descriptors import converter_date, environ_getter
req = Request.blank("/")
UTC = GMT()
@@ -458,8 +476,7 @@ def test_converter_date():
def test_converter_date_docstring():
- from webob.descriptors import converter_date
- from webob.descriptors import environ_getter
+ from webob.descriptors import converter_date, environ_getter
desc = converter_date(
environ_getter("HTTP_DATE", "Tue, 15 Nov 1994 08:12:31 GMT", "14.8")
@@ -481,6 +498,7 @@ def test_date_header_fget_none():
def test_date_header_fset_fget():
import datetime
+
from webob import Response
from webob.descriptors import date_header
@@ -505,7 +523,7 @@ def test_date_header_fdel():
def test_deprecated_property():
from webob.descriptors import deprecated_property
- class Foo(object):
+ class Foo:
pass
Foo.attr = deprecated_property("attr", "attr", "whatever", "1.2")
@@ -562,6 +580,7 @@ def test_serialize_if_range_unicode():
def test_serialize_if_range_datetime():
import datetime
+
from webob.descriptors import serialize_if_range
UTC = GMT()
@@ -926,7 +945,7 @@ def test_serialize_auth_digest_tuple():
_nodefault = object()
-class _TestEnvironDecoder(object):
+class _TestEnvironDecoder:
def _callFUT(self, key, default=_nodefault, rfc_section=None, encattr=None):
from webob.descriptors import environ_decoder
@@ -995,7 +1014,7 @@ class _TestEnvironDecoder(object):
desc = self._callFUT("HTTP_X_AKEY", encattr="url_encoding")
req = self._makeRequest()
desc.fset(req, text_(b"\xc3\xab", "utf-8"))
- assert req.environ["HTTP_X_AKEY"] == native_(b"\xc3\xab", "latin-1")
+ assert req.environ["HTTP_X_AKEY"] == str(b"\xc3\xab", "latin-1")
class TestEnvironDecoder(_TestEnvironDecoder):
@@ -1003,31 +1022,12 @@ class TestEnvironDecoder(_TestEnvironDecoder):
from webob.request import BaseRequest
req = BaseRequest.blank("/")
- return req
- def test_fget_nonascii(self):
- desc = self._callFUT("HTTP_X_AKEY", encattr="url_encoding")
- req = self._makeRequest()
- req.environ["HTTP_X_AKEY"] = native_(b"\xc3\xab")
- result = desc.fget(req)
- assert result == text_(b"\xc3\xab", "utf-8")
-
-
-class TestEnvironDecoderLegacy(_TestEnvironDecoder):
- def _makeRequest(self):
- from webob.request import LegacyRequest
-
- req = LegacyRequest.blank("/")
return req
def test_fget_nonascii(self):
desc = self._callFUT("HTTP_X_AKEY", encattr="url_encoding")
req = self._makeRequest()
- req.environ["HTTP_X_AKEY"] = native_(b"\xc3\xab", "latin-1")
+ req.environ["HTTP_X_AKEY"] = str(b"\xc3\xab", "latin-1")
result = desc.fget(req)
- assert result == native_(b"\xc3\xab", "latin-1")
-
- def test_default_fget_nonascii(self):
- req = self._makeRequest()
- desc = self._callFUT("akey", default=b"the_default")
- assert desc.fget(req).__class__ == bytes
+ assert result == text_(b"\xc3\xab", "utf-8")
diff --git a/tests/test_etag.py b/tests/test_etag.py
index f256636..279fc08 100644
--- a/tests/test_etag.py
+++ b/tests/test_etag.py
@@ -1,7 +1,7 @@
from webob.etag import ETagMatcher, IfRange, etag_property
-class Test_etag_properties(object):
+class Test_etag_properties:
def _makeDummyRequest(self, **kw):
"""
Return a DummyRequest object with attrs from kwargs.
@@ -9,7 +9,7 @@ class Test_etag_properties(object):
Then you can: uid = dr.environment.get('userid', 'SomeDefault')
"""
- class Dummy(object):
+ class Dummy:
def __init__(self, **kwargs):
self.__dict__.update(**kwargs)
@@ -59,7 +59,7 @@ class Test_etag_properties(object):
assert req.environ["QUAY"] == "VALYOU"
-class Test_AnyETag(object):
+class Test_AnyETag:
def _getTargetClass(self):
from webob.etag import _AnyETag
@@ -85,7 +85,7 @@ class Test_AnyETag(object):
assert str(etag) == "*"
-class Test_NoETag(object):
+class Test_NoETag:
def _getTargetClass(self):
from webob.etag import _NoETag
@@ -111,7 +111,7 @@ class Test_NoETag(object):
assert str(etag) == ""
-class Test_Parse(object):
+class Test_Parse:
def test_parse_None(self):
et = ETagMatcher.parse(None)
assert et.etags == []
@@ -158,7 +158,7 @@ class Test_Parse(object):
assert et.etags, ["ONE" == "TWO"]
-class Test_IfRange(object):
+class Test_IfRange:
def test___repr__(self):
assert repr(IfRange(None)) == "IfRange(None)"
diff --git a/tests/test_etag_nose.py b/tests/test_etag_nose.py
index 8a8cd5d..dd0f0cc 100644
--- a/tests/test_etag_nose.py
+++ b/tests/test_etag_nose.py
@@ -1,5 +1,5 @@
-from webob.etag import IfRange, ETagMatcher
from webob import Response
+from webob.etag import ETagMatcher, IfRange
def test_if_range_None():
diff --git a/tests/test_exc.py b/tests/test_exc.py
index 35454c9..c1fa721 100644
--- a/tests/test_exc.py
+++ b/tests/test_exc.py
@@ -1,16 +1,17 @@
-import pytest
-
import json
-from webob.request import Request
-from webob.dec import wsgify
+import pytest
+
from webob import exc as webob_exc
+from webob.dec import wsgify
+from webob.request import Request
@wsgify
def method_not_allowed_app(req):
if req.method != "GET":
raise webob_exc.HTTPMethodNotAllowed()
+
return "hello!"
@@ -23,14 +24,18 @@ def test_noescape_not_basestring():
def test_noescape_unicode():
- class DummyUnicodeObject(object):
- def __unicode__(self):
+ class DummyUnicodeObject:
+ def __str__(self):
return "42"
duo = DummyUnicodeObject()
assert webob_exc.no_escape(duo) == "42"
+def test_noescape_bytes():
+ assert webob_exc.no_escape(b"test") == "test"
+
+
def test_strip_tags_empty():
assert webob_exc.strip_tags("") == ""
@@ -61,6 +66,7 @@ def test_HTTPException():
def _response(environ, start_response):
_called.append((environ, start_response))
+
return _result
environ = {}
@@ -471,7 +477,7 @@ def test_HTTPMove_call_query_string():
def test_HTTPFound_unused_environ_variable():
- class Crashy(object):
+ class Crashy:
def __str__(self):
raise Exception("I crashed!")
diff --git a/tests/test_in_wsgiref.py b/tests/test_in_wsgiref.py
index 72b71ea..5d3e3a9 100644
--- a/tests/test_in_wsgiref.py
+++ b/tests/test_in_wsgiref.py
@@ -1,17 +1,15 @@
-import sys
+import cgi
import logging
import socket
-import cgi
+import sys
+from urllib.request import urlopen as url_open
import pytest
+from webob.compat import Empty, Queue
from webob.request import Request
from webob.response import Response
-from webob.compat import url_open
-from webob.compat import bytes_
-from webob.compat import reraise
-from webob.compat import Queue
-from webob.compat import Empty
+from webob.util import bytes_
log = logging.getLogger(__name__)
@@ -35,6 +33,7 @@ def _test_app_req_reading(env, sr):
test_op(req)
log.debug("done")
r = Response("ok")
+
return r(env, sr)
@@ -55,9 +54,10 @@ def test_interrupted_request(serve):
res = _global_res.get(timeout=1)
except Empty:
raise AssertionError("Error during test %s", path)
+
if res is not None:
print("Error during test:", path)
- reraise(res)
+ raise res[0](res[1]).with_traceback(res[2])
_global_res = Queue()
@@ -68,6 +68,7 @@ def _test_app_req_interrupt(env, sr):
try:
req = Request(env)
cl = req.content_length
+
if cl != target_cl:
raise AssertionError(
"request.content_length is %s instead of %s" % (cl, target_cl)
@@ -81,6 +82,7 @@ def _test_app_req_interrupt(env, sr):
else:
_global_res.put(None)
sr("200 OK", [])
+
return []
@@ -92,7 +94,7 @@ def _req_int_cgi(req):
def _req_int_readline(req):
try:
assert req.body_file.readline() == b"a=b\n"
- except IOError:
+ except OSError:
# too early to detect disconnect
raise AssertionError("False disconnect alert")
req.body_file.readline()
diff --git a/tests/test_misc.py b/tests/test_misc.py
index 1b6c629..e356f33 100644
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -1,23 +1,19 @@
import pytest
-from webob.util import html_escape
-from webob.compat import text_
+from webob.util import html_escape, text_
-py2only = pytest.mark.skipif("sys.version_info >= (3, 0)")
-py3only = pytest.mark.skipif("sys.version_info < (3, 0)")
-
-class t_esc_HTML(object):
+class t_esc_HTML:
def __html__(self):
return "<div>hello</div>"
-class t_esc_Unicode(object):
- def __unicode__(self):
- return text_(b"\xe9")
+class t_esc_Unicode:
+ def __str__(self):
+ return "\xe9"
-class t_esc_UnsafeAttrs(object):
+class t_esc_UnsafeAttrs:
attr = "value"
def __getattr__(self, k):
@@ -27,12 +23,9 @@ class t_esc_UnsafeAttrs(object):
return "<UnsafeAttrs>"
-class t_esc_SuperMoose(object):
+class t_esc_SuperMoose:
def __str__(self):
- return text_(b"m\xf8ose").encode("utf-8")
-
- def __unicode__(self):
- return text_(b"m\xf8ose")
+ return "m\xf8ose"
@pytest.mark.parametrize(
@@ -43,26 +36,20 @@ class t_esc_SuperMoose(object):
("&egrave;", "&amp;egrave;"),
# The apostrophe is *not* escaped, which some might consider to be
# a serious bug (see, e.g. http://www.cvedetails.com/cve/CVE-2010-2480/)
- pytest.param("'", "'", marks=py2only),
- pytest.param("'", "&#x27;", marks=py3only),
- (text_("the majestic m\xf8ose"), "the majestic m&#248;ose"),
+ pytest.param("'", "&#x27;"),
+ ("the majestic m\xf8ose", "the majestic m&#248;ose"),
# 8-bit strings are passed through
- (text_("\xe9"), "&#233;"),
+ ("\xe9", "&#233;"),
# ``None`` is treated specially, and returns the empty string.
(None, ""),
# Objects that define a ``__html__`` method handle their own escaping
(t_esc_HTML(), "<div>hello</div>"),
# Things that are not strings are converted to strings and then escaped
(42, "42"),
- # If an object implements both ``__str__`` and ``__unicode__``, the latter
- # is preferred
(t_esc_SuperMoose(), "m&#248;ose"),
(t_esc_Unicode(), "&#233;"),
(t_esc_UnsafeAttrs(), "&lt;UnsafeAttrs&gt;"),
- pytest.param(Exception("expected a '<'."), "expected a '&lt;'.", marks=py2only),
- pytest.param(
- Exception("expected a '<'."), "expected a &#x27;&lt;&#x27;.", marks=py3only
- ),
+ pytest.param(Exception("expected a '<'."), "expected a &#x27;&lt;&#x27;."),
],
)
def test_html_escape(input, expected):
diff --git a/tests/test_multidict.py b/tests/test_multidict.py
index 5c9fa85..1d04bad 100644
--- a/tests/test_multidict.py
+++ b/tests/test_multidict.py
@@ -1,11 +1,10 @@
-# -*- coding: utf-8 -*-
-
import unittest
+
from webob import multidict
-from webob.compat import text_
+from webob.util import text_
-class BaseDictTests(object):
+class BaseDictTests:
def setUp(self):
self._list = [("a", text_("\xe9")), ("a", "e"), ("a", "f"), ("b", "1")]
self.data = multidict.MultiDict(self._list)
@@ -41,11 +40,8 @@ class BaseDictTests(object):
def test_dict_api(self):
self.assertTrue("a" in self.d.mixed())
self.assertTrue("a" in self.d.keys())
- self.assertTrue("a" in self.d.iterkeys())
self.assertTrue(("b", "1") in self.d.items())
- self.assertTrue(("b", "1") in self.d.iteritems())
self.assertTrue("1" in self.d.values())
- self.assertTrue("1" in self.d.itervalues())
self.assertEqual(len(self.d), 4)
def test_set_del_item(self):
@@ -161,8 +157,9 @@ class BaseDictTests(object):
def test_from_fieldstorage_with_charset(self):
from cgi import FieldStorage
- from webob.request import BaseRequest
+
from webob.multidict import MultiDict
+ from webob.request import BaseRequest
multipart_type = "multipart/form-data; boundary=foobar"
from io import BytesIO
@@ -189,8 +186,9 @@ class BaseDictTests(object):
def test_from_fieldstorage_with_base64_encoding(self):
from cgi import FieldStorage
- from webob.request import BaseRequest
+
from webob.multidict import MultiDict
+ from webob.request import BaseRequest
multipart_type = "multipart/form-data; boundary=foobar"
from io import BytesIO
@@ -218,8 +216,9 @@ class BaseDictTests(object):
def test_from_fieldstorage_with_quoted_printable_encoding(self):
from cgi import FieldStorage
- from webob.request import BaseRequest
+
from webob.multidict import MultiDict
+ from webob.request import BaseRequest
multipart_type = "multipart/form-data; boundary=foobar"
from io import BytesIO
@@ -502,12 +501,8 @@ class NoVarsTestCase(unittest.TestCase):
d = self._get_instance()
self.assertEqual(list(d.keys()), [])
- def test_iterkeys(self):
- d = self._get_instance()
- self.assertEqual(list(d.iterkeys()), [])
-
-class DummyField(object):
+class DummyField:
def __init__(self, name, value, filename=None):
self.name = name
self.value = value
@@ -516,6 +511,6 @@ class DummyField(object):
self.headers = {}
-class DummyFieldStorage(object):
+class DummyFieldStorage:
def __init__(self, name, value, filename=None):
self.list = [DummyField(name, value, filename)]
diff --git a/tests/test_request.py b/tests/test_request.py
index 9956824..294a511 100644
--- a/tests/test_request.py
+++ b/tests/test_request.py
@@ -1,10 +1,8 @@
-# -*- coding: utf-8 -*-
-
+from collections.abc import MutableMapping
+from io import BytesIO, StringIO
import sys
import warnings
-from io import BytesIO, StringIO
-
import pytest
from webob.acceptparse import (
@@ -21,15 +19,11 @@ from webob.acceptparse import (
AcceptNoHeader,
AcceptValidHeader,
)
-from webob.compat import MutableMapping, bytes_, native_, text_type, text_
from webob.multidict import NoVars
+from webob.util import bytes_, text_
-py2only = pytest.mark.skipif("sys.version_info >= (3, 0)")
-py3only = pytest.mark.skipif("sys.version_info < (3, 0)")
-
-
-class TestRequestCommon(object):
+class TestRequestCommon:
# unit tests of non-bytes-vs-text-specific methods of request object
def _getTargetClass(self):
from webob.request import Request
@@ -530,8 +524,8 @@ class TestRequestCommon(object):
@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
def test_POST_json_no_content_type(self, method):
- from webob.request import NoVars
from webob.multidict import MultiDict
+ from webob.request import NoVars
data = b'{"password": "last centurion", "email": "rory@wiggy.net"}'
wsgi_input = BytesIO(data)
@@ -546,6 +540,7 @@ class TestRequestCommon(object):
r_2 = req.POST
r_3 = req.body
assert r_1 == b'{"password": "last centurion", "email": "rory@wiggy.net"}'
+
if method == "POST":
assert isinstance(r_2, MultiDict)
else:
@@ -679,9 +674,10 @@ class TestRequestCommon(object):
assert bool(req.accept_encoding) is False
def test_remove_conditional_headers_if_modified_since(self):
- from webob.datetime_utils import UTC
from datetime import datetime
+ from webob.datetime_utils import UTC
+
req = self._blankOne("/")
req.if_modified_since = datetime(2006, 1, 1, 12, 0, tzinfo=UTC)
req.remove_conditional_headers()
@@ -930,7 +926,7 @@ class TestRequestCommon(object):
def application(environ, start_response):
write = start_response("200 OK", [("content-type", "text/plain")])
- class AppIter(object):
+ class AppIter:
def __iter__(self):
yield "...\n"
@@ -1065,8 +1061,9 @@ class TestRequestCommon(object):
def test_blank__post_files(self):
import cgi
- from webob.request import _get_multipart_boundary
+
from webob.multidict import MultiDict
+ from webob.request import _get_multipart_boundary
POST = MultiDict()
POST["first"] = ("filename1", BytesIO(b"1"))
@@ -1182,7 +1179,7 @@ class TestRequestCommon(object):
req.copy_body()
-class TestBaseRequest(object):
+class TestBaseRequest:
# tests of methods of a base request which are encoding-specific
def _getTargetClass(self):
from webob.request import BaseRequest
@@ -1267,20 +1264,20 @@ class TestBaseRequest(object):
def test_uscript_name(self):
environ = {"SCRIPT_NAME": "/script"}
req = self._makeOne(environ)
- assert isinstance(req.uscript_name, text_type)
+ assert isinstance(req.uscript_name, str)
assert req.uscript_name == "/script"
def test_upath_info(self):
environ = {"PATH_INFO": "/path/info"}
req = self._makeOne(environ)
- assert isinstance(req.upath_info, text_type)
+ assert isinstance(req.upath_info, str)
assert req.upath_info == "/path/info"
def test_upath_info_set_unicode(self):
environ = {"PATH_INFO": "/path/info"}
req = self._makeOne(environ)
req.upath_info = text_("/another")
- assert isinstance(req.upath_info, text_type)
+ assert isinstance(req.upath_info, str)
assert req.upath_info == "/another"
def test_content_type_getter_no_parameters(self):
@@ -1770,18 +1767,18 @@ class TestBaseRequest(object):
assert inst.encget("a", None) is None
def test_encget_with_encattr(self):
- val = native_(b"\xc3\xab", "latin-1")
+ val = str(b"\xc3\xab", "latin-1")
inst = self._makeOne({"a": val})
assert inst.encget("a", encattr="url_encoding") == text_(b"\xc3\xab", "utf-8")
def test_encget_with_encattr_latin_1(self):
- val = native_(b"\xc3\xab", "latin-1")
+ val = str(b"\xc3\xab", "latin-1")
inst = self._makeOne({"a": val})
inst.my_encoding = "latin-1"
assert inst.encget("a", encattr="my_encoding") == text_(b"\xc3\xab", "latin-1")
def test_encget_no_encattr(self):
- val = native_(b"\xc3\xab", "latin-1")
+ val = str(b"\xc3\xab", "latin-1")
inst = self._makeOne({"a": val})
assert inst.encget("a") == val
@@ -1792,7 +1789,7 @@ class TestBaseRequest(object):
assert result == "http://localhost/%C3%AB/a"
def test_header_getter(self):
- val = native_(b"abc", "latin-1")
+ val = str(b"abc", "latin-1")
inst = self._makeOne({"HTTP_FLUB": val})
result = inst.headers["Flub"]
assert result.__class__ == str
@@ -1818,698 +1815,7 @@ class TestBaseRequest(object):
assert result == "example.com:80"
-class TestLegacyRequest(object):
- # tests of methods of a bytesrequest which deal with http environment vars
- def _getTargetClass(self):
- from webob.request import LegacyRequest
-
- return LegacyRequest
-
- def _makeOne(self, *arg, **kw):
- cls = self._getTargetClass()
-
- return cls(*arg, **kw)
-
- def _blankOne(self, *arg, **kw):
- cls = self._getTargetClass()
-
- return cls.blank(*arg, **kw)
-
- def test_method(self):
- environ = {"REQUEST_METHOD": "OPTIONS"}
- req = self._makeOne(environ)
- assert req.method == "OPTIONS"
-
- def test_http_version(self):
- environ = {"SERVER_PROTOCOL": "1.1"}
- req = self._makeOne(environ)
- assert req.http_version == "1.1"
-
- def test_script_name(self):
- environ = {"SCRIPT_NAME": "/script"}
- req = self._makeOne(environ)
- assert req.script_name == "/script"
-
- def test_path_info(self):
- environ = {"PATH_INFO": "/path/info"}
- req = self._makeOne(environ)
- assert req.path_info == "/path/info"
-
- def test_content_length_getter(self):
- environ = {"CONTENT_LENGTH": "1234"}
- req = self._makeOne(environ)
- assert req.content_length == 1234
-
- def test_content_length_setter_w_str(self):
- environ = {"CONTENT_LENGTH": "1234"}
- req = self._makeOne(environ)
- req.content_length = "3456"
- assert req.content_length == 3456
-
- def test_remote_user(self):
- environ = {"REMOTE_USER": "phred"}
- req = self._makeOne(environ)
- assert req.remote_user == "phred"
-
- def test_remote_addr(self):
- environ = {"REMOTE_ADDR": "1.2.3.4"}
- req = self._makeOne(environ)
- assert req.remote_addr == "1.2.3.4"
-
- def test_remote_host(self):
- environ = {"REMOTE_HOST": "example.com"}
- req = self._makeOne(environ)
- assert req.remote_host == "example.com"
-
- def test_remote_host_not_set(self):
- environ = {}
- req = self._makeOne(environ)
- assert req.remote_host is None
-
- def test_query_string(self):
- environ = {"QUERY_STRING": "foo=bar&baz=bam"}
- req = self._makeOne(environ)
- assert req.query_string == "foo=bar&baz=bam"
-
- def test_server_name(self):
- environ = {"SERVER_NAME": "somehost.tld"}
- req = self._makeOne(environ)
- assert req.server_name == "somehost.tld"
-
- def test_server_port_getter(self):
- environ = {"SERVER_PORT": "6666"}
- req = self._makeOne(environ)
- assert req.server_port == 6666
-
- def test_server_port_setter_with_string(self):
- environ = {"SERVER_PORT": "6666"}
- req = self._makeOne(environ)
- req.server_port = "6667"
- assert req.server_port == 6667
-
- def test_uscript_name(self):
- environ = {"SCRIPT_NAME": "/script"}
- req = self._makeOne(environ)
- assert isinstance(req.uscript_name, text_type)
- result = req.uscript_name
- assert result.__class__ == text_type
- assert result == "/script"
-
- def test_upath_info(self):
- environ = {"PATH_INFO": "/path/info"}
- req = self._makeOne(environ)
- result = req.upath_info
- assert isinstance(result, text_type)
- assert result == "/path/info"
-
- def test_upath_info_set_unicode(self):
- environ = {"PATH_INFO": "/path/info"}
- req = self._makeOne(environ)
- req.upath_info = text_("/another")
- result = req.upath_info
- assert isinstance(result, text_type)
- assert result == "/another"
-
- def test_content_type_getter_no_parameters(self):
- environ = {"CONTENT_TYPE": "application/xml+foobar"}
- req = self._makeOne(environ)
- assert req.content_type == "application/xml+foobar"
-
- def test_content_type_getter_w_parameters(self):
- environ = {"CONTENT_TYPE": 'application/xml+foobar;charset="utf8"'}
- req = self._makeOne(environ)
- assert req.content_type == "application/xml+foobar"
-
- def test_content_type_setter_w_None(self):
- environ = {"CONTENT_TYPE": 'application/xml+foobar;charset="utf8"'}
- req = self._makeOne(environ)
- req.content_type = None
- assert req.content_type == ""
- assert "CONTENT_TYPE" not in environ
-
- def test_content_type_setter_existing_paramter_no_new_paramter(self):
- environ = {"CONTENT_TYPE": 'application/xml+foobar;charset="utf8"'}
- req = self._makeOne(environ)
- req.content_type = "text/xml"
- assert req.content_type == "text/xml"
- assert environ["CONTENT_TYPE"] == 'text/xml;charset="utf8"'
-
- def test_content_type_deleter_clears_environ_value(self):
- environ = {"CONTENT_TYPE": 'application/xml+foobar;charset="utf8"'}
- req = self._makeOne(environ)
- del req.content_type
- assert req.content_type == ""
- assert "CONTENT_TYPE" not in environ
-
- def test_content_type_deleter_no_environ_value(self):
- environ = {}
- req = self._makeOne(environ)
- del req.content_type
- assert req.content_type == ""
- assert "CONTENT_TYPE" not in environ
-
- def test_headers_getter(self):
- CONTENT_TYPE = 'application/xml+foobar;charset="utf8"'
- environ = {"CONTENT_TYPE": CONTENT_TYPE, "CONTENT_LENGTH": "123"}
- req = self._makeOne(environ)
- headers = req.headers
- assert headers == {"Content-Type": CONTENT_TYPE, "Content-Length": "123"}
-
- def test_headers_setter(self):
- CONTENT_TYPE = 'application/xml+foobar;charset="utf8"'
- environ = {"CONTENT_TYPE": CONTENT_TYPE, "CONTENT_LENGTH": "123"}
- req = self._makeOne(environ)
- req.headers = {"Qux": "Spam"}
- assert req.headers == {"Qux": "Spam"}
- assert environ["HTTP_QUX"] == native_("Spam")
- assert environ == {"HTTP_QUX": "Spam"}
-
- def test_no_headers_deleter(self):
- CONTENT_TYPE = 'application/xml+foobar;charset="utf8"'
- environ = {"CONTENT_TYPE": CONTENT_TYPE, "CONTENT_LENGTH": "123"}
- req = self._makeOne(environ)
-
- def _test():
- del req.headers
-
- with pytest.raises(AttributeError):
- _test()
-
- def test_client_addr_xff_singleval(self):
- environ = {"HTTP_X_FORWARDED_FOR": "192.168.1.1"}
- req = self._makeOne(environ)
- assert req.client_addr == "192.168.1.1"
-
- def test_client_addr_xff_multival(self):
- environ = {"HTTP_X_FORWARDED_FOR": "192.168.1.1, 192.168.1.2"}
- req = self._makeOne(environ)
- assert req.client_addr == "192.168.1.1"
-
- def test_client_addr_prefers_xff(self):
- environ = {"REMOTE_ADDR": "192.168.1.2", "HTTP_X_FORWARDED_FOR": "192.168.1.1"}
- req = self._makeOne(environ)
- assert req.client_addr == "192.168.1.1"
-
- def test_client_addr_no_xff(self):
- environ = {"REMOTE_ADDR": "192.168.1.2"}
- req = self._makeOne(environ)
- assert req.client_addr == "192.168.1.2"
-
- def test_client_addr_no_xff_no_remote_addr(self):
- environ = {}
- req = self._makeOne(environ)
- assert req.client_addr is None
-
- def test_host_port_w_http_host_and_no_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com"}
- req = self._makeOne(environ)
- assert req.host_port == "80"
-
- def test_host_port_w_http_host_and_standard_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com:80"}
- req = self._makeOne(environ)
- assert req.host_port == "80"
-
- def test_host_port_w_http_host_and_oddball_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com:8888"}
- req = self._makeOne(environ)
- assert req.host_port == "8888"
-
- def test_host_port_w_http_host_https_and_no_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com"}
- req = self._makeOne(environ)
- assert req.host_port == "443"
-
- def test_host_port_w_http_host_https_and_standard_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com:443"}
- req = self._makeOne(environ)
- assert req.host_port == "443"
-
- def test_host_port_w_http_host_https_and_oddball_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com:8888"}
- req = self._makeOne(environ)
- assert req.host_port == "8888"
-
- def test_host_port_wo_http_host(self):
- environ = {"wsgi.url_scheme": "https", "SERVER_PORT": "4333"}
- req = self._makeOne(environ)
- assert req.host_port == "4333"
-
- def test_host_port_ipv6(self):
- environ = {"HTTP_HOST": "[2001:DB8::1]:6453"}
- req = self._makeOne(environ)
- assert req.host_port == "6453"
-
- def test_host_ipv6(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "[2001:DB8::1]"}
- req = self._makeOne(environ)
- assert req.host_port == "443"
-
- def test_host_url_w_http_host_and_no_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com"}
- req = self._makeOne(environ)
- assert req.host_url == "http://example.com"
-
- def test_host_url_w_http_host_and_standard_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com:80"}
- req = self._makeOne(environ)
- assert req.host_url == "http://example.com"
-
- def test_host_url_w_http_host_and_oddball_port(self):
- environ = {"wsgi.url_scheme": "http", "HTTP_HOST": "example.com:8888"}
- req = self._makeOne(environ)
- assert req.host_url == "http://example.com:8888"
-
- def test_host_url_w_http_host_https_and_no_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com"}
- req = self._makeOne(environ)
- assert req.host_url == "https://example.com"
-
- def test_host_url_w_http_host_https_and_standard_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com:443"}
- req = self._makeOne(environ)
- assert req.host_url == "https://example.com"
-
- def test_host_url_w_http_host_https_and_oddball_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "example.com:4333"}
- req = self._makeOne(environ)
- assert req.host_url == "https://example.com:4333"
-
- def test_host_url_wo_http_host(self):
- environ = {
- "wsgi.url_scheme": "https",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "4333",
- }
- req = self._makeOne(environ)
- assert req.host_url == "https://example.com:4333"
-
- def test_host_url_http_ipv6_host(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "[2001:DB8::1]:6453"}
- req = self._makeOne(environ)
- assert req.host_url == "https://[2001:DB8::1]:6453"
-
- def test_host_url_http_ipv6_host_no_port(self):
- environ = {"wsgi.url_scheme": "https", "HTTP_HOST": "[2001:DB8::1]"}
- req = self._makeOne(environ)
- assert req.host_url == "https://[2001:DB8::1]"
-
- @py2only
- def test_application_url_py2(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- app_url = inst.application_url
- assert app_url == "http://localhost/%C3%AB"
-
- @py3only
- def test_application_url(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- app_url = inst.application_url
- assert app_url == "http://localhost/%C3%83%C2%AB"
-
- @py2only
- def test_path_url_py2(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- result = inst.path_url
- assert result == "http://localhost/%C3%AB/%C3%AB"
-
- @py3only
- def test_path_url(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- result = inst.path_url
- assert result == "http://localhost/%C3%83%C2%AB/%C3%83%C2%AB"
-
- @py2only
- def test_path_py2(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- result = inst.path
- assert result == "/%C3%AB/%C3%AB"
-
- @py3only
- def test_path(self):
- inst = self._blankOne("/%C3%AB")
- inst.script_name = b"/\xc3\xab"
- result = inst.path
- assert result == "/%C3%83%C2%AB/%C3%83%C2%AB"
-
- def test_path_qs_no_qs(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- }
- req = self._makeOne(environ)
- assert req.path_qs == "/script/path/info"
-
- def test_path_qs_w_qs(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert req.path_qs == "/script/path/info?foo=bar&baz=bam"
-
- def test_url_no_qs(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- }
- req = self._makeOne(environ)
- assert req.url == "http://example.com/script/path/info"
-
- def test_url_w_qs(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert req.url == "http://example.com/script/path/info?foo=bar&baz=bam"
-
- def test_relative_url_to_app_true_wo_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert req.relative_url(
- "other/page" is True, "http://example.com/script/other/page"
- )
-
- def test_relative_url_to_app_true_w_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert req.relative_url("/other/page" is True, "http://example.com/other/page")
-
- def test_relative_url_to_app_false_other_w_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert req.relative_url("/other/page", False) == "http://example.com/other/page"
-
- def test_relative_url_to_app_false_other_wo_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- "QUERY_STRING": "foo=bar&baz=bam",
- }
- req = self._makeOne(environ)
- assert (
- req.relative_url("other/page", False)
- == "http://example.com/script/path/other/page"
- )
-
- def test_path_info_pop_empty(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop()
- assert popped is None
- assert environ["SCRIPT_NAME"] == "/script"
-
- def test_path_info_pop_just_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop()
- assert popped == ""
- assert environ["SCRIPT_NAME"] == "/script/"
- assert environ["PATH_INFO"] == ""
-
- def test_path_info_pop_non_empty_no_pattern(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop()
- assert popped == "path"
- assert environ["SCRIPT_NAME"] == "/script/path"
- assert environ["PATH_INFO"] == "/info"
-
- def test_path_info_pop_non_empty_w_pattern_miss(self):
- import re
-
- PATTERN = re.compile("miss")
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop(PATTERN)
- assert popped is None
- assert environ["SCRIPT_NAME"] == "/script"
- assert environ["PATH_INFO"] == "/path/info"
-
- def test_path_info_pop_non_empty_w_pattern_hit(self):
- import re
-
- PATTERN = re.compile("path")
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path/info",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop(PATTERN)
- assert popped == "path"
- assert environ["SCRIPT_NAME"] == "/script/path"
- assert environ["PATH_INFO"] == "/info"
-
- def test_path_info_pop_skips_empty_elements(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "//path/info",
- }
- req = self._makeOne(environ)
- popped = req.path_info_pop()
- assert popped == "path"
- assert environ["SCRIPT_NAME"] == "/script//path"
- assert environ["PATH_INFO"] == "/info"
-
- def test_path_info_peek_empty(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "",
- }
- req = self._makeOne(environ)
- peeked = req.path_info_peek()
- assert peeked is None
- assert environ["SCRIPT_NAME"] == "/script"
- assert environ["PATH_INFO"] == ""
-
- def test_path_info_peek_just_leading_slash(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/",
- }
- req = self._makeOne(environ)
- peeked = req.path_info_peek()
- assert peeked == ""
- assert environ["SCRIPT_NAME"] == "/script"
- assert environ["PATH_INFO"] == "/"
-
- def test_path_info_peek_non_empty(self):
- environ = {
- "wsgi.url_scheme": "http",
- "SERVER_NAME": "example.com",
- "SERVER_PORT": "80",
- "SCRIPT_NAME": "/script",
- "PATH_INFO": "/path",
- }
- req = self._makeOne(environ)
- peeked = req.path_info_peek()
- assert peeked == "path"
- assert environ["SCRIPT_NAME"] == "/script"
- assert environ["PATH_INFO"] == "/path"
-
- def test_is_xhr_no_header(self):
- req = self._makeOne({})
- assert not req.is_xhr
-
- def test_is_xhr_header_miss(self):
- environ = {"HTTP_X_REQUESTED_WITH": "notAnXMLHTTPRequest"}
- req = self._makeOne(environ)
- assert not req.is_xhr
-
- def test_is_xhr_header_hit(self):
- environ = {"HTTP_X_REQUESTED_WITH": "XMLHttpRequest"}
- req = self._makeOne(environ)
- assert req.is_xhr
-
- # host
- def test_host_getter_w_HTTP_HOST(self):
- environ = {"HTTP_HOST": "example.com:8888"}
- req = self._makeOne(environ)
- assert req.host == "example.com:8888"
-
- def test_host_getter_wo_HTTP_HOST(self):
- environ = {"SERVER_NAME": "example.com", "SERVER_PORT": "8888"}
- req = self._makeOne(environ)
- assert req.host == "example.com:8888"
-
- def test_host_setter(self):
- environ = {}
- req = self._makeOne(environ)
- req.host = "example.com:8888"
- assert environ["HTTP_HOST"] == "example.com:8888"
-
- def test_host_deleter_hit(self):
- environ = {"HTTP_HOST": "example.com:8888"}
- req = self._makeOne(environ)
- del req.host
- assert "HTTP_HOST" not in environ
-
- def test_host_deleter_miss(self):
- environ = {}
- req = self._makeOne(environ)
- del req.host # doesn't raise
-
- def test_encget_raises_without_default(self):
- inst = self._makeOne({})
- with pytest.raises(KeyError):
- inst.encget("a")
-
- def test_encget_doesnt_raises_with_default(self):
- inst = self._makeOne({})
- assert inst.encget("a", None) is None
-
- def test_encget_with_encattr(self):
- val = native_(b"\xc3\xab", "latin-1")
- inst = self._makeOne({"a": val})
- assert inst.encget("a", encattr="url_encoding") == native_(
- b"\xc3\xab", "latin-1"
- )
-
- def test_encget_no_encattr(self):
- val = native_(b"\xc3\xab", "latin-1")
- inst = self._makeOne({"a": val})
- assert inst.encget("a"), native_(b"\xc3\xab" == "latin-1")
-
- @py2only
- def test_relative_url_py2(self):
- inst = self._blankOne("/%C3%AB/c")
- result = inst.relative_url("a")
- assert result == "http://localhost/%C3%AB/a"
-
- @py3only
- def test_relative_url(self):
- inst = self._blankOne("/%C3%AB/c")
- result = inst.relative_url("a")
- assert result == "http://localhost/%C3%83%C2%AB/a"
-
- def test_header_getter(self):
- val = native_(b"abc", "latin-1")
- inst = self._makeOne({"HTTP_FLUB": val})
- result = inst.headers["Flub"]
- assert result == "abc"
-
- def test_json_body(self):
- inst = self._makeOne({})
- inst.body = b'{"a":"1"}'
- assert inst.json_body == {"a": "1"}
-
- def test_host_get_w_http_host(self):
- inst = self._makeOne({"HTTP_HOST": "example.com"})
- result = inst.host
- assert result == "example.com"
-
- def test_host_get_w_no_http_host(self):
- inst = self._makeOne({"SERVER_NAME": "example.com", "SERVER_PORT": "80"})
- result = inst.host
- assert result == "example.com:80"
-
-
-class TestRequestConstructorWarnings(object):
- def _getTargetClass(self):
- from webob.request import Request
-
- return Request
-
- def _makeOne(self, *arg, **kw):
- cls = self._getTargetClass()
-
- return cls(*arg, **kw)
-
- def test_ctor_w_unicode_errors(self):
- with warnings.catch_warnings(record=True) as w:
- # still emit if warning was printed previously
- warnings.simplefilter("always")
- self._makeOne({}, unicode_errors=True)
- assert len(w) == 1
-
- def test_ctor_w_decode_param_names(self):
- with warnings.catch_warnings(record=True) as w:
- # still emit if warning was printed previously
- warnings.simplefilter("always")
- self._makeOne({}, decode_param_names=True)
- assert len(w) == 1
-
-
-class TestRequestWithAdhocAttr(object):
+class TestRequestWithAdhocAttr:
def _blankOne(self, *arg, **kw):
from webob.request import Request
@@ -2545,7 +1851,7 @@ class TestRequestWithAdhocAttr(object):
delattr(req, "some_attr")
-class TestRequest_functional(object):
+class TestRequest_functional:
# functional tests of request
def _getTargetClass(self):
from webob.request import Request
@@ -2770,7 +2076,7 @@ class TestRequest_functional(object):
assert list(new_params.items()) == [("a", "1"), ("b", "4")]
# The key name is \u1000:
req = self._blankOne("/?%E1%80%80=x")
- val = text_type(b"\\u1000", "unicode_escape")
+ val = str(b"\\u1000", "unicode_escape")
assert val in list(req.GET.keys())
assert req.GET[val] == "x"
@@ -2800,7 +2106,7 @@ class TestRequest_functional(object):
def test_already_consumed_stream(self):
from webob.request import Request
- body = "something".encode("latin-1")
+ body = b"something"
content_type = "application/x-www-form-urlencoded; charset=latin-1"
environ = {
"wsgi.input": BytesIO(body),
@@ -2965,7 +2271,7 @@ class TestRequest_functional(object):
for i in headers.keys():
assert i in r.headers and "HTTP_" + i.upper().replace("-", "_") in r.environ
r.headers = {"Server": "Apache"}
- assert set(r.environ.keys()) == set(["a", "HTTP_SERVER"])
+ assert set(r.environ.keys()) == {"a", "HTTP_SERVER"}
def test_host_url(self):
# Request has a read only property host_url that combines several
@@ -2974,12 +2280,12 @@ class TestRequest_functional(object):
assert a.host_url == "http://www.example.com"
a = self._makeOne(
{"wsgi.url_scheme": "http"},
- **{"server_name": "localhost", "server_port": 5000}
+ **{"server_name": "localhost", "server_port": 5000},
)
assert a.host_url == "http://localhost:5000"
a = self._makeOne(
{"wsgi.url_scheme": "https"},
- **{"server_name": "localhost", "server_port": 443}
+ **{"server_name": "localhost", "server_port": 443},
)
assert a.host_url == "https://localhost"
@@ -3360,9 +2666,7 @@ class TestRequest_functional(object):
# port from doctest (docs/reference.txt)
# Query & POST variables
- from webob.multidict import MultiDict
- from webob.multidict import NestedMultiDict
- from webob.multidict import GetDict
+ from webob.multidict import GetDict, MultiDict, NestedMultiDict
req = self._blankOne("/test?check=a&check=b&name=Bob")
GET = GetDict([("check", "a"), ("check", "b"), ("name", "Bob")], {})
@@ -3397,13 +2701,12 @@ class TestRequest_functional(object):
@pytest.mark.filterwarnings("ignore:.*best_match.*")
def test_request_put(self):
from datetime import datetime
- from webob import Response
- from webob import UTC
+
+ from webob import UTC, Response
from webob.acceptparse import Accept
from webob.byterange import Range
from webob.etag import ETagMatcher
- from webob.multidict import MultiDict
- from webob.multidict import GetDict
+ from webob.multidict import GetDict, MultiDict
req = self._blankOne("/test?check=a&check=b&name=Bob")
req.method = "PUT"
@@ -3498,8 +2801,7 @@ class TestRequest_functional(object):
assert server_token not in req.if_match
def test_request_patch(self):
- from webob.multidict import MultiDict
- from webob.multidict import GetDict
+ from webob.multidict import GetDict, MultiDict
req = self._blankOne("/test?check=a&check=b&name=Bob")
req.method = "PATCH"
@@ -3620,120 +2922,14 @@ class TestRequest_functional(object):
assert req_body == req2_body
-@pytest.mark.filterwarnings("ignore:FakeCGIBody")
-class TestFakeCGIBody(object):
- def test_encode_multipart_value_type_options(self):
- from cgi import FieldStorage
- from webob.request import BaseRequest, FakeCGIBody
- from webob.multidict import MultiDict
-
- multipart_type = "multipart/form-data; boundary=foobar"
- from io import BytesIO
-
- body = (
- b"--foobar\r\n"
- b'Content-Disposition: form-data; name="bananas"; '
- b'filename="bananas.txt"\r\n'
- b'Content-type: text/plain; charset="utf-7"\r\n'
- b"\r\n"
- b"these are the contents of the file 'bananas.txt'\r\n"
- b"\r\n"
- b"--foobar--"
- )
- multipart_body = BytesIO(body)
- environ = BaseRequest.blank("/").environ
- environ.update(CONTENT_TYPE=multipart_type)
- environ.update(REQUEST_METHOD="POST")
- environ.update(CONTENT_LENGTH=len(body))
- fs = FieldStorage(multipart_body, environ=environ)
- vars = MultiDict.from_fieldstorage(fs)
- assert vars["bananas"].__class__ == FieldStorage
- fake_body = FakeCGIBody(vars, multipart_type)
- assert fake_body.read() == body
-
- def test_encode_multipart_no_boundary(self):
- from webob.request import FakeCGIBody
-
- with pytest.raises(ValueError):
- FakeCGIBody({}, "multipart/form-data")
-
- def test_repr(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody(
- {"bananas": "bananas"}, "multipart/form-data; boundary=foobar"
- )
- body.read(1)
- import re
-
- assert (
- re.sub(r"\b0x[0-9a-f]+\b", "<whereitsat>", repr(body))
- == "<FakeCGIBody at <whereitsat> viewing {'bananas': 'ba...nas'}>"
- )
-
- def test_fileno(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody(
- {"bananas": "bananas"}, "multipart/form-data; boundary=foobar"
- )
- assert body.fileno() is None
-
- def test_iter(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody(
- {"bananas": "bananas"}, "multipart/form-data; boundary=foobar"
- )
- assert list(body) == [
- b"--foobar\r\n",
- b'Content-Disposition: form-data; name="bananas"\r\n',
- b"\r\n",
- b"bananas\r\n",
- b"--foobar--",
- ]
-
- def test_readline(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody(
- {"bananas": "bananas"}, "multipart/form-data; boundary=foobar"
- )
- assert body.readline() == b"--foobar\r\n"
- assert body.readline() == b'Content-Disposition: form-data; name="bananas"\r\n'
- assert body.readline() == b"\r\n"
- assert body.readline() == b"bananas\r\n"
- assert body.readline() == b"--foobar--"
- # subsequent calls to readline will return ''
-
- def test_read_bad_content_type(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody({"bananas": "bananas"}, "application/jibberjabber")
- with pytest.raises(AssertionError):
- body.read()
-
- def test_read_urlencoded(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody({"bananas": "bananas"}, "application/x-www-form-urlencoded")
- assert body.read() == b"bananas=bananas"
-
- def test_readable(self):
- from webob.request import FakeCGIBody
-
- body = FakeCGIBody({"bananas": "bananas"}, "application/something")
- assert body.readable()
-
-
-class Test_cgi_FieldStorage__repr__patch(object):
+class Test_cgi_FieldStorage__repr__patch:
def _callFUT(self, fake):
- from webob.request import _cgi_FieldStorage__repr__patch
+ from webob.compat import cgi_FieldStorage
- return _cgi_FieldStorage__repr__patch(fake)
+ return cgi_FieldStorage.__repr__(fake)
def test_with_file(self):
- class Fake(object):
+ class Fake:
name = "name"
file = "file"
filename = "filename"
@@ -3744,7 +2940,7 @@ class Test_cgi_FieldStorage__repr__patch(object):
assert result, "FieldStorage('name' == 'filename')"
def test_without_file(self):
- class Fake(object):
+ class Fake:
name = "name"
file = None
filename = "filename"
@@ -3755,14 +2951,14 @@ class Test_cgi_FieldStorage__repr__patch(object):
assert result, "FieldStorage('name', 'filename' == 'value')"
-class TestLimitedLengthFile(object):
+class TestLimitedLengthFile:
def _makeOne(self, file, maxlen):
from webob.request import LimitedLengthFile
return LimitedLengthFile(file, maxlen)
def test_fileno(self):
- class DummyFile(object):
+ class DummyFile:
def fileno(self):
return 1
@@ -3771,7 +2967,7 @@ class TestLimitedLengthFile(object):
assert inst.fileno() == 1
-class Test_environ_from_url(object):
+class Test_environ_from_url:
def _callFUT(self, *arg, **kw):
from webob.request import environ_from_url
@@ -3828,12 +3024,12 @@ class Test_environ_from_url(object):
assert request.upath_info == expected
def test_fileupload_mime_type_detection(self):
- from webob.request import Request
-
# sometimes on win the detected mime type for .jpg will be
# image/pjpeg for ex. so use a non-standard extesion to avoid that
import mimetypes
+ from webob.request import Request
+
mimetypes.add_type("application/x-foo", ".foo")
request = Request.blank(
"/", POST=dict(file1=("foo.foo", "xxx"), file2=("bar.mp3", "xxx"))
@@ -3842,7 +3038,33 @@ class Test_environ_from_url(object):
assert "application/x-foo" in request.body.decode("ascii", str(request))
-class TestRequestMultipart(object):
+def test_environ_add_POST_file_with_content_type():
+ # For benefit of _encode_multipart which did not have adequate coverage for
+ # type_options
+ class FakeFile:
+ def __init__(self, filename, content_type, type_options, value):
+ self.filename = filename
+ self.type = content_type
+ self.type_options = type_options or {}
+ self.value = value
+
+ from webob.request import environ_add_POST, environ_from_url
+
+ env = environ_from_url("http://example.com/")
+ environ_add_POST(
+ env,
+ {
+ "sample_file": FakeFile(
+ "test.txt", "text/plain", {"charset": "utf-8"}, "this is data"
+ ),
+ },
+ )
+ wsgi_input = env["wsgi.input"].read()
+ assert b"this is data" in wsgi_input
+ assert b'Content-type: text/plain; charset="utf-8"' in wsgi_input
+
+
+class TestRequestMultipart:
def test_multipart_with_charset(self):
from webob.request import Request
@@ -3998,7 +3220,7 @@ _test_req2 = _norm_req(_test_req2) + b"\r\n"
_test_req_multipart_charset = _norm_req(_test_req_multipart_charset)
-class UnseekableInput(object):
+class UnseekableInput:
def __init__(self, data):
self.data = data
self.pos = 0
@@ -4020,10 +3242,10 @@ class UnseekableInput(object):
class UnseekableInputWithSeek(UnseekableInput):
def seek(self, pos, rel=0):
- raise IOError("Invalid seek!")
+ raise OSError("Invalid seek!")
-class _Helper_test_request_wrong_clen(object):
+class _Helper_test_request_wrong_clen:
def __init__(self, f):
self.f = f
self.file_ended = False
diff --git a/tests/test_response.py b/tests/test_response.py
index 0acdc06..7f76bbe 100644
--- a/tests/test_response.py
+++ b/tests/test_response.py
@@ -1,15 +1,13 @@
-import zlib
import io
import sys
+import zlib
import pytest
-from webob.request import BaseRequest
-from webob.request import Request
-from webob.response import Response
-from webob.compat import text_
-from webob.compat import bytes_
from webob import cookies
+from webob.request import BaseRequest, Request
+from webob.response import Response
+from webob.util import bytes_, text_
def setup_module(module):
@@ -22,6 +20,7 @@ def teardown_module(module):
def simple_app(environ, start_response):
start_response("200 OK", [("Content-Type", "text/html; charset=UTF-8")])
+
return ["OK"]
@@ -167,7 +166,7 @@ def test_init_doesnt_add_default_content_type_with_bodyless_status():
def test_content_type_supports_unicode():
- content_type = u"text/html"
+ content_type = "text/html"
resp = Response()
resp.content_type = content_type
assert isinstance(resp.headers["Content-Type"], str)
@@ -309,6 +308,7 @@ def test_conditional_response_if_none_match_weak():
resp_weak = Response(
app_iter=["foo\n"], conditional_response=True, headers={"etag": 'W/"bar"'}
)
+
for rq in [req, req_weak]:
for rp in [resp, resp_weak]:
rq.get_response(rp).status_code == 304
@@ -460,6 +460,7 @@ def test_content_length():
def test_app_iter_range():
req = Request.blank("/", range=(2, 5))
+
for app_iter in [
[b"012345"],
[b"0", b"12345"],
@@ -660,6 +661,7 @@ def test_response_file_body_writelines():
)
def test_response_file_body_tell():
import zipfile
+
from webob.response import ResponseBodyFile
rbo = ResponseBodyFile(Response())
@@ -945,73 +947,6 @@ def test_set_cookie_expires_is_None_and_max_age_is_timedelta():
assert val[3].startswith("expires")
-@pytest.mark.filterwarnings('ignore:"expires" will be removed"')
-def test_set_cookie_expires_is_datetime_and_max_age_is_None():
- import datetime
-
- res = Response()
- then = datetime.datetime.utcnow() + datetime.timedelta(days=1)
- res.set_cookie("a", "1", expires=then)
- assert res.headerlist[-1][0] == "Set-Cookie"
- val = [x.strip() for x in res.headerlist[-1][1].split(";")]
- assert len(val) == 4
- val.sort()
- assert val[0] in ("Max-Age=86399", "Max-Age=86400")
- assert val[1] == "Path=/"
- assert val[2] == "a=1"
- assert val[3].startswith("expires")
-
-
-@pytest.mark.filterwarnings('ignore:"expires" will be removed"')
-def test_set_cookie_expires_is_timedelta_and_max_age_is_None():
- import datetime
-
- res = Response()
- then = datetime.timedelta(days=1)
- res.set_cookie("a", "1", expires=then)
- assert res.headerlist[-1][0] == "Set-Cookie"
- val = [x.strip() for x in res.headerlist[-1][1].split(";")]
- assert len(val) == 4
- val.sort()
- assert val[0] in ("Max-Age=86399", "Max-Age=86400")
- assert val[1] == "Path=/"
- assert val[2] == "a=1"
- assert val[3].startswith("expires")
-
-
-@pytest.mark.filterwarnings('ignore:"expires" will be removed"')
-def test_set_cookie_expires_is_datetime_tz_and_max_age_is_None():
- import datetime
-
- res = Response()
-
- class FixedOffset(datetime.tzinfo):
- def __init__(self, offset, name):
- self.__offset = datetime.timedelta(minutes=offset)
- self.__name = name
-
- def utcoffset(self, dt):
- return self.__offset
-
- def tzname(self, dt):
- return self.__name
-
- def dst(self, dt):
- return datetime.timedelta(0)
-
- then = datetime.datetime.now(FixedOffset(60, "UTC+1")) + datetime.timedelta(days=1)
-
- res.set_cookie("a", "1", expires=then)
- assert res.headerlist[-1][0] == "Set-Cookie"
- val = [x.strip() for x in res.headerlist[-1][1].split(";")]
- assert len(val) == 4
- val.sort()
- assert val[0] in ("Max-Age=86399", "Max-Age=86400")
- assert val[1] == "Path=/"
- assert val[2] == "a=1"
- assert val[3].startswith("expires")
-
-
def test_delete_cookie():
res = Response()
res.headers["Set-Cookie"] = "a=2; Path=/"
@@ -1095,6 +1030,7 @@ def test_merge_cookies_resp_is_wsgi_callable():
def dummy_wsgi_callable(environ, start_response):
L.append((environ, start_response))
+
return "abc"
res = Response()
@@ -1147,7 +1083,7 @@ def test_location_unicode():
environ = {
"REQUEST_METHOD": "GET",
"wsgi.url_scheme": "http",
- "HTTP_HOST": u"test.com",
+ "HTTP_HOST": "test.com",
}
res = Response()
res.status = "301"
@@ -1238,7 +1174,7 @@ def test_cache_control_set_unicode():
def test_cache_control_set_control_obj_is_not_None():
- class DummyCacheControl(object):
+ class DummyCacheControl:
def __init__(self):
self.header_value = 1
self.properties = {"bleh": 1}
@@ -1536,7 +1472,7 @@ def test_content_type_has_charset():
def test_app_iter_is_same():
- class app_iter(object):
+ class app_iter:
pass
my_app_iter = app_iter()
diff --git a/tests/test_static.py b/tests/test_static.py
index 5845e6d..0d8ad2e 100644
--- a/tests/test_static.py
+++ b/tests/test_static.py
@@ -1,15 +1,15 @@
from io import BytesIO
+import os
from os.path import getmtime
+import shutil
import tempfile
from time import gmtime
-import os
-import shutil
import unittest
from webob import static
-from webob.compat import bytes_
from webob.request import Request, environ_from_url
from webob.response import Response
+from webob.util import bytes_
def get_response(app, path="/", **req_kw):
@@ -84,7 +84,7 @@ class TestFileApp(unittest.TestCase):
# Mock the built-in ``open()`` function to allow finner control about
# what we are testing.
def open_ioerror(*args, **kwargs):
- raise IOError()
+ raise OSError()
def open_oserror(*args, **kwargs):
raise OSError()
@@ -98,7 +98,7 @@ class TestFileApp(unittest.TestCase):
self.assertEqual(403, get_response(app).status_code)
def test_use_wsgi_filewrapper(self):
- class TestWrapper(object):
+ class TestWrapper:
__slots__ = ("file", "block_size")
def __init__(self, file, block_size):
diff --git a/tests/test_transcode.py b/tests/test_transcode.py
index 3a85a12..87a48ef 100644
--- a/tests/test_transcode.py
+++ b/tests/test_transcode.py
@@ -1,6 +1,6 @@
from webob.request import Request, Transcoder
from webob.response import Response
-from webob.compat import text_, native_
+from webob.util import text_
t1 = (
b'--BOUNDARY\r\nContent-Disposition: form-data; name="a"\r\n\r\n\xea\xf3...'
@@ -45,7 +45,7 @@ def test_transcode():
r = test(t2)
assert r.text == "file\n%r" % text.encode("cp1251")
r = test(t3)
- assert r.text, "%s\n%r" % (text == b"foo")
+ assert r.text == "%s\n%r" % (text, b"foo")
def test_transcode_query():
@@ -58,14 +58,14 @@ def test_transcode_non_multipart():
req = Request.blank("/?a", POST="%EF%F0%E8=%E2%E5%F2")
req._content_type_raw = "application/x-www-form-urlencoded"
req2 = req.decode("cp1251")
- assert native_(req2.body) == "%D0%BF%D1%80%D0%B8=%D0%B2%D0%B5%D1%82"
+ assert text_(req2.body) == "%D0%BF%D1%80%D0%B8=%D0%B2%D0%B5%D1%82"
def test_transcode_non_form():
req = Request.blank("/?a", POST="%EF%F0%E8=%E2%E5%F2")
req._content_type_raw = "application/x-foo"
req2 = req.decode("cp1251")
- assert native_(req2.body) == "%EF%F0%E8=%E2%E5%F2"
+ assert text_(req2.body) == "%EF%F0%E8=%E2%E5%F2"
def test_transcode_noop():
diff --git a/tests/test_util.py b/tests/test_util.py
index 7c32491..3f71f2b 100644
--- a/tests/test_util.py
+++ b/tests/test_util.py
@@ -1,23 +1,23 @@
import unittest
+import warnings
+
from webob.response import Response
+from webob.util import warn_deprecation
class Test_warn_deprecation(unittest.TestCase):
def setUp(self):
- import warnings
self.oldwarn = warnings.warn
warnings.warn = self._warn
self.warnings = []
def tearDown(self):
- import warnings
warnings.warn = self.oldwarn
del self.warnings
def _callFUT(self, text, version, stacklevel):
- from webob.util import warn_deprecation
return warn_deprecation(text, version, stacklevel)
@@ -41,59 +41,11 @@ class Test_warn_deprecation(unittest.TestCase):
def test_warn_deprecation(self):
v = "1.3.0"
- from webob.util import warn_deprecation
self.assertRaises(DeprecationWarning, warn_deprecation, "foo", v[:3], 1)
def test_warn_deprecation_future_version(self):
v = "9.9.9"
- from webob.util import warn_deprecation
warn_deprecation("foo", v[:3], 1)
self.assertEqual(len(self.warnings), 1)
-
-
-class Test_strings_differ(unittest.TestCase):
- def _callFUT(self, *args, **kw):
- from webob.util import strings_differ
-
- return strings_differ(*args, **kw)
-
- def test_it(self):
- self.assertFalse(self._callFUT(b"foo", b"foo"))
- self.assertTrue(self._callFUT(b"123", b"345"))
- self.assertTrue(self._callFUT(b"1234", b"123"))
- self.assertTrue(self._callFUT(b"123", b"1234"))
-
- def test_it_with_internal_comparator(self):
- result = self._callFUT(b"foo", b"foo", compare_digest=None)
- self.assertFalse(result)
-
- result = self._callFUT(b"123", b"abc", compare_digest=None)
- self.assertTrue(result)
-
- def test_it_with_external_comparator(self):
- class DummyComparator(object):
- called = False
-
- def __init__(self, ret_val):
- self.ret_val = ret_val
-
- def __call__(self, a, b):
- self.called = True
- return self.ret_val
-
- dummy_compare = DummyComparator(True)
- result = self._callFUT(b"foo", b"foo", compare_digest=dummy_compare)
- self.assertTrue(dummy_compare.called)
- self.assertFalse(result)
-
- dummy_compare = DummyComparator(False)
- result = self._callFUT(b"123", b"345", compare_digest=dummy_compare)
- self.assertTrue(dummy_compare.called)
- self.assertTrue(result)
-
- dummy_compare = DummyComparator(False)
- result = self._callFUT(b"abc", b"abc", compare_digest=dummy_compare)
- self.assertTrue(dummy_compare.called)
- self.assertTrue(result)
diff --git a/tox.ini b/tox.ini
index 0e2c154..3709486 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,25 +1,14 @@
[tox]
envlist =
- py27,py34,py35,py36,py37,py38,pypy,pypy3,
- docs,coverage,lint
-skip_missing_interpreters = True
+ lint,
+ py36,py37,py38,py39,pypy3,
+ docs,
+ coverage
+isolated_build = True
[testenv]
-# Most of these are defaults but if you specify any you can't fall back
-# to defaults for others.
-basepython =
- py27: python2.7
- py34: python3.4
- py35: python3.5
- py36: python3.6
- py37: python3.7
- py38: python3.8
- pypy: pypy
- pypy3: pypy3
- py2: python2.7
- py3: python3.5
-
commands =
+ python --version
pytest {posargs:}
extras =
testing
@@ -27,71 +16,76 @@ setenv =
COVERAGE_FILE=.coverage.{envname}
[testenv:coverage]
-basepython = python3.6
+skip_install = True
commands =
coverage combine
coverage xml
- coverage report --show-missing --fail-under=100
+ coverage report --fail-under=100
deps =
coverage
setenv =
COVERAGE_FILE=.coverage
-
-[testenv:docs]
-basepython = python3.6
-whitelist_externals = make
-commands =
- make -C docs html epub BUILDDIR={envdir} "SPHINXOPTS=-W -E"
-extras =
- docs
+depends = py38
[testenv:lint]
skip_install = True
-basepython = python3.7
commands =
black --check --diff .
- python setup.py check -r -s -m
+ isort --check-only --df src/webob tests
check-manifest
+ # flake8 src/webob/ tests
+ # build sdist/wheel
+ python -m pep517.build .
+ twine check dist/*
deps =
black
- readme_renderer
check-manifest
+ flake8
+ flake8-bugbear
+ isort
+ pep517
+ readme_renderer
+ twine
+
+[testenv:docs]
+whitelist_externals =
+ make
+commands =
+ make -C docs html BUILDDIR={envdir} SPHINXOPTS="-W -E"
+extras =
+ docs
[testenv:run-flake8]
skip_install = True
-basepython = python3.7
commands =
flake8 src/webob/ tests
deps =
flake8
flake8-bugbear
-[testenv:run-black]
+[testenv:run-format]
skip_install = True
-basepython = python3.7
commands =
+ isort src/webob tests
black .
deps =
black
+ isort
[testenv:build]
skip_install = true
-basepython = python3.7
commands =
# clean up build/ and dist/ folders
- python -c 'import shutil; shutil.rmtree("dist", ignore_errors=True)'
- python setup.py clean --all
- # Do some basic sanity checks
- python setup.py check -r -s -m
+ python -c 'import shutil; shutil.rmtree("build", ignore_errors=True)'
# Make sure we aren't forgetting anything
check-manifest
- # build sdist
- python setup.py sdist --dist-dir {toxinidir}/dist
- # build wheel from sdist
- pip wheel -v --no-deps --no-index --no-build-isolation --wheel-dir {toxinidir}/dist --find-links {toxinidir}/dist webob
+ # build sdist/wheel
+ python -m pep517.build .
+ # Verify all is well
+ twine check dist/*
deps =
readme_renderer
check-manifest
- setuptools
- wheel
+ pep517
+ twine