Add tests (#1430)

tests/test_app.py

@@ -0,0 +1,82 @@
+import asyncio
+
+import pytest
+
+from sanic.exceptions import SanicException
+from sanic.response import text
+
+
+def test_app_loop_running(app):
+
+    @app.get('/test')
+    async def handler(request):
+        assert isinstance(app.loop, asyncio.AbstractEventLoop)
+        return text('pass')
+
+    request, response = app.test_client.get('/test')
+    assert response.text == 'pass'
+
+
+def test_app_loop_not_running(app):
+    with pytest.raises(SanicException) as excinfo:
+        app.loop
+
+    assert str(excinfo.value) == (
+        'Loop can only be retrieved after the app has started '
+        'running. Not supported with `create_server` function'
+    )
+
+
+def test_app_run_raise_type_error(app):
+
+    with pytest.raises(TypeError) as excinfo:
+        app.run(loop='loop')
+
+    assert str(excinfo.value) == (
+        'loop is not a valid argument. To use an existing loop, '
+        'change to create_server().\nSee more: '
+        'https://sanic.readthedocs.io/en/latest/sanic/deploying.html'
+        '#asynchronous-support'
+    )
+
+
+def test_app_route_raise_value_error(app):
+
+    with pytest.raises(ValueError) as excinfo:
+        @app.route('/test')
+        async def handler():
+            return text('test')
+
+    assert str(excinfo.value) == 'Required parameter `request` missing in the handler() route?'
+
+
+def test_app_handle_request_handler_is_none(app, monkeypatch):
+
+    def mockreturn(*args, **kwargs):
+        return None, [], {}, ''
+
+    # Not sure how to make app.router.get() return None, so use mock here.
+    monkeypatch.setattr(app.router, 'get', mockreturn)
+
+    @app.get('/test')
+    def handler(request):
+        return text('test')
+
+    request, response = app.test_client.get('/test')
+
+    assert response.text == 'Error: \'None\' was returned while requesting a handler from the router'
+
+
+@pytest.mark.parametrize('websocket_enabled', [True, False])
+@pytest.mark.parametrize('enable', [True, False])
+def test_enable_websocket(app, websocket_enabled, enable):
+    app.websocket_enabled = websocket_enabled
+    app.enable_websocket(enable=enable)
+
+    assert app.websocket_enabled == enable
+
+    @app.websocket('/ws')
+    async def handler(request, ws):
+        await ws.send('test')
+
+    assert app.websocket_enabled == True

tests/test_blueprints.py

@@ -673,3 +673,34 @@ def test_websocket_route(app: Sanic):
     })
     assert response.status == 101
     assert event.is_set()
+
+
+def test_duplicate_blueprint(app):
+    bp_name = 'bp'
+    bp = Blueprint(bp_name)
+    bp1 = Blueprint(bp_name)
+
+    app.blueprint(bp)
+
+    with pytest.raises(AssertionError) as excinfo:
+        app.blueprint(bp1)
+
+    assert str(excinfo.value) == (
+        'A blueprint with the name "{}" is already registered.  '
+        'Blueprint names must be unique.'
+    ).format(bp_name)
+
+
+@pytest.mark.parametrize('debug', [True, False, None])
+def test_register_blueprint(app, debug):
+    bp = Blueprint('bp')
+
+    app.debug = debug
+    with pytest.warns(DeprecationWarning) as record:
+        app.register_blueprint(bp)
+    
+    assert record[0].message.args[0] == (
+        "Use of register_blueprint will be deprecated in "
+        "version 1.0.  Please use the blueprint method"
+        " instead"
+    )

tests/test_exceptions.py

@@ -66,6 +66,10 @@ def exception_app():
         abort(500)
         return text("OK")
 
+    @app.route('/abort/message')
+    def handler_abort_message(request):
+        abort(500, message='Abort')
+
     @app.route('/divide_by_zero')
     def handle_unhandled_exception(request):
         1 / 0
@@ -201,3 +205,7 @@ def test_abort(exception_app):
 
     request, response = exception_app.test_client.get('/abort')
     assert response.status == 500
+
+    request, response = exception_app.test_client.get('/abort/message')
+    assert response.status == 500
+    assert response.text == 'Error: Abort'

tests/test_requests.py

@@ -340,7 +340,7 @@ def test_url_attributes_no_ssl(app, path, query, expected_url):
         ('/bar/baz', '', 'https://{}:{}/bar/baz'),
         ('/moo/boo', 'arg1=val1', 'https://{}:{}/moo/boo?arg1=val1')
     ])
-def test_url_attributes_with_ssl(app, path, query, expected_url):
+def test_url_attributes_with_ssl_context(app, path, query, expected_url):
     current_dir = os.path.dirname(os.path.realpath(__file__))
     context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
     context.load_cert_chain(
@@ -365,6 +365,58 @@ def test_url_attributes_with_ssl(app, path, query, expected_url):
     assert parsed.netloc == request.host
 
 
+@pytest.mark.parametrize(
+    'path,query,expected_url', [
+        ('/foo', '', 'https://{}:{}/foo'),
+        ('/bar/baz', '', 'https://{}:{}/bar/baz'),
+        ('/moo/boo', 'arg1=val1', 'https://{}:{}/moo/boo?arg1=val1')
+    ])
+def test_url_attributes_with_ssl_dict(app, path, query, expected_url):
+
+    current_dir = os.path.dirname(os.path.realpath(__file__))
+    ssl_cert = os.path.join(current_dir, 'certs/selfsigned.cert')
+    ssl_key = os.path.join(current_dir, 'certs/selfsigned.key')
+
+    ssl_dict = {
+        'cert': ssl_cert,
+        'key': ssl_key
+    }
+
+    async def handler(request):
+        return text('OK')
+
+    app.add_route(handler, path)
+
+    request, response = app.test_client.get(
+        'https://{}:{}'.format(HOST, PORT) + path + '?{}'.format(query),
+        server_kwargs={'ssl': ssl_dict})
+    assert request.url == expected_url.format(HOST, PORT)
+
+    parsed = urlparse(request.url)
+
+    assert parsed.scheme == request.scheme
+    assert parsed.path == request.path
+    assert parsed.query == request.query_string
+    assert parsed.netloc == request.host
+
+
+def test_invalid_ssl_dict(app):
+
+    @app.get('/test')
+    async def handler(request):
+        return text('ssl test')
+
+    ssl_dict = {
+        'cert': None,
+        'key': None
+    }
+
+    with pytest.raises(ValueError) as excinfo:
+        request, response = app.test_client.get('/test', server_kwargs={'ssl': ssl_dict})
+
+    assert str(excinfo.value) == 'SSLContext or certificate and key required.'
+
+
 def test_form_with_multiple_values(app):
 
     @app.route('/', methods=['POST'])

tests/test_response.py

@@ -1,6 +1,7 @@
 import asyncio
 import inspect
 import os
+from collections import namedtuple
 from mimetypes import guess_type
 from random import choice
 from unittest.mock import MagicMock
@@ -9,9 +10,8 @@ from urllib.parse import unquote
 import pytest
 from aiofiles import os as async_os
 
-from sanic.response import (
+from sanic.response import (HTTPResponse, StreamingHTTPResponse, file,
-    HTTPResponse, stream, StreamingHTTPResponse, file, file_stream, json
+                            file_stream, json, raw, stream, text)
-)
 from sanic.server import HttpProtocol
 from sanic.testing import HOST, PORT
 
@@ -422,3 +422,41 @@ def test_file_stream_head_response(app, file_name, static_file_directory):
     assert int(response.headers[
                'Content-Length']) == len(
                    get_file_content(static_file_directory, file_name))
+
+
+@pytest.mark.parametrize('file_name',
+                         ['test.file', 'decode me.txt', 'python.png'])
+@pytest.mark.parametrize('size,start,end', [
+    (1024, 0, 1024),
+    (4096, 1024, 8192),
+    ])
+def test_file_stream_response_range(app, file_name, static_file_directory, size, start, end):
+
+    Range = namedtuple('Range', ['size', 'start', 'end', 'total'])
+    total = len(get_file_content(static_file_directory, file_name))
+    range = Range(size=size, start=start, end=end, total=total)
+
+    @app.route('/files/<filename>', methods=['GET'])
+    def file_route(request, filename):
+        file_path = os.path.join(static_file_directory, filename)
+        file_path = os.path.abspath(unquote(file_path))
+        return file_stream(
+            file_path,
+            chunk_size=32,
+            mime_type=guess_type(file_path)[0] or 'text/plain',
+            _range=range)
+
+    request, response = app.test_client.get('/files/{}'.format(file_name))
+    assert response.status == 206
+    assert 'Content-Range' in response.headers
+    assert response.headers['Content-Range'] == 'bytes {}-{}/{}'.format(range.start, range.end, range.total)
+
+def test_raw_response(app):
+
+    @app.get('/test')
+    def handler(request):
+        return raw(b'raw_response')
+
+    request, response = app.test_client.get('/test')
+    assert response.content_type == 'application/octet-stream'
+    assert response.body == b'raw_response'

tests/test_routes.py

@@ -1,11 +1,11 @@
 import asyncio
+
 import pytest
 
 from sanic import Sanic
-from sanic.response import text, json
-from sanic.router import RouteExists, RouteDoesNotExist, ParameterNameConflicts
 from sanic.constants import HTTP_METHODS
-
+from sanic.response import json, text
+from sanic.router import ParameterNameConflicts, RouteDoesNotExist, RouteExists
 
 # ------------------------------------------------------------ #
 #  UTF-8
@@ -462,10 +462,11 @@ def test_dynamic_route_unhashable(app):
     assert response.status == 404
 
 
-def test_websocket_route(app):
+@pytest.mark.parametrize('url', ['/ws', 'ws'])
+def test_websocket_route(app, url):
     ev = asyncio.Event()
 
-    @app.websocket('/ws')
+    @app.websocket(url)
     async def handler(request, ws):
         assert ws.subprotocol is None
         ev.set()
@@ -520,6 +521,24 @@ def test_websocket_route_with_subprotocols(app):
     assert results == ['bar', 'bar', None, None]
 
 
+@pytest.mark.parametrize('strict_slashes', [True, False, None])
+def test_add_webscoket_route(app, strict_slashes):
+    ev = asyncio.Event()
+
+    async def handler(request, ws):
+        assert ws.subprotocol is None
+        ev.set()
+
+    app.add_websocket_route(handler, '/ws', strict_slashes=strict_slashes)
+    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_route_duplicate(app):
 
     with pytest.raises(RouteExists):
@@ -554,7 +573,8 @@ def test_method_not_allowed(app):
     assert response.status == 405
 
 
-def test_static_add_route(app):
+@pytest.mark.parametrize('strict_slashes', [True, False, None])
+def test_static_add_route(app, strict_slashes):
 
     async def handler1(request):
         return text('OK1')
@@ -562,8 +582,8 @@ def test_static_add_route(app):
     async def handler2(request):
         return text('OK2')
 
-    app.add_route(handler1, '/test')
+    app.add_route(handler1, '/test', strict_slashes=strict_slashes)
-    app.add_route(handler2, '/test2')
+    app.add_route(handler2, '/test2', strict_slashes=strict_slashes)
 
     request, response = app.test_client.get('/test')
     assert response.text == 'OK1'

tests/test_url_building.py

@@ -68,6 +68,25 @@ def test_simple_url_for_getting_with_more_params(app, args, url):
     assert response.text == 'this should pass'
 
 
+def test_url_for_with_server_name(app):
+
+    server_name = '{}:{}'.format(test_host, test_port)
+    app.config.update({
+        'SERVER_NAME': server_name
+    })
+    path = '/myurl'
+
+    @app.route(path)
+    def passes(request):
+        return text('this should pass')
+
+    url = 'http://{}{}'.format(server_name, path)
+    assert url == app.url_for('passes', _server=None, _external=True)
+    request, response = app.test_client.get(url)
+    assert response.status == 200
+    assert response.text == 'this should pass'
+
+
 def test_fails_if_endpoint_not_found(app):
 
     @app.route('/fail')

tests/test_views.py

@@ -232,10 +232,16 @@ def test_composition_view_runs_methods_as_expected(app, method):
         request, response = getattr(app.test_client, method.lower())('/')
         assert response.text == 'first method'
 
+        response = view(request)
+        assert response.body.decode() == 'first method'
+
     if method in ['DELETE', 'PATCH']:
         request, response = getattr(app.test_client, method.lower())('/')
         assert response.text == 'second method'
 
+        response = view(request)
+        assert response.body.decode() == 'second method'
+
 
 @pytest.mark.parametrize('method', HTTP_METHODS)
 def test_composition_view_rejects_invalid_methods(app, method):