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
|
#!/usr/bin/env python3
import sys
import asyncio
import tempfile
import string
import random
import aiohttp
from pathlib import Path
def truncate(file: Path) -> None:
with open(file, "w") as f:
f.truncate(0)
def generate_name() -> str:
characters = string.ascii_letters
return "".join(random.choice(characters) for _ in range(10))
async def main() -> int:
with tempfile.TemporaryDirectory() as tmpdir:
socket = Path(tmpdir) / (generate_name() + ".sock")
database = Path(tmpdir) / generate_name()
truncate(database)
proc = await asyncio.create_subprocess_exec(
"python",
"-m",
"pypaste.server",
"--socket",
socket,
"--site",
"http://dummy",
"--content-length-max-bytes",
"200000",
"--key-length",
"3",
"--database",
database,
"--storage-max-bytes",
"500000000",
"sqlite",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
assert proc.stdout is not None
line = await proc.stdout.readline()
if b"starting" not in line:
print(line, file=sys.stderr)
return 1
connection = aiohttp.UnixConnector(path=str(socket))
async with aiohttp.ClientSession(connector=connection) as client:
async with client.post("http://dummy/paste", data="hello pypaste") as post:
assert post.status == 200
data = await post.read()
url = data.decode()
async with client.get(url) as get:
assert get.status == 200
assert b"hello pypaste" in await get.read()
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
|
