granian/tests/test_asgi.py at 3f89c46def40b4e99952a17f3d7cfe5a963ef86b · emmett-framework/granian · GitHub

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
117
118
119
120
121
122
123
import os

import httpx
import pytest


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_scope(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.get(f'http://localhost:{port}/info?test=true')

    assert res.status_code == 200
    assert res.headers['content-type'] == 'application/json'

    data = res.json()
    assert data['asgi'] == {'version': '3.0', 'spec_version': '2.3'}
    assert data['type'] == 'http'
    assert data['http_version'] == '1.1'
    assert data['scheme'] == 'http'
    assert data['method'] == 'GET'
    assert data['path'] == '/info'
    assert data['query_string'] == 'test=true'
    assert data['headers']['host'] == f'localhost:{port}'
    assert 'http.response.pathsend' in data['extensions']
    assert data['state']['global'] == 'test'


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_scope_path_utf8_lossy(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.get(f'http://localhost:{port}/%c0')

    assert res.status_code == 200
    data = res.json()
    assert data['path'] == '/\ufffd'


@pytest.mark.asyncio
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_body(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.post(f'http://localhost:{port}/echo', content='test')

    assert res.status_code == 200
    assert res.text == 'test'


@pytest.mark.asyncio
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_body_large(asgi_server, runtime_mode):
    data = ''.join([f'{idx}test'.zfill(8) for idx in range(0, 5000)])
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.post(f'http://localhost:{port}/echo', content=data)

    assert res.status_code == 200
    assert res.text == data


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_app_error(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.get(f'http://localhost:{port}/err_app')

    assert res.status_code == 500


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_protocol_error(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.get(f'http://localhost:{port}/err_proto/type')
        assert res.status_code == 200
        assert res.text == 'Unsupported ASGI message'

        res = httpx.get(f'http://localhost:{port}/err_proto/flow')
        assert res.status_code == 200
        assert res.text == 'msg1'


@pytest.mark.asyncio
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_file(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False) as port:
        res = httpx.get(f'http://localhost:{port}/file')

    assert res.status_code == 200
    assert res.headers['content-type'] == 'image/png'
    assert res.headers['content-length'] == '95'


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_sniffio(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False, task_impl='rust') as port:
        res = httpx.get(f'http://localhost:{port}/sniffio')

    assert res.status_code == 200
    assert res.text == 'asyncio'


@pytest.mark.asyncio
@pytest.mark.skipif(bool(os.getenv('PGO_RUN')), reason='PGO build')
@pytest.mark.parametrize('runtime_mode', ['mt', 'st'])
async def test_timeout(asgi_server, runtime_mode):
    async with asgi_server(runtime_mode, ws=False, task_impl='rust') as port:
        res = httpx.get(f'http://localhost:{port}/timeout_n')

    assert res.status_code == 200
    assert res.text == 'ok'

    async with asgi_server(runtime_mode, ws=False, task_impl='rust') as port:
        res = httpx.get(f'http://localhost:{port}/timeout_w')

    assert res.status_code == 200
    assert res.text == 'timeout'