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
125
126
127
128
129
130
131
132
133
|
#!/usr/bin/env python3
import sys
import os
import asyncio
import tempfile
import aiosqlite
import string
import random
from pypaste.server import Paste, Storage, keygen
from pypaste.server.sqlite import Sqlite
from pypaste.server.s3 import S3
from datetime import datetime
from pathlib import Path
from typing import List
def truncate(path: Path) -> None:
with open(path, "w") as f:
f.truncate(0)
def generate_key() -> str:
chars = string.ascii_letters
return "".join(random.choice(chars) for _ in range(10))
async def test_exists_but_not_in_our_table(storage: Storage) -> None:
key = keygen(6)
await storage.connection.execute(
"insert into pastes values(?, ?, ?, ?, ?)",
(key.data, key.length, datetime.now().isoformat(), None, bytes()),
)
assert not await storage.exists(key)
async def test_exists(storage: Storage) -> None:
dt = datetime.now()
key = keygen(6)
await storage.insert(Paste(dt, "test", "hello world"), key)
assert await storage.exists(key)
assert not await storage.exists(keygen(6))
async def test_delete(storage: Storage) -> None:
dt = datetime.now()
key = keygen(6)
await storage.insert(Paste(dt, "test", "hello world"), key)
assert await storage.exists(key)
await storage.delete(key)
assert not await storage.exists(key)
async def test_insert_retrieve(storage: Storage) -> None:
dt = datetime.now()
key = keygen(6)
await storage.insert(Paste(dt, "test", "hello world"), key)
paste = await storage.retrieve(key)
assert paste is not None
assert paste.dt == dt
assert paste.syntax == "test"
assert paste.text == "hello world"
async def main() -> int:
stores: List[Storage] = []
with tempfile.TemporaryDirectory() as tmpdir:
f = Path(tmpdir) / "database"
truncate(f)
async with aiosqlite.connect(f) as connection:
await connection.execute(
(
"create table pastes("
"key blob,"
"key_length int,"
"datetime text,"
"size int,"
"syntax text"
")"
)
)
sqlite_storage = Sqlite(connection)
await sqlite_storage.setup()
stores.append(sqlite_storage)
try:
os.environ["PYPASTE_TEST_S3"]
test_s3 = True
except KeyError:
test_s3 = False
if test_s3:
s3_storage = S3(
connection,
os.environ["PYPASTE_TEST_ENDPOINT"],
os.environ["PYPASTE_TEST_REGION"],
os.environ["PYPASTE_TEST_BUCKET"],
os.environ["PYPASTE_TEST_ACCESS_KEY"],
os.environ["PYPASTE_TEST_SECRET_KEY"],
)
await s3_storage.setup()
stores.append(s3_storage)
for store in stores:
await asyncio.gather(
test_insert_retrieve(store),
test_insert_retrieve(store),
test_delete(store),
test_delete(store),
test_exists_but_not_in_our_table(store),
test_exists_but_not_in_our_table(store),
)
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
|
