diff options
| author | Bert JW Regeer <xistence@0x58.com> | 2020-11-28 17:56:52 -0800 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2020-11-28 17:56:52 -0800 |
| commit | 536df8686cc815f0e0ed22e6a7402b83dbec5a00 (patch) | |
| tree | 1e4ba57db9579d499273344d317c34422acd1c6f | |
| parent | 07ce90f2da58990b83d06f41b035d3b8ee22547b (diff) | |
| parent | 0db42eb954a65c759aaaf12490f4b2611b2439b5 (diff) | |
| download | webob-536df8686cc815f0e0ed22e6a7402b83dbec5a00.tar.gz | |
Merge pull request #390 from Pylons/webob-ng
WebOb 2.0 - Now with less Python 2
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" @@ -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;"), # 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("'", "'", marks=py3only), - (text_("the majestic m\xf8ose"), "the majestic møose"), + pytest.param("'", "'"), + ("the majestic m\xf8ose", "the majestic møose"), # 8-bit strings are passed through - (text_("\xe9"), "é"), + ("\xe9", "é"), # ``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øose"), (t_esc_Unicode(), "é"), (t_esc_UnsafeAttrs(), "<UnsafeAttrs>"), - pytest.param(Exception("expected a '<'."), "expected a '<'.", marks=py2only), - pytest.param( - Exception("expected a '<'."), "expected a '<'.", marks=py3only - ), + pytest.param(Exception("expected a '<'."), "expected a '<'."), ], ) 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) @@ -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 |
