Codebase list python-webargs / master tests / test_aiohttpparser.py
master

Tree @master (Download .tar.gz)

test_aiohttpparser.py @master — raw · history · blame 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
from io import BytesIO
from unittest import mock

import webtest
import webtest_aiohttp
import pytest

from webargs import fields
from webargs.aiohttpparser import AIOHTTPParser
from webargs.testing import CommonTestCase
from tests.apps.aiohttp_app import create_app


@pytest.fixture
def web_request():
    req = mock.Mock()
    req.query = {}
    yield req
    req.query = {}


class TestAIOHTTPParser(CommonTestCase):
    def create_app(self):
        return create_app()

    def create_testapp(self, app, loop):
        return webtest_aiohttp.TestApp(app, loop=loop)

    @pytest.fixture
    def testapp(self, loop):
        return self.create_testapp(self.create_app(), loop)

    @pytest.mark.skip(reason="files location not supported for aiohttpparser")
    def test_parse_files(self, testapp):
        pass

    def test_parse_match_info(self, testapp):
        assert testapp.get("/echo_match_info/42").json == {"mymatch": 42}

    def test_use_args_on_method_handler(self, testapp):
        assert testapp.get("/echo_method").json == {"name": "World"}
        assert testapp.get("/echo_method?name=Steve").json == {"name": "Steve"}
        assert testapp.get("/echo_method_view").json == {"name": "World"}
        assert testapp.get("/echo_method_view?name=Steve").json == {"name": "Steve"}

    # regression test for https://github.com/marshmallow-code/webargs/issues/165
    def test_multiple_args(self, testapp):
        res = testapp.post_json("/echo_multiple_args", {"first": "1", "last": "2"})
        assert res.json == {"first": "1", "last": "2"}

    # regression test for https://github.com/marshmallow-code/webargs/issues/145
    def test_nested_many_with_data_key(self, testapp):
        res = testapp.post_json("/echo_nested_many_data_key", {"X-Field": [{"id": 24}]})
        assert res.json == {"x_field": [{"id": 24}]}

        res = testapp.post_json("/echo_nested_many_data_key", {})
        assert res.json == {}

    def test_schema_as_kwargs_view(self, testapp):
        assert testapp.get("/echo_use_schema_as_kwargs").json == {"name": "World"}
        assert testapp.get("/echo_use_schema_as_kwargs?name=Chandler").json == {
            "name": "Chandler"
        }

    # https://github.com/marshmallow-code/webargs/pull/297
    def test_empty_json_body(self, testapp):
        environ = {"CONTENT_TYPE": "application/json", "wsgi.input": BytesIO(b"")}
        req = webtest.TestRequest.blank("/echo", environ)
        resp = testapp.do_request(req)
        assert resp.json == {"name": "World"}

    def test_use_args_multiple(self, testapp):
        res = testapp.post_json(
            "/echo_use_args_multiple?page=2&q=10", {"name": "Steve"}
        )
        assert res.json == {
            "query_parsed": {"page": 2, "q": 10},
            "json_parsed": {"name": "Steve"},
        }


async def test_aiohttpparser_synchronous_error_handler(web_request):
    parser = AIOHTTPParser()

    class CustomError(Exception):
        pass

    @parser.error_handler
    def custom_handle_error(error, req, schema, *, error_status_code, error_headers):
        raise CustomError("foo")

    with pytest.raises(CustomError):
        await parser.parse(
            {"foo": fields.Int(required=True)}, web_request, location="query"
        )


async def test_aiohttpparser_asynchronous_error_handler(web_request):
    parser = AIOHTTPParser()

    class CustomError(Exception):
        pass

    @parser.error_handler
    async def custom_handle_error(
        error, req, schema, *, error_status_code, error_headers
    ):
        async def inner():
            raise CustomError("foo")

        await inner()

    with pytest.raises(CustomError):
        await parser.parse(
            {"foo": fields.Int(required=True)}, web_request, location="query"
        )