From 06ea9477b95caed298bc8cadac94fa98fc97e8d2 Mon Sep 17 00:00:00 2001 From: John Turner Date: Fri, 26 Sep 2025 02:24:00 -0400 Subject: add tests for s3 storage backend --- tests/meson.build | 2 +- tests/test_sqlite_storage.py | 105 ---------------------------------- tests/test_storage.py | 133 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 134 insertions(+), 106 deletions(-) delete mode 100755 tests/test_sqlite_storage.py create mode 100755 tests/test_storage.py (limited to 'tests') diff --git a/tests/meson.build b/tests/meson.build index c15bbab..6e66e76 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -1,4 +1,4 @@ -tests = files('test_client_zen.py', 'test_server.py', 'test_sqlite_storage.py') +tests = files('test_client_zen.py', 'test_server.py', 'test_storage.py') foreach test : tests name = fs.stem(test) diff --git a/tests/test_sqlite_storage.py b/tests/test_sqlite_storage.py deleted file mode 100755 index 6a93561..0000000 --- a/tests/test_sqlite_storage.py +++ /dev/null @@ -1,105 +0,0 @@ -#!/usr/bin/env python3 - -import sys -import asyncio -import tempfile -import aiosqlite -import string -import random -from pypaste.server import Paste, Key, keygen -from pypaste.server.sqlite import Sqlite -from datetime import datetime -from pathlib import Path - - -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: Sqlite) -> 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: Sqlite) -> 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: Sqlite) -> 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: Sqlite) -> 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: - 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" - ")" - ) - ) - - storage = Sqlite(connection) - await storage.setup() - - await asyncio.gather( - test_insert_retrieve(storage), - test_delete(storage), - test_exists_but_not_in_our_table(storage), - ) - - return 0 - - -if __name__ == "__main__": - sys.exit(asyncio.run(main())) diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100755 index 0000000..e5f312c --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 + +import sys +import os +import asyncio +import tempfile +import aiosqlite +import string +import random +from pypaste.server import Paste, Key, 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())) -- cgit v1.2.3