add test_sqlite_storage.py

2 files changed, 104 insertions, 1 deletions

diff --git a/tests/meson.build b/tests/meson.build
index e39a559..801b30d 100644
--- a/tests/meson.build
+++ b/tests/meson.build
@@ -1,4 +1,4 @@
-tests = files('test_server.py')
+tests = files('test_server.py', 'test_sqlite_storage.py')
 
 foreach test : tests
     name = fs.stem(test)
diff --git a/tests/test_sqlite_storage.py b/tests/test_sqlite_storage.py
new file mode 100755
index 0000000..000d1a7
--- /dev/null
+++ b/tests/test_sqlite_storage.py
@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+
+import sys
+import asyncio
+import tempfile
+import aiosqlite
+import string
+import random
+from pypaste.server import Paste
+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 = generate_key()
+
+    await storage.connection.execute(
+        "insert into pastes values(?, ?, ?, ?)",
+        (key, datetime.now().isoformat(), None, bytes()),
+    )
+
+    assert not await storage.exists(key)
+
+
+async def test_exists(storage: Sqlite) -> None:
+    dt = datetime.now()
+    key = generate_key()
+
+    await storage.insert(Paste(key, dt, "test", "hello world"))
+
+    assert await storage.exists(key)
+
+
+async def test_delete(storage: Sqlite) -> None:
+    dt = datetime.now()
+    key = generate_key()
+
+    await storage.insert(Paste(key, dt, "test", "hello world"))
+
+    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 = generate_key()
+
+    await storage.insert(Paste(key, dt, "test", "hello world"))
+
+    paste = await storage.retrieve(key)
+
+    assert paste is not None
+    assert paste.key == key
+    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 text,"
+                    "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()))