Skip to content

Instantly share code, notes, and snippets.

@tudormunteanu

tudormunteanu/router.py

Created November 9, 2018 11:44
Show Gist options
  • Save tudormunteanu/b293df878c00a0faa5dc782fa0cf354e to your computer and use it in GitHub Desktop.
Save tudormunteanu/b293df878c00a0faa5dc782fa0cf354e to your computer and use it in GitHub Desktop.
Download ZIP
Sanic Header Based Versioning Router
Raw
router.py
import re
import uuid
from collections import defaultdict, namedtuple
from collections.abc import Iterable
from functools import lru_cache
from urllib.parse import unquote
from sanic.constants import HTTP_METHODS
from sanic.exceptions import InvalidUsage
from sanic.exceptions import MethodNotSupported, NotFound
# from sanic.views import CompositionView
DEFAULT_ROUTE_VERSION = '1'
class CompositionView:
"""Simple method-function mapped view for the sanic.
You can add handler functions to methods (get, post, put, patch, delete)
for every HTTP method you want to support.
For example:
view = CompositionView()
view.add(['GET'], lambda request: text('I am get method'))
view.add(['POST', 'PUT'], lambda request: text('I am post/put method'))
etc.
If someone tries to use a non-implemented method, there will be a
405 response.
"""
def __init__(self):
self.handlers = {}
def add(self, methods, handler, stream=False, version=None):
if stream:
handler.is_stream = stream
for method in methods:
if method not in HTTP_METHODS:
raise InvalidUsage(
"{} is not a valid HTTP method.".format(method)
)
if (method, version) in self.handlers:
import pdb; pdb.set_trace()
raise InvalidUsage(
"Method {} is already registered.".format(method)
)
self.handlers[(method, version)] = handler
def __call__(self, request, *args, **kwargs):
handler = self.handlers[request.method.upper()]
return handler(request, *args, **kwargs)
Route = namedtuple(
"Route", ["handler", "methods", "pattern", "parameters", "name", "uri", "version"]
)
Parameter = namedtuple("Parameter", ["name", "cast"])
REGEX_TYPES = {
"string": (str, r"[^/]+"),
"int": (int, r"\d+"),
"number": (float, r"[0-9\\.]+"),
"alpha": (str, r"[A-Za-z]+"),
"path": (str, r"[^/].*?"),
"uuid": (
uuid.UUID,
r"[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-"
r"[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}",
),
}
ROUTER_CACHE_SIZE = 1024
def url_hash(url):
return url.count("/")
class RouteExists(Exception):
pass
class RouteDoesNotExist(Exception):
pass
class ParameterNameConflicts(Exception):
pass
class Router:
"""Router supports basic routing with parameters and method checks
Usage:
.. code-block:: python
@sanic.route('/my/url/<my_param>', methods=['GET', 'POST', ...])
def my_route(request, my_param):
do stuff...
or
.. code-block:: python
@sanic.route('/my/url/<my_param:my_type>', methods['GET', 'POST', ...])
def my_route_with_type(request, my_param: my_type):
do stuff...
Parameters will be passed as keyword arguments to the request handling
function. Provided parameters can also have a type by appending :type to
the <parameter>. Given parameter must be able to be type-casted to this.
If no type is provided, a string is expected. A regular expression can
also be passed in as the type. The argument given to the function will
always be a string, independent of the type.
"""
routes_static = None
routes_dynamic = None
routes_always_check = None
parameter_pattern = re.compile(r"<(.+?)>")
def __init__(self):
self.routes_all = {}
self.routes_names = {}
self.routes_static_files = {}
self.routes_static = {}
self.routes_dynamic = defaultdict(list)
self.routes_always_check = []
self.hosts = set()
@classmethod
def parse_parameter_string(cls, parameter_string):
"""Parse a parameter string into its constituent name, type, and
pattern
For example::
parse_parameter_string('<param_one:[A-z]>')` ->
('param_one', str, '[A-z]')
:param parameter_string: String to parse
:return: tuple containing
(parameter_name, parameter_type, parameter_pattern)
"""
# We could receive NAME or NAME:PATTERN
name = parameter_string
pattern = "string"
if ":" in parameter_string:
name, pattern = parameter_string.split(":", 1)
if not name:
raise ValueError(
"Invalid parameter syntax: {}".format(parameter_string)
)
default = (str, pattern)
# Pull from pre-configured types
_type, pattern = REGEX_TYPES.get(pattern, default)
return name, _type, pattern
def add(
self,
uri,
methods,
handler,
host=None,
strict_slashes=False,
version=None,
name=None,
):
"""Add a handler to the route list
:param uri: path to match
:param methods: sequence of accepted method names. If none are
provided, any method is allowed
:param handler: request handler function.
When executed, it should provide a response object.
:param strict_slashes: strict to trailing slash
:param version: current version of the route or blueprint. See
docs for further details.
:return: Nothing
"""
# if version is not None:
# version = re.escape(str(version).strip("/").lstrip("v"))
# uri = "/".join(["/v{}".format(version), uri.lstrip("/")])
# add regular version
if not version:
version = DEFAULT_ROUTE_VERSION
self._add(uri, methods, handler, host, name, str(version))
if strict_slashes:
return
if not isinstance(host, str) and host is not None:
# we have gotten back to the top of the recursion tree where the
# host was originally a list. By now, we've processed the strict
# slashes logic on the leaf nodes (the individual host strings in
# the list of host)
return
# Add versions with and without trailing /
slashed_methods = self.routes_all.get(uri + "/", frozenset({}))
unslashed_methods = self.routes_all.get(uri[:-1], frozenset({}))
if isinstance(methods, Iterable):
_slash_is_missing = all(
method in slashed_methods for method in methods
)
_without_slash_is_missing = all(
method in unslashed_methods for method in methods
)
else:
_slash_is_missing = methods in slashed_methods
_without_slash_is_missing = methods in unslashed_methods
slash_is_missing = not uri[-1] == "/" and not _slash_is_missing
without_slash_is_missing = (
uri[-1] == "/" and not _without_slash_is_missing and not uri == "/"
)
# add version with trailing slash
if slash_is_missing:
self._add(uri + "/", methods, handler, host, name, str(version))
# add version without trailing slash
elif without_slash_is_missing:
self._add(uri[:-1], methods, handler, host, name, str(version))
def _add(self, uri, methods, handler, host=None, name=None, version=None):
"""Add a handler to the route list
:param uri: path to match
:param methods: sequence of accepted method names. If none are
provided, any method is allowed
:param handler: request handler function.
When executed, it should provide a response object.
:param name: user defined route name for url_for
:return: Nothing
"""
if host is not None:
if isinstance(host, str):
uri = host + uri
self.hosts.add(host)
else:
if not isinstance(host, Iterable):
raise ValueError(
"Expected either string or Iterable of "
"host strings, not {!r}".format(host)
)
for host_ in host:
self.add(uri, methods, handler, host_, name)
return
# Dict for faster lookups of if method allowed
if methods:
methods = frozenset(methods)
parameters = []
parameter_names = set()
properties = {"unhashable": None}
def add_parameter(match):
name = match.group(1)
name, _type, pattern = self.parse_parameter_string(name)
if name in parameter_names:
raise ParameterNameConflicts(
"Multiple parameter named <{name}> "
"in route uri {uri}".format(name=name, uri=uri)
)
parameter_names.add(name)
parameter = Parameter(name=name, cast=_type)
parameters.append(parameter)
# Mark the whole route as unhashable if it has the hash key in it
if re.search(r"(^|[^^]){1}/", pattern):
properties["unhashable"] = True
# Mark the route as unhashable if it matches the hash key
elif re.search(r"/", pattern):
properties["unhashable"] = True
return "({})".format(pattern)
pattern_string = re.sub(self.parameter_pattern, add_parameter, uri)
pattern = re.compile(r"^{}$".format(pattern_string))
def merge_route(route, methods, handler, version=None):
# merge to the existing route when possible.
if not route.methods or not methods:
# method-unspecified routes are not mergeable.
raise RouteExists("Route already registered: {}".format(uri))
elif route.methods.intersection(methods) and route.version == version:
# already existing method is not overloadable.
duplicated = methods.intersection(route.methods)
raise RouteExists(
"Route already registered: {} [{}]".format(
uri, ",".join(list(duplicated))
)
)
if isinstance(route.handler, CompositionView):
view = route.handler
else:
view = CompositionView()
view.add(route.methods, route.handler, version=route.version)
view.add(methods, handler, version=version)
route = route._replace(
handler=view, methods=methods.union(route.methods)
)
return route
if parameters:
# TODO: This is too complex, we need to reduce the complexity
if properties["unhashable"]:
routes_to_check = self.routes_always_check
ndx, route = self.check_dynamic_route_exists(
pattern, routes_to_check, parameters
)
else:
routes_to_check = self.routes_dynamic[url_hash(uri)]
ndx, route = self.check_dynamic_route_exists(
pattern, routes_to_check, parameters
)
if ndx != -1:
# Pop the ndx of the route, no dups of the same route
routes_to_check.pop(ndx)
else:
route = self.routes_all.get(uri)
# prefix the handler name with the blueprint name
# if available
# special prefix for static files
is_static = False
if name and name.startswith("_static_"):
is_static = True
name = name.split("_static_", 1)[-1]
if hasattr(handler, "__blueprintname__"):
handler_name = "{}.{}".format(
handler.__blueprintname__, name or handler.__name__
)
else:
handler_name = name or getattr(handler, "__name__", None)
if route:
route = merge_route(route, methods, handler, version)
else:
route = Route(
handler=handler,
methods=methods,
pattern=pattern,
parameters=parameters,
name=handler_name,
uri=uri,
version=version
)
self.routes_all[uri] = route
if is_static:
pair = self.routes_static_files.get(handler_name)
if not (pair and (pair[0] + "/" == uri or uri + "/" == pair[0])):
self.routes_static_files[handler_name] = (uri, route)
else:
pair = self.routes_names.get(handler_name)
if not (pair and (pair[0] + "/" == uri or uri + "/" == pair[0])):
self.routes_names[handler_name] = (uri, route)
if properties["unhashable"]:
self.routes_always_check.append(route)
elif parameters:
self.routes_dynamic[url_hash(uri)].append(route)
else:
self.routes_static[uri] = route
@staticmethod
def check_dynamic_route_exists(pattern, routes_to_check, parameters):
for ndx, route in enumerate(routes_to_check):
if route.pattern == pattern and route.parameters == parameters:
return ndx, route
else:
return -1, None
def remove(self, uri, clean_cache=True, host=None):
if host is not None:
uri = host + uri
try:
route = self.routes_all.pop(uri)
for handler_name, pairs in self.routes_names.items():
if pairs[0] == uri:
self.routes_names.pop(handler_name)
break
for handler_name, pairs in self.routes_static_files.items():
if pairs[0] == uri:
self.routes_static_files.pop(handler_name)
break
except KeyError:
raise RouteDoesNotExist("Route was not registered: {}".format(uri))
if route in self.routes_always_check:
self.routes_always_check.remove(route)
elif (
url_hash(uri) in self.routes_dynamic
and route in self.routes_dynamic[url_hash(uri)]
):
self.routes_dynamic[url_hash(uri)].remove(route)
else:
self.routes_static.pop(uri)
if clean_cache:
self._get.cache_clear()
@lru_cache(maxsize=ROUTER_CACHE_SIZE)
def find_route_by_view_name(self, view_name, name=None):
"""Find a route in the router based on the specified view name.
:param view_name: string of view name to search by
:param kwargs: additional params, usually for static files
:return: tuple containing (uri, Route)
"""
if not view_name:
return (None, None)
if view_name == "static" or view_name.endswith(".static"):
return self.routes_static_files.get(name, (None, None))
return self.routes_names.get(view_name, (None, None))
def get(self, request):
"""Get a request handler based on the URL of the request, or raises an
error
:param request: Request object
:return: handler, arguments, keyword arguments
"""
def get_version(accept_header):
try:
return str(int(accept_header))
except ValueError:
return '1'
# No virtual hosts specified; default behavior
if not self.hosts:
return self._get(request.path, request.method, "", get_version(request.headers.get('Accept')))
# virtual hosts specified; try to match route to the host header
try:
return self._get(
request.path, request.method, request.headers.get("Host", "")
)
# try default hosts
except NotFound:
return self._get(request.path, request.method, "")
def get_supported_methods(self, url):
"""Get a list of supported methods for a url and optional host.
:param url: URL string (including host)
:return: frozenset of supported methods
"""
route = self.routes_all.get(url)
# if methods are None then this logic will prevent an error
return getattr(route, "methods", None) or frozenset()
@lru_cache(maxsize=ROUTER_CACHE_SIZE)
def _get(self, url, method, host, version=None):
"""Get a request handler based on the URL of the request, or raises an
error. Internal method for caching.
:param url: request URL
:param method: request method
:return: handler, arguments, keyword arguments
"""
url = unquote(host + url)
# Check against known static routes
route = self.routes_static.get(url)
method_not_supported = MethodNotSupported(
"Method {} not allowed for URL {}".format(method, url),
method=method,
allowed_methods=self.get_supported_methods(url),
)
if route:
if route.methods and method not in route.methods:
raise method_not_supported
match = route.pattern.match(url)
else:
route_found = False
# Move on to testing all regex routes
for route in self.routes_dynamic[url_hash(url)]:
match = route.pattern.match(url)
route_found |= match is not None
# Do early method checking
if match and method in route.methods:
break
else:
# Lastly, check against all regex routes that cannot be hashed
for route in self.routes_always_check:
match = route.pattern.match(url)
route_found |= match is not None
# Do early method checking
if match and method in route.methods:
break
else:
# Route was found but the methods didn't match
if route_found:
raise method_not_supported
raise NotFound("Requested URL {} not found".format(url))
kwargs = {
p.name: p.cast(value)
for value, p in zip(match.groups(1), route.parameters)
}
route_handler = route.handler
if hasattr(route_handler, "handlers"):
route_handler = route_handler.handlers[(method, version)]
return route_handler, [], kwargs, route.uri
def is_stream_handler(self, request):
""" Handler for request is stream or not.
:param request: Request object
:return: bool
"""
try:
handler = self.get(request)[0]
except (NotFound, MethodNotSupported):
return False
if hasattr(handler, "view_class") and hasattr(
handler.view_class, request.method.lower()
):
handler = getattr(handler.view_class, request.method.lower())
return hasattr(handler, "is_stream")
Raw
test_routes.py
import asyncio
import pytest
from sanic import Sanic
from sanic.response import text, json
from sanic.constants import HTTP_METHODS
from guacamole.router import RouteExists, RouteDoesNotExist, ParameterNameConflicts, Router
@pytest.fixture
def app(request):
return Sanic(request.node.name, router=Router())
# ------------------------------------------------------------ #
# UTF-8
# ------------------------------------------------------------ #
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_routes(app, method):
method = method.lower()
func = getattr(app, method)
if callable(func):
@func('/{}'.format(method), version='1')
def handler1(request):
return text('OK v1')
@func('/{}'.format(method), version='2')
def handler2(request):
return text('OK v2')
else:
print(func)
raise
client_method = getattr(app.test_client, method)
request1, response1 = client_method('/{}'.format(method), headers={'Accept': '1'})
request2, response2 = client_method('/{}'.format(method), headers={'Accept': '2'})
assert response1.status == 200
assert response2.status == 200
if method not in ('head',):
assert response1.text == 'OK v1'
assert response2.text == 'OK v2'
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_routes_default(app, method):
method = method.lower()
func = getattr(app, method)
if callable(func):
@func('/{}'.format(method), version='1')
def handler1(request):
return text('OK v1')
@func('/{}'.format(method), version='2')
def handler2(request):
return text('OK v2')
else:
print(func)
raise
client_method = getattr(app.test_client, method)
request1, response1 = client_method('/{}'.format(method))
request2, response2 = client_method('/{}'.format(method), headers={'Accept': '2'})
assert response1.status == 200
assert response2.status == 200
if method not in ('head',):
assert response1.text == 'OK v1'
assert response2.text == 'OK v2'
def test_shorthand_routes_get(app):
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get')
assert response.text == 'OK'
request, response = app.test_client.post('/get')
assert response.status == 405
def test_shorthand_routes_multiple(app):
@app.get('/get')
def get_handler(request):
return text('OK')
@app.options('/get')
def options_handler(request):
return text('')
request, response = app.test_client.get('/get/')
assert response.status == 200
assert response.text == 'OK'
request, response = app.test_client.options('/get/')
assert response.status == 200
def test_route_strict_slash(app):
@app.get('/get', strict_slashes=True)
def handler1(request):
assert request.stream is None
return text('OK')
@app.post('/post/', strict_slashes=True)
def handler2(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.get('/get')
assert response.text == 'OK'
request, response = app.test_client.get('/get/')
assert response.status == 404
request, response = app.test_client.post('/post/')
assert response.text == 'OK'
request, response = app.test_client.post('/post')
assert response.status == 404
def test_route_invalid_parameter_syntax(app):
with pytest.raises(ValueError):
@app.get('/get/<:string>', strict_slashes=True)
def handler(request):
return text('OK')
request, response = app.test_client.get('/get')
def test_route_strict_slash_default_value():
app = Sanic('test_route_strict_slash', strict_slashes=True)
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.status == 404
def test_route_strict_slash_without_passing_default_value(app):
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_strict_slash_default_value_can_be_overwritten():
app = Sanic('test_route_strict_slash', strict_slashes=True)
@app.get('/get', strict_slashes=False)
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_slashes_overload(app):
@app.get('/hello/')
def handler_get(request):
return text('OK')
@app.post('/hello/')
def handler_post(request):
return text('OK')
request, response = app.test_client.get('/hello')
assert response.text == 'OK'
request, response = app.test_client.get('/hello/')
assert response.text == 'OK'
request, response = app.test_client.post('/hello')
assert response.text == 'OK'
request, response = app.test_client.post('/hello/')
assert response.text == 'OK'
def test_route_optional_slash(app):
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get')
assert response.text == 'OK'
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
# Part of regression test for issue #1120
site1 = '127.0.0.1:{}'.format(app.test_client.port)
# before fix, this raises a RouteExists error
@app.get('/get', host=[site1, 'site2.com'], strict_slashes=False)
def get_handler(request):
return text('OK')
request, response = app.test_client.get('http://' + site1 + '/get')
assert response.text == 'OK'
@app.post('/post', host=[site1, 'site2.com'], strict_slashes=False)
def post_handler(request):
return text('OK')
request, response = app.test_client.post('http://' + site1 + '/post')
assert response.text == 'OK'
@app.put('/put', host=[site1, 'site2.com'], strict_slashes=False)
def put_handler(request):
return text('OK')
request, response = app.test_client.put('http://' + site1 + '/put')
assert response.text == 'OK'
@app.delete('/delete', host=[site1, 'site2.com'], strict_slashes=False)
def delete_handler(request):
return text('OK')
request, response = app.test_client.delete('http://' + site1 + '/delete')
assert response.text == 'OK'
def test_shorthand_routes_post(app):
@app.post('/post')
def handler(request):
return text('OK')
request, response = app.test_client.post('/post')
assert response.text == 'OK'
request, response = app.test_client.get('/post')
assert response.status == 405
def test_shorthand_routes_put(app):
@app.put('/put')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.put('/put')
assert response.text == 'OK'
request, response = app.test_client.get('/put')
assert response.status == 405
def test_shorthand_routes_delete(app):
@app.delete('/delete')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.delete('/delete')
assert response.text == 'OK'
request, response = app.test_client.get('/delete')
assert response.status == 405
def test_shorthand_routes_patch(app):
@app.patch('/patch')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.patch('/patch')
assert response.text == 'OK'
request, response = app.test_client.get('/patch')
assert response.status == 405
def test_shorthand_routes_head(app):
@app.head('/head')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.head('/head')
assert response.status == 200
request, response = app.test_client.get('/head')
assert response.status == 405
def test_shorthand_routes_options(app):
@app.options('/options')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
request, response = app.test_client.options('/options')
assert response.status == 200
request, response = app.test_client.get('/options')
assert response.status == 405
def test_static_routes(app):
@app.route('/test')
async def handler1(request):
return text('OK1')
@app.route('/pizazz')
async def handler2(request):
return text('OK2')
request, response = app.test_client.get('/test')
assert response.text == 'OK1'
request, response = app.test_client.get('/pizazz')
assert response.text == 'OK2'
def test_dynamic_route(app):
results = []
@app.route('/folder/<name>')
async def handler(request, name):
results.append(name)
return text('OK')
request, response = app.test_client.get('/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
def test_dynamic_route_string(app):
results = []
@app.route('/folder/<name:string>')
async def handler(request, name):
results.append(name)
return text('OK')
request, response = app.test_client.get('/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
request, response = app.test_client.get('/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
def test_dynamic_route_int(app):
results = []
@app.route('/folder/<folder_id:int>')
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
request, response = app.test_client.get('/folder/12345')
assert response.text == 'OK'
assert type(results[0]) is int
request, response = app.test_client.get('/folder/asdf')
assert response.status == 404
def test_dynamic_route_number(app):
results = []
@app.route('/weight/<weight:number>')
async def handler(request, weight):
results.append(weight)
return text('OK')
request, response = app.test_client.get('/weight/12345')
assert response.text == 'OK'
assert type(results[0]) is float
request, response = app.test_client.get('/weight/1234.56')
assert response.status == 200
request, response = app.test_client.get('/weight/1234-56')
assert response.status == 404
def test_dynamic_route_regex(app):
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>')
async def handler(request, folder_id):
return text('OK')
request, response = app.test_client.get('/folder/test')
assert response.status == 200
request, response = app.test_client.get('/folder/test1')
assert response.status == 404
request, response = app.test_client.get('/folder/test-123')
assert response.status == 404
request, response = app.test_client.get('/folder/')
assert response.status == 200
def test_dynamic_route_uuid(app):
import uuid
results = []
@app.route('/quirky/<unique_id:uuid>')
async def handler(request, unique_id):
results.append(unique_id)
return text('OK')
url = '/quirky/123e4567-e89b-12d3-a456-426655440000'
request, response = app.test_client.get(url)
assert response.text == 'OK'
assert type(results[0]) is uuid.UUID
request, response = app.test_client.get('/quirky/{}'.format(uuid.uuid4()))
assert response.status == 200
request, response = app.test_client.get('/quirky/non-existing')
assert response.status == 404
def test_dynamic_route_path(app):
@app.route('/<path:path>/info')
async def handler(request, path):
return text('OK')
request, response = app.test_client.get('/path/1/info')
assert response.status == 200
request, response = app.test_client.get('/info')
assert response.status == 404
@app.route('/<path:path>')
async def handler1(request, path):
return text('OK')
request, response = app.test_client.get('/info')
assert response.status == 200
request, response = app.test_client.get('/whatever/you/set')
assert response.status == 200
def test_dynamic_route_unhashable(app):
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
async def handler(request, unhashable):
return text('OK')
request, response = app.test_client.get('/folder/test/asdf/end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test///////end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test/end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test/nope/')
assert response.status == 404
def test_websocket_route(app):
ev = asyncio.Event()
@app.websocket('/ws')
async def handler(request, ws):
assert ws.subprotocol is None
ev.set()
request, response = app.test_client.get('/ws', headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13'})
assert response.status == 101
assert ev.is_set()
def test_websocket_route_with_subprotocols(app):
results = []
@app.websocket('/ws', subprotocols=['foo', 'bar'])
async def handler(request, ws):
results.append(ws.subprotocol)
request, response = app.test_client.get('/ws', headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13',
'Sec-WebSocket-Protocol': 'bar'})
assert response.status == 101
request, response = app.test_client.get('/ws', headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13',
'Sec-WebSocket-Protocol': 'bar, foo'})
assert response.status == 101
request, response = app.test_client.get('/ws', headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13',
'Sec-WebSocket-Protocol': 'baz'})
assert response.status == 101
request, response = app.test_client.get('/ws', headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13'})
assert response.status == 101
assert results == ['bar', 'bar', None, None]
def test_route_duplicate(app):
with pytest.raises(RouteExists):
@app.route('/test')
async def handler1(request):
pass
@app.route('/test')
async def handler2(request):
pass
with pytest.raises(RouteExists):
@app.route('/test/<dynamic>/')
async def handler3(request, dynamic):
pass
@app.route('/test/<dynamic>/')
async def handler4(request, dynamic):
pass
def test_method_not_allowed(app):
@app.route('/test', methods=['GET'])
async def handler(request):
return text('OK')
request, response = app.test_client.get('/test')
assert response.status == 200
request, response = app.test_client.post('/test')
assert response.status == 405
def test_static_add_route(app):
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test')
app.add_route(handler2, '/test2')
request, response = app.test_client.get('/test')
assert response.text == 'OK1'
request, response = app.test_client.get('/test2')
assert response.text == 'OK2'
def test_dynamic_add_route(app):
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name>')
request, response = app.test_client.get('/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
def test_dynamic_add_route_string(app):
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name:string>')
request, response = app.test_client.get('/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
request, response = app.test_client.get('/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
def test_dynamic_add_route_int(app):
results = []
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
app.add_route(handler, '/folder/<folder_id:int>')
request, response = app.test_client.get('/folder/12345')
assert response.text == 'OK'
assert type(results[0]) is int
request, response = app.test_client.get('/folder/asdf')
assert response.status == 404
def test_dynamic_add_route_number(app):
results = []
async def handler(request, weight):
results.append(weight)
return text('OK')
app.add_route(handler, '/weight/<weight:number>')
request, response = app.test_client.get('/weight/12345')
assert response.text == 'OK'
assert type(results[0]) is float
request, response = app.test_client.get('/weight/1234.56')
assert response.status == 200
request, response = app.test_client.get('/weight/1234-56')
assert response.status == 404
def test_dynamic_add_route_regex(app):
async def handler(request, folder_id):
return text('OK')
app.add_route(handler, '/folder/<folder_id:[A-Za-z0-9]{0,4}>')
request, response = app.test_client.get('/folder/test')
assert response.status == 200
request, response = app.test_client.get('/folder/test1')
assert response.status == 404
request, response = app.test_client.get('/folder/test-123')
assert response.status == 404
request, response = app.test_client.get('/folder/')
assert response.status == 200
def test_dynamic_add_route_unhashable(app):
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = app.test_client.get('/folder/test/asdf/end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test///////end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test/end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test/nope/')
assert response.status == 404
def test_add_route_duplicate(app):
with pytest.raises(RouteExists):
async def handler1(request):
pass
async def handler2(request):
pass
app.add_route(handler1, '/test')
app.add_route(handler2, '/test')
with pytest.raises(RouteExists):
async def handler1(request, dynamic):
pass
async def handler2(request, dynamic):
pass
app.add_route(handler1, '/test/<dynamic>/')
app.add_route(handler2, '/test/<dynamic>/')
def test_add_route_method_not_allowed(app):
async def handler(request):
return text('OK')
app.add_route(handler, '/test', methods=['GET'])
request, response = app.test_client.get('/test')
assert response.status == 200
request, response = app.test_client.post('/test')
assert response.status == 405
def test_remove_static_route(app):
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test')
app.add_route(handler2, '/test2')
request, response = app.test_client.get('/test')
assert response.status == 200
request, response = app.test_client.get('/test2')
assert response.status == 200
app.remove_route('/test')
app.remove_route('/test2')
request, response = app.test_client.get('/test')
assert response.status == 404
request, response = app.test_client.get('/test2')
assert response.status == 404
def test_remove_dynamic_route(app):
async def handler(request, name):
return text('OK')
app.add_route(handler, '/folder/<name>')
request, response = app.test_client.get('/folder/test123')
assert response.status == 200
app.remove_route('/folder/<name>')
request, response = app.test_client.get('/folder/test123')
assert response.status == 404
def test_remove_inexistent_route(app):
with pytest.raises(RouteDoesNotExist):
app.remove_route('/test')
def test_removing_slash(app):
@app.get('/rest/<resource>')
def get(_):
pass
@app.post('/rest/<resource>')
def post(_):
pass
assert len(app.router.routes_all.keys()) == 2
def test_remove_unhashable_route(app):
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = app.test_client.get('/folder/test/asdf/end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test///////end/')
assert response.status == 200
request, response = app.test_client.get('/folder/test/end/')
assert response.status == 200
app.remove_route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = app.test_client.get('/folder/test/asdf/end/')
assert response.status == 404
request, response = app.test_client.get('/folder/test///////end/')
assert response.status == 404
request, response = app.test_client.get('/folder/test/end/')
assert response.status == 404
def test_remove_route_without_clean_cache(app):
async def handler(request):
return text('OK')
app.add_route(handler, '/test')
request, response = app.test_client.get('/test')
assert response.status == 200
app.remove_route('/test', clean_cache=True)
app.remove_route('/test/', clean_cache=True)
request, response = app.test_client.get('/test')
assert response.status == 404
app.add_route(handler, '/test')
request, response = app.test_client.get('/test')
assert response.status == 200
app.remove_route('/test', clean_cache=False)
request, response = app.test_client.get('/test')
assert response.status == 200
def test_overload_routes(app):
@app.route('/overload', methods=['GET'])
async def handler1(request):
return text('OK1')
@app.route('/overload', methods=['POST', 'PUT'])
async def handler2(request):
return text('OK2')
request, response = app.test_client.get('/overload')
assert response.text == 'OK1'
request, response = app.test_client.post('/overload')
assert response.text == 'OK2'
request, response = app.test_client.put('/overload')
assert response.text == 'OK2'
request, response = app.test_client.delete('/overload')
assert response.status == 405
with pytest.raises(RouteExists):
@app.route('/overload', methods=['PUT', 'DELETE'])
async def handler3(request):
return text('Duplicated')
def test_unmergeable_overload_routes(app):
@app.route('/overload_whole', methods=None)
async def handler1(request):
return text('OK1')
with pytest.raises(RouteExists):
@app.route('/overload_whole', methods=['POST', 'PUT'])
async def handler2(request):
return text('Duplicated')
request, response = app.test_client.get('/overload_whole')
assert response.text == 'OK1'
request, response = app.test_client.post('/overload_whole')
assert response.text == 'OK1'
@app.route('/overload_part', methods=['GET'])
async def handler3(request):
return text('OK1')
with pytest.raises(RouteExists):
@app.route('/overload_part')
async def handler4(request):
return text('Duplicated')
request, response = app.test_client.get('/overload_part')
assert response.text == 'OK1'
request, response = app.test_client.post('/overload_part')
assert response.status == 405
def test_unicode_routes(app):
@app.get('/你好')
def handler1(request):
return text('OK1')
request, response = app.test_client.get('/你好')
assert response.text == 'OK1'
@app.route('/overload/<param>', methods=['GET'])
async def handler2(request, param):
return text('OK2 ' + param)
request, response = app.test_client.get('/overload/你好')
assert response.text == 'OK2 你好'
def test_uri_with_different_method_and_different_params(app):
@app.route('/ads/<ad_id>', methods=['GET'])
async def ad_get(request, ad_id):
return json({'ad_id': ad_id})
@app.route('/ads/<action>', methods=['POST'])
async def ad_post(request, action):
return json({'action': action})
request, response = app.test_client.get('/ads/1234')
assert response.status == 200
assert response.json == {
'ad_id': '1234'
}
request, response = app.test_client.post('/ads/post')
assert response.status == 200
assert response.json == {
'action': 'post'
}
def test_route_raise_ParameterNameConflicts(app):
with pytest.raises(ParameterNameConflicts):
@app.get('/api/v1/<user>/<user>/')
def handler(request, user):
return text('OK')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment