diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/aiosqlite/tests')
5 files changed, 695 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/aiosqlite/tests/__init__.py b/.venv/lib/python3.12/site-packages/aiosqlite/tests/__init__.py new file mode 100644 index 00000000..4b173f68 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/aiosqlite/tests/__init__.py @@ -0,0 +1,4 @@ +# Copyright 2022 Amethyst Reese +# Licensed under the MIT license + +from .smoke import SmokeTest diff --git a/.venv/lib/python3.12/site-packages/aiosqlite/tests/__main__.py b/.venv/lib/python3.12/site-packages/aiosqlite/tests/__main__.py new file mode 100644 index 00000000..648131e7 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/aiosqlite/tests/__main__.py @@ -0,0 +1,7 @@ +# Copyright 2022 Amethyst Reese +# Licensed under the MIT license + +import unittest + +if __name__ == "__main__": + unittest.main(module="aiosqlite.tests", verbosity=2) diff --git a/.venv/lib/python3.12/site-packages/aiosqlite/tests/helpers.py b/.venv/lib/python3.12/site-packages/aiosqlite/tests/helpers.py new file mode 100644 index 00000000..f7b53fe2 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/aiosqlite/tests/helpers.py @@ -0,0 +1,29 @@ +# Copyright 2022 Amethyst Reese +# Licensed under the MIT license + +import logging +import sys + + +def setup_logger(): + log = logging.getLogger("") + log.setLevel(logging.INFO) + + logging.addLevelName(logging.ERROR, "E") + logging.addLevelName(logging.WARNING, "W") + logging.addLevelName(logging.INFO, "I") + logging.addLevelName(logging.DEBUG, "V") + + date_fmt = r"%H:%M:%S" + verbose_fmt = ( + "%(asctime)s,%(msecs)d %(levelname)s " + "%(module)s:%(funcName)s():%(lineno)d " + "%(message)s" + ) + + handler = logging.StreamHandler(sys.stdout) + handler.setLevel(logging.INFO) + handler.setFormatter(logging.Formatter(verbose_fmt, date_fmt)) + log.addHandler(handler) + + return log diff --git a/.venv/lib/python3.12/site-packages/aiosqlite/tests/perf.py b/.venv/lib/python3.12/site-packages/aiosqlite/tests/perf.py new file mode 100644 index 00000000..08f03359 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/aiosqlite/tests/perf.py @@ -0,0 +1,203 @@ +# Copyright 2022 Amethyst Reese +# Licensed under the MIT license + +""" +Simple perf tests for aiosqlite and the asyncio run loop. +""" +import string +import tempfile +import time + +from unittest import IsolatedAsyncioTestCase as TestCase + +import aiosqlite +from .smoke import setup_logger + +TEST_DB = ":memory:" +TARGET = 2.0 +RESULTS = {} + + +def timed(fn, name=None): + """ + Decorator for perf testing a block of async code. + + Expects the wrapped function to return an async generator. + The generator should do setup, then yield when ready to start perf testing. + The decorator will then pump the generator repeatedly until the target + time has been reached, then close the generator and print perf results. + """ + + name = name or fn.__name__ + + async def wrapper(*args, **kwargs): + gen = fn(*args, **kwargs) + + await gen.asend(None) + count = 0 + before = time.time() + + while True: + count += 1 + value = time.time() - before < TARGET + try: + if value: + await gen.asend(value) + else: + await gen.aclose() + break + + except StopAsyncIteration: + break + + except Exception as e: + print(f"exception occurred: {e}") + return + + duration = time.time() - before + + RESULTS[name] = (count, duration) + + return wrapper + + +class PerfTest(TestCase): + @classmethod + def setUpClass(cls): + print(f"Running perf tests for at least {TARGET:.1f}s each...") + setup_logger() + + @classmethod + def tearDownClass(cls): + print(f"\n{'Perf Test':<25} Iterations Duration {'Rate':>11}") + for name in sorted(RESULTS): + count, duration = RESULTS[name] + rate = count / duration + name = name.replace("test_", "") + print(f"{name:<25} {count:>10} {duration:>7.1f}s {rate:>9.1f}/s") + + @timed + async def test_connection_memory(self): + while True: + yield + async with aiosqlite.connect(TEST_DB): + pass + + @timed + async def test_connection_file(self): + with tempfile.NamedTemporaryFile(delete=False) as tf: + path = tf.name + tf.close() + + async with aiosqlite.connect(path) as db: + await db.execute( + "create table perf (i integer primary key asc, k integer)" + ) + await db.execute("insert into perf (k) values (2), (3)") + await db.commit() + + while True: + yield + async with aiosqlite.connect(path): + pass + + @timed + async def test_atomics(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + await db.execute("insert into perf (k) values (2), (3)") + await db.commit() + + while True: + yield + async with db.execute("select last_insert_rowid()") as cursor: + await cursor.fetchone() + + @timed + async def test_inserts(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + await db.commit() + + while True: + yield + await db.execute("insert into perf (k) values (1), (2), (3)") + await db.commit() + + @timed + async def test_insert_ids(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + await db.commit() + + while True: + yield + cursor = await db.execute("insert into perf (k) values (1)") + await cursor.execute("select last_insert_rowid()") + await cursor.fetchone() + await db.commit() + + @timed + async def test_insert_macro_ids(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + await db.commit() + + while True: + yield + await db.execute_insert("insert into perf (k) values (1)") + await db.commit() + + @timed + async def test_select(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + for i in range(100): + await db.execute("insert into perf (k) values (%d)" % (i,)) + await db.commit() + + while True: + yield + cursor = await db.execute("select i, k from perf") + assert len(await cursor.fetchall()) == 100 + + @timed + async def test_select_macro(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table perf (i integer primary key asc, k integer)") + for i in range(100): + await db.execute("insert into perf (k) values (%d)" % (i,)) + await db.commit() + + while True: + yield + assert len(await db.execute_fetchall("select i, k from perf")) == 100 + + async def test_iterable_cursor_perf(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute( + "create table ic_perf (" + "i integer primary key asc, k integer, a integer, b integer, c char(16))" + ) + for batch in range(128): # add 128k rows + r_start = batch * 1024 + await db.executemany( + "insert into ic_perf (k, a, b, c) values(?, 1, 2, ?)", + [ + *[ + (i, string.ascii_lowercase) + for i in range(r_start, r_start + 1024) + ] + ], + ) + await db.commit() + + async def test_perf(chunk_size: int): + while True: + async with db.execute("SELECT * FROM ic_perf") as cursor: + cursor.iter_chunk_size = chunk_size + async for _ in cursor: + yield + + for chunk_size in [2**i for i in range(4, 11)]: + await timed(test_perf, f"iterable_cursor @ {chunk_size}")(chunk_size) diff --git a/.venv/lib/python3.12/site-packages/aiosqlite/tests/smoke.py b/.venv/lib/python3.12/site-packages/aiosqlite/tests/smoke.py new file mode 100644 index 00000000..f42106c7 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/aiosqlite/tests/smoke.py @@ -0,0 +1,452 @@ +# Copyright 2022 Amethyst Reese +# Licensed under the MIT license +import asyncio +import sqlite3 +from pathlib import Path +from sqlite3 import OperationalError +from threading import Thread +from unittest import IsolatedAsyncioTestCase as TestCase, SkipTest + +import aiosqlite +from .helpers import setup_logger + +TEST_DB = Path("test.db") + +# pypy uses non-standard text factory for low-level sqlite implementation +try: + from _sqlite3 import _unicode_text_factory as default_text_factory +except ImportError: + default_text_factory = str + + +class SmokeTest(TestCase): + @classmethod + def setUpClass(cls): + setup_logger() + + def setUp(self): + if TEST_DB.exists(): + TEST_DB.unlink() + + def tearDown(self): + if TEST_DB.exists(): + TEST_DB.unlink() + + async def test_connection_await(self): + db = await aiosqlite.connect(TEST_DB) + self.assertIsInstance(db, aiosqlite.Connection) + + async with db.execute("select 1, 2") as cursor: + rows = await cursor.fetchall() + self.assertEqual(rows, [(1, 2)]) + + await db.close() + + async def test_connection_context(self): + async with aiosqlite.connect(TEST_DB) as db: + self.assertIsInstance(db, aiosqlite.Connection) + + async with db.execute("select 1, 2") as cursor: + rows = await cursor.fetchall() + self.assertEqual(rows, [(1, 2)]) + + async def test_connection_locations(self): + class Fake: # pylint: disable=too-few-public-methods + def __str__(self): + return str(TEST_DB) + + locs = ("test.db", b"test.db", Path("test.db"), Fake()) + + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("create table foo (i integer, k integer)") + await db.execute("insert into foo (i, k) values (1, 5)") + await db.commit() + + cursor = await db.execute("select * from foo") + rows = await cursor.fetchall() + + for loc in locs: + async with aiosqlite.connect(loc) as db: + cursor = await db.execute("select * from foo") + self.assertEqual(await cursor.fetchall(), rows) + + async def test_multiple_connections(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute( + "create table multiple_connections " + "(i integer primary key asc, k integer)" + ) + + async def do_one_conn(i): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute("insert into multiple_connections (k) values (?)", [i]) + await db.commit() + + await asyncio.gather(*[do_one_conn(i) for i in range(10)]) + + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.execute("select * from multiple_connections") + rows = await cursor.fetchall() + + assert len(rows) == 10 + + async def test_multiple_queries(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute( + "create table multiple_queries " + "(i integer primary key asc, k integer)" + ) + + await asyncio.gather( + *[ + db.execute("insert into multiple_queries (k) values (?)", [i]) + for i in range(10) + ] + ) + + await db.commit() + + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.execute("select * from multiple_queries") + rows = await cursor.fetchall() + + assert len(rows) == 10 + + async def test_iterable_cursor(self): + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.cursor() + await cursor.execute( + "create table iterable_cursor " "(i integer primary key asc, k integer)" + ) + await cursor.executemany( + "insert into iterable_cursor (k) values (?)", [[i] for i in range(10)] + ) + await db.commit() + + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.execute("select * from iterable_cursor") + rows = [] + async for row in cursor: + rows.append(row) + + assert len(rows) == 10 + + async def test_multi_loop_usage(self): + results = {} + + def runner(k, conn): + async def query(): + async with conn.execute("select * from foo") as cursor: + rows = await cursor.fetchall() + self.assertEqual(len(rows), 2) + return rows + + with self.subTest(k): + loop = asyncio.new_event_loop() + rows = loop.run_until_complete(query()) + loop.close() + results[k] = rows + + async with aiosqlite.connect(":memory:") as db: + await db.execute("create table foo (id int, name varchar)") + await db.execute( + "insert into foo values (?, ?), (?, ?)", (1, "Sally", 2, "Janet") + ) + await db.commit() + + threads = [Thread(target=runner, args=(k, db)) for k in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(len(results), 4) + for rows in results.values(): + self.assertEqual(len(rows), 2) + + async def test_context_cursor(self): + async with aiosqlite.connect(TEST_DB) as db: + async with db.cursor() as cursor: + await cursor.execute( + "create table context_cursor " + "(i integer primary key asc, k integer)" + ) + await cursor.executemany( + "insert into context_cursor (k) values (?)", + [[i] for i in range(10)], + ) + await db.commit() + + async with aiosqlite.connect(TEST_DB) as db: + async with db.execute("select * from context_cursor") as cursor: + rows = [] + async for row in cursor: + rows.append(row) + + assert len(rows) == 10 + + async def test_cursor_return_self(self): + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.cursor() + + result = await cursor.execute( + "create table test_cursor_return_self (i integer, k integer)" + ) + self.assertEqual(result, cursor, "cursor execute returns itself") + + result = await cursor.executemany( + "insert into test_cursor_return_self values (?, ?)", [(1, 1), (2, 2)] + ) + self.assertEqual(result, cursor) + + result = await cursor.executescript( + "insert into test_cursor_return_self values (3, 3);" + "insert into test_cursor_return_self values (4, 4);" + "insert into test_cursor_return_self values (5, 5);" + ) + self.assertEqual(result, cursor) + + async def test_connection_properties(self): + async with aiosqlite.connect(TEST_DB) as db: + self.assertEqual(db.total_changes, 0) + + async with db.cursor() as cursor: + self.assertFalse(db.in_transaction) + await cursor.execute( + "create table test_properties " + "(i integer primary key asc, k integer, d text)" + ) + await cursor.execute( + "insert into test_properties (k, d) values (1, 'hi')" + ) + self.assertTrue(db.in_transaction) + await db.commit() + self.assertFalse(db.in_transaction) + + self.assertEqual(db.total_changes, 1) + + self.assertIsNone(db.row_factory) + self.assertEqual(db.text_factory, default_text_factory) + + async with db.cursor() as cursor: + await cursor.execute("select * from test_properties") + row = await cursor.fetchone() + self.assertIsInstance(row, tuple) + self.assertEqual(row, (1, 1, "hi")) + with self.assertRaises(TypeError): + _ = row["k"] + + async with db.cursor() as cursor: + cursor.row_factory = aiosqlite.Row + self.assertEqual(cursor.row_factory, aiosqlite.Row) + await cursor.execute("select * from test_properties") + row = await cursor.fetchone() + self.assertIsInstance(row, aiosqlite.Row) + self.assertEqual(row[1], 1) + self.assertEqual(row[2], "hi") + self.assertEqual(row["k"], 1) + self.assertEqual(row["d"], "hi") + + db.row_factory = aiosqlite.Row + db.text_factory = bytes + self.assertEqual(db.row_factory, aiosqlite.Row) + self.assertEqual(db.text_factory, bytes) + + async with db.cursor() as cursor: + await cursor.execute("select * from test_properties") + row = await cursor.fetchone() + self.assertIsInstance(row, aiosqlite.Row) + self.assertEqual(row[1], 1) + self.assertEqual(row[2], b"hi") + self.assertEqual(row["k"], 1) + self.assertEqual(row["d"], b"hi") + + async def test_fetch_all(self): + async with aiosqlite.connect(TEST_DB) as db: + await db.execute( + "create table test_fetch_all (i integer primary key asc, k integer)" + ) + await db.execute( + "insert into test_fetch_all (k) values (10), (24), (16), (32)" + ) + await db.commit() + + async with aiosqlite.connect(TEST_DB) as db: + cursor = await db.execute("select k from test_fetch_all where k < 30") + rows = await cursor.fetchall() + self.assertEqual(rows, [(10,), (24,), (16,)]) + + async def test_enable_load_extension(self): + """Assert that after enabling extension loading, they can be loaded""" + async with aiosqlite.connect(TEST_DB) as db: + try: + await db.enable_load_extension(True) + await db.load_extension("test") + except OperationalError as e: + assert "not authorized" not in e.args + except AttributeError as e: + raise SkipTest( + "python was not compiled with sqlite3 " + "extension support, so we can't test it" + ) from e + + async def test_set_progress_handler(self): + """ + Assert that after setting a progress handler returning 1, DB operations are aborted + """ + async with aiosqlite.connect(TEST_DB) as db: + await db.set_progress_handler(lambda: 1, 1) + with self.assertRaises(OperationalError): + await db.execute( + "create table test_progress_handler (i integer primary key asc, k integer)" + ) + + async def test_create_function(self): + """Assert that after creating a custom function, it can be used""" + + def no_arg(): + return "no arg" + + def one_arg(num): + return num * 2 + + async with aiosqlite.connect(TEST_DB) as db: + await db.create_function("no_arg", 0, no_arg) + await db.create_function("one_arg", 1, one_arg) + + async with db.execute("SELECT no_arg();") as res: + row = await res.fetchone() + self.assertEqual(row[0], "no arg") + + async with db.execute("SELECT one_arg(10);") as res: + row = await res.fetchone() + self.assertEqual(row[0], 20) + + async def test_create_function_deterministic(self): + """Assert that after creating a deterministic custom function, it can be used. + + https://sqlite.org/deterministic.html + """ + + def one_arg(num): + return num * 2 + + async with aiosqlite.connect(TEST_DB) as db: + await db.create_function("one_arg", 1, one_arg, deterministic=True) + await db.execute("create table foo (id int, bar int)") + + # Non-deterministic functions cannot be used in indexes + await db.execute("create index t on foo(one_arg(bar))") + + async def test_set_trace_callback(self): + statements = [] + + def callback(statement: str): + statements.append(statement) + + async with aiosqlite.connect(TEST_DB) as db: + await db.set_trace_callback(callback) + + await db.execute("select 10") + self.assertIn("select 10", statements) + + async def test_connect_error(self): + bad_db = Path("/something/that/shouldnt/exist.db") + with self.assertRaisesRegex(OperationalError, "unable to open database"): + async with aiosqlite.connect(bad_db) as db: + self.assertIsNone(db) # should never be reached + + with self.assertRaisesRegex(OperationalError, "unable to open database"): + await aiosqlite.connect(bad_db) + + async def test_iterdump(self): + async with aiosqlite.connect(":memory:") as db: + await db.execute("create table foo (i integer, k charvar(250))") + await db.executemany( + "insert into foo values (?, ?)", [(1, "hello"), (2, "world")] + ) + + lines = [line async for line in db.iterdump()] + self.assertEqual( + lines, + [ + "BEGIN TRANSACTION;", + "CREATE TABLE foo (i integer, k charvar(250));", + "INSERT INTO \"foo\" VALUES(1,'hello');", + "INSERT INTO \"foo\" VALUES(2,'world');", + "COMMIT;", + ], + ) + + async def test_cursor_on_closed_connection(self): + db = await aiosqlite.connect(TEST_DB) + + cursor = await db.execute("select 1, 2") + await db.close() + with self.assertRaisesRegex(ValueError, "Connection closed"): + await cursor.fetchall() + with self.assertRaisesRegex(ValueError, "Connection closed"): + await cursor.fetchall() + + async def test_cursor_on_closed_connection_loop(self): + db = await aiosqlite.connect(TEST_DB) + + cursor = await db.execute("select 1, 2") + tasks = [] + for i in range(100): + if i == 50: + tasks.append(asyncio.ensure_future(db.close())) + tasks.append(asyncio.ensure_future(cursor.fetchall())) + for task in tasks: + try: + await task + except sqlite3.ProgrammingError: + pass + + async def test_close_twice(self): + db = await aiosqlite.connect(TEST_DB) + + await db.close() + + # no error + await db.close() + + async def test_backup_aiosqlite(self): + def progress(a, b, c): + print(a, b, c) + + async with aiosqlite.connect(":memory:") as db1, aiosqlite.connect( + ":memory:" + ) as db2: + await db1.execute("create table foo (i integer, k charvar(250))") + await db1.executemany( + "insert into foo values (?, ?)", [(1, "hello"), (2, "world")] + ) + await db1.commit() + + with self.assertRaisesRegex(OperationalError, "no such table: foo"): + await db2.execute("select * from foo") + + await db1.backup(db2, progress=progress) + + async with db2.execute("select * from foo") as cursor: + rows = await cursor.fetchall() + self.assertEqual(rows, [(1, "hello"), (2, "world")]) + + async def test_backup_sqlite(self): + async with aiosqlite.connect(":memory:") as db1: + with sqlite3.connect(":memory:") as db2: + await db1.execute("create table foo (i integer, k charvar(250))") + await db1.executemany( + "insert into foo values (?, ?)", [(1, "hello"), (2, "world")] + ) + await db1.commit() + + with self.assertRaisesRegex(OperationalError, "no such table: foo"): + db2.execute("select * from foo") + + await db1.backup(db2) + + cursor = db2.execute("select * from foo") + rows = cursor.fetchall() + self.assertEqual(rows, [(1, "hello"), (2, "world")]) |