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
|
from __future__ import annotations
import asyncio
import pytest
from hypercorn.app_wrappers import ASGIWrapper
from hypercorn.asyncio.tcp_server import TCPServer
from hypercorn.asyncio.worker_context import WorkerContext
from hypercorn.config import Config
from .helpers import MemoryReader, MemoryWriter
from ..helpers import echo_framework
@pytest.mark.asyncio
async def test_completes_on_closed() -> None:
event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
server = TCPServer(
ASGIWrapper(echo_framework),
event_loop,
Config(),
WorkerContext(None),
{},
MemoryReader(), # type: ignore
MemoryWriter(), # type: ignore
)
server.reader.close() # type: ignore
await server.run()
# Key is that this line is reached, rather than the above line
# hanging.
@pytest.mark.asyncio
async def test_complets_on_half_close() -> None:
event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
server = TCPServer(
ASGIWrapper(echo_framework),
event_loop,
Config(),
WorkerContext(None),
{},
MemoryReader(), # type: ignore
MemoryWriter(), # type: ignore
)
task = event_loop.create_task(server.run())
await server.reader.send(b"GET / HTTP/1.1\r\nHost: hypercorn\r\n\r\n") # type: ignore
server.reader.close() # type: ignore
await task
data = await server.writer.receive() # type: ignore
assert (
data
== b"HTTP/1.1 200 \r\ncontent-length: 348\r\ndate: Thu, 01 Jan 1970 01:23:20 GMT\r\nserver: hypercorn-h11\r\n\r\n" # noqa: E501
)
|
