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
|
"""Simple tests verifying basic functionality."""
import asyncio
from aiofiles import threadpool
async def test_serve_small_bin_file_sync(tmpdir, unused_tcp_port):
"""Fire up a small simple file server, and fetch a file.
The file is read into memory synchronously, so this test doesn't actually
test anything except the general test concept.
"""
# First we'll write a small file.
filename = "test.bin"
file_content = b"0123456789"
file = tmpdir.join(filename)
file.write_binary(file_content)
async def serve_file(reader, writer):
full_filename = str(file)
with open(full_filename, "rb") as f:
writer.write(f.read())
writer.close()
server = await asyncio.start_server(serve_file, port=unused_tcp_port)
reader, _ = await asyncio.open_connection(host="localhost", port=unused_tcp_port)
payload = await reader.read()
assert payload == file_content
server.close()
await server.wait_closed()
async def test_serve_small_bin_file(tmpdir, unused_tcp_port):
"""Fire up a small simple file server, and fetch a file."""
# First we'll write a small file.
filename = "test.bin"
file_content = b"0123456789"
file = tmpdir.join(filename)
file.write_binary(file_content)
async def serve_file(reader, writer):
full_filename = str(file)
f = await threadpool.open(full_filename, mode="rb")
writer.write(await f.read())
await f.close()
writer.close()
server = await asyncio.start_server(serve_file, port=unused_tcp_port)
reader, _ = await asyncio.open_connection(host="localhost", port=unused_tcp_port)
payload = await reader.read()
assert payload == file_content
server.close()
await server.wait_closed()
|
