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
124
|
from __future__ import annotations
import asyncio
from unittest.mock import call, Mock
import pytest
from h2.connection import H2Connection
from h2.events import ConnectionTerminated
from hypercorn.asyncio.worker_context import EventWrapper, WorkerContext
from hypercorn.config import Config
from hypercorn.events import Closed, RawData
from hypercorn.protocol.h2 import BUFFER_HIGH_WATER, BufferCompleteError, H2Protocol, StreamBuffer
from hypercorn.typing import ConnectionState
try:
from unittest.mock import AsyncMock
except ImportError:
# Python < 3.8
from mock import AsyncMock # type: ignore
@pytest.mark.asyncio
async def test_stream_buffer_push_and_pop() -> None:
event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
stream_buffer = StreamBuffer(EventWrapper)
async def _push_over_limit() -> bool:
await stream_buffer.push(b"a" * (BUFFER_HIGH_WATER + 1))
return True
task = event_loop.create_task(_push_over_limit())
assert not task.done() # Blocked as over high water
await stream_buffer.pop(BUFFER_HIGH_WATER // 4)
assert not task.done() # Blocked as over low water
await stream_buffer.pop(BUFFER_HIGH_WATER // 4)
assert (await task) is True
@pytest.mark.asyncio
async def test_stream_buffer_drain() -> None:
event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
stream_buffer = StreamBuffer(EventWrapper)
await stream_buffer.push(b"a" * 10)
async def _drain() -> bool:
await stream_buffer.drain()
return True
task = event_loop.create_task(_drain())
assert not task.done() # Blocked
await stream_buffer.pop(20)
assert (await task) is True
@pytest.mark.asyncio
async def test_stream_buffer_closed() -> None:
stream_buffer = StreamBuffer(EventWrapper)
await stream_buffer.close()
await stream_buffer._is_empty.wait()
await stream_buffer._paused.wait()
assert True
with pytest.raises(BufferCompleteError):
await stream_buffer.push(b"a")
@pytest.mark.asyncio
async def test_stream_buffer_complete() -> None:
stream_buffer = StreamBuffer(EventWrapper)
await stream_buffer.push(b"a" * 10)
assert not stream_buffer.complete
stream_buffer.set_complete()
assert not stream_buffer.complete
await stream_buffer.pop(20)
assert stream_buffer.complete
@pytest.mark.asyncio
async def test_protocol_handle_protocol_error() -> None:
protocol = H2Protocol(
Mock(),
Config(),
WorkerContext(None),
AsyncMock(),
ConnectionState({}),
False,
None,
None,
AsyncMock(),
)
await protocol.handle(RawData(data=b"broken nonsense\r\n\r\n"))
protocol.send.assert_awaited() # type: ignore
assert protocol.send.call_args_list == [call(Closed())] # type: ignore
@pytest.mark.asyncio
async def test_protocol_keep_alive_max_requests() -> None:
protocol = H2Protocol(
Mock(),
Config(),
WorkerContext(None),
AsyncMock(),
ConnectionState({}),
False,
None,
None,
AsyncMock(),
)
protocol.config.keep_alive_max_requests = 0
client = H2Connection()
client.initiate_connection()
headers = [
(":method", "GET"),
(":path", "/reqinfo"),
(":authority", "hypercorn"),
(":scheme", "https"),
]
client.send_headers(1, headers, end_stream=True)
await protocol.handle(RawData(data=client.data_to_send()))
protocol.send.assert_awaited() # type: ignore
events = client.receive_data(protocol.send.call_args_list[1].args[0].data) # type: ignore
assert isinstance(events[-1], ConnectionTerminated)
|
