#!/usr/bin/env python3 # This testing code deliberately does nasty stuff so mypy isn't helpful # mypy: ignore-errors # type: ignore # This code verifies the async_meta.json file and the code match each # other. It also provides a get_meta function used when building type # stubs and documentation. import functools import importlib.resources import inspect import io import json import queue import threading import unittest from typing import Any, Callable, Literal import apsw import apsw.aio # The result is one of # sync - can only be used on sync connections (async gives exception) # async - can only be used on async connections (sync gives exception) # dual - returns awaitable in async, value in sync # value - returns a value in sync or async # # for attributes dual means awaitable get and worker thread set def get_meta( klass: str, member: str, kind: Literal["function"] | Literal["attribute"] ) -> Literal["sync"] | Literal["async"] | Literal["dual"] | Literal["value"]: assert kind in {"function", "attribute"} data = json.loads(importlib.resources.files("apsw.tests").joinpath("async_meta.json").read_text(encoding="utf8")) assert klass in data, f"{klass} not found" section = data[klass] # explicit? for k in "sync async dual value".split(): if member in section.get(k, tuple()): return k # defaults match member: case "__enter__" | "__exit__" | "__iter__" | "__next__": return "sync" case "__aenter__" | "__aexit__" | "aclose" | "__aiter__" | "__anext__": # aiter is technically value but we only want it on async objects # so that is handled elsewhere return "async" case "__repr__" | "__str__" | "__bool__" | "close": return "value" case _: return "dual" if kind == "function" else "value" # Classes where we fake there being an async version ASYNCABLE = {"Connection", "Cursor", "Blob", "Backup", "Session"} def sync_await(obj): it = obj.__await__() try: while True: next(it) except StopIteration as exc: return exc.value skip = set(dir(object())) - {"__repr__", "__str__"} def is_method(object, name): return inspect.ismethoddescriptor(getattr(type(object), name)) class AsyncMeta(unittest.TestCase): def tearDown(self): for c in apsw.connections(): try: c.close(True) except apsw.ThreadingViolationError: pass def classifyOne(self, send, is_attr, object, klass, member, value=None): # send is None for async access, callable for sync if is_attr: try: if send: if value is not None: send(functools.partial(setattr, object, member, value)) else: send(functools.partial(getattr, object, member)) else: v = getattr(object, member) if value is None else setattr(object, member, value) if hasattr(v, "__await__"): return "async" return "value" except TypeError as exc: if not send and exc.args[0] == "Using sync method in async context": return "exception" if send and exc.args[0] == "Using async method in sync context": return "exception" raise if member in {"__aexit__", "__exit__"}: args = (None, None, None) else: args = tuple() match klass: case "Connection": match member: case "async_run": args = (lambda : 1,) case "autovacuum_pages" | "collation_needed": args = (lambda *args: 1 / 0,) case "backup": args = "main", apsw.Connection(""), "main" case "blob_open": args = "main", "dummy", "column", 73, False case "column_metadata": args = "main", "dummy", "column" case "config": args = apsw.SQLITE_DBCONFIG_ENABLE_FKEY, -1 case ( "create_aggregate_function" | "create_collation" | "create_module" | "create_scalar_function" | "create_window_function" | "register_fts5_function" | "register_fts5_tokenizer" ): args = "foo", lambda *args: 1 / 0 case "db_filename" | "readonly" | "serialize" | "vfsname": args = ("main",) case "deserialize": args = "main", apsw.Connection("").serialize("main") case ( "drop_modules" | "preupdate_hook" | "set_authorizer" | "set_busy_handler" | "set_commit_hook" | "set_exec_trace" | "set_profile" | "set_progress_handler" | "set_rollback_hook" | "set_row_trace" | "set_update_hook" | "set_wal_hook" ): args = (None,) case ( "enable_load_extension" | "set_busy_timeout" | "set_last_insert_rowid" | "wal_autocheckpoint" ): # some of these depend on True being subclass of int args = (True,) case "execute": args = ("select 3",) case "executemany": args = "select ?", ((i,) for i in range(10)) case "file_control": args = "main", 12343, 0 case "fts5_tokenizer" | "fts5_tokenizer_available": args = ("ascii",) case "limit": args = (apsw.SQLITE_LIMIT_COLUMN,) case "load_extension": args = ("this does not exist I hope",) case "overload_function": args = "dummy", 33 case "pragma": args = ("user_version",) case "read": args = "main", 0, 0, 4096 case "setlk_timeout": args = 1, 1 case "status": args = (apsw.SQLITE_DBSTATUS_LOOKASIDE_USED,) case "table_exists": args = (None, "hello") case "trace_v2" | "vtab_config": args = (0,) case "Cursor": match member: case "execute": args = ("select 3",) case "executemany": args = "insert into dummy values(?)", ((i,) for i in range(5)) case "set_exec_trace" | "set_row_trace": args = (None,) case "Blob": match member: case "read": args = (1,) case "read_into": args = bytearray(10), 0, 1 case "reopen": args = (74,) case "seek": args = (1,) case "write": args = (b"a",) case "Session": match member: case "changeset_stream" | "patchset_stream" | "table_filter": args = (lambda x: True,) case "config": args = apsw.SQLITE_SESSION_OBJCONFIG_SIZE, -1 case "diff": args = "other", "dummy" try: if send: sync_await(send(functools.partial(getattr(object, member), *args))) else: if hasattr(getattr(object, member)(*args), "__await__"): return "async" return "value" except TypeError as exc: if not send and exc.args[0] == "Using sync in async context": return "exception" if send and exc.args[0] == "Using async in sync context": return "exception" raise except apsw.ExtensionLoadingError: return "value" except apsw.SQLError: if (klass, member) == ("Connection", "read"): return "value" raise except apsw.InvalidContextError: if klass == "Connection" and member in {"vtab_config", "vtab_on_conflict"}: return "value" raise except io.UnsupportedOperation: if klass == "Blob" and member in {"readline", "readlines", "truncate", "writelines"}: return "value" if send else "async" raise except OSError: if (klass, member) == ("Blob", "fileno"): return "value" raise def testMetaJson(self): apsw.async_controller.set(SimpleController) objects = { "Connection": None, "Cursor": None, "Blob": None, "Backup": None, "Session": None, # Changeset.apply is too difficult to include } assert ASYNCABLE == set(objects.keys()) changeset = None def ensure_objects(klass): # various operations result in the database etc being closed # so this ensures they are open but also some things can't be # done while active cursors etc are in play nonlocal changeset if objects["Connection"] is None: self.tearDown() for k in objects: objects[k] = None if klass != "Connection": ensure_objects("Connection") if not objects[klass]: match klass: case "Connection": value = sync_await(apsw.Connection.as_async("")) sync_await( value.execute(""" create table dummy(column); create table dummy2(column INTEGER PRIMARY KEY); insert into dummy(rowid, column) values(73, x'aabbcc'), (74, x'aabbcc'); """) ) case "Cursor": value = sync_await(objects["Connection"].execute("SELECT ?, * from dummy", ("hello",))) case "Blob": value = sync_await(objects["Connection"].blob_open("main", "dummy", "column", 73, True)) case "Backup": value = sync_await(objects["Connection"].backup("main", apsw.Connection(""), "main")) case "Session": sync_await(objects["Connection"].execute("attach ':memory:' as other; create table other.dummy(column INTEGER PRIMARY KEY)")) value = sync_await(apsw.aio.make_session(objects["Connection"], "main")) sync_await(value.attach()) sync_await( objects["Connection"].execute(""" insert into dummy2 values(1), (2), (3), (4); """) ) if changeset is None: changeset = sync_await(value.changeset()) case _: 1 / 0 objects[klass] = value def all_the_things(): res = [] for name in objects: ensure_objects(name) for attr in dir(objects[name]): if attr in skip or attr in {"as_async", "async_run", "__next__", "aclose"} or (name, attr) in old_names: continue res.append((name, attr)) self.tearDown() for name in objects: objects[name] = None return res last_klass = None # potentially screw up functioning malfunction = { "cursor_factory", "exec_trace", "row_trace", "finish", "afinish", "close", "aclose", "__exit__", "__aexit__", } pre = {"__aexit__": "__aenter__", "__exit__": "__enter__", "__next__": "__iter__", "__anext__": "__aiter__"} for klass, name in all_the_things(): with self.subTest(klass=klass, name=name): if klass != last_klass or name in malfunction: objects["Connection"] = None ensure_objects(klass) is_attr = not is_method(objects[klass], name) ensure_objects(klass) if name in pre: self.classifyOne(None, is_attr, objects[klass], klass, pre[name]) kind_async = self.classifyOne(None, is_attr, objects[klass], klass, name) if name in malfunction: objects["Connection"] = None ensure_objects(klass) if name in pre: self.classifyOne(objects["Connection"].async_controller.send, is_attr, objects[klass], klass, pre[name]) kind_sync = self.classifyOne( objects["Connection"].async_controller.send, is_attr, objects[klass], klass, name ) if is_attr: # check writable (mutex assertions) match name: case "transaction_mode": value = "DEFERRED" case "enabled" | "indirect": value = True case _: value = lambda *args: False try: setattr(objects[klass], name, value) except AttributeError as exc: if "objects is not writable" in str(exc) or "readonly attribute" in str(exc): pass else: raise except TypeError as exc: if exc.args[0] in {"Using sync in async context", "Using async in sync context"}: pass else: raise match (kind_sync, kind_async): case ("value", "value"): kind = "value" case ("exception", "async"): kind = "async" case ("value", "exception"): kind = "sync" case ("value", "async"): kind = "dual" case ("exception", "value"): kind = "async" case _: raise ValueError(f"{kind_sync=} {kind_async=}") expected_kind = get_meta(klass, name, "attribute" if is_attr else "function") self.assertEqual(kind, expected_kind, f"{klass=} {name=}") class SimpleController: # This is awaitable that blocks the caller class Request: call: Callable event: threading.Event result: Any is_exception: bool def __await__(self): if False: yield self.event.wait() if self.is_exception: raise self.result return self.result def __init__(self): self.queue = queue.SimpleQueue() threading.Thread(daemon=True, target=self.worker_thread_run, args=(self.queue,)).start() def close(self): self.queue.put(None) self.queue = None def send(self, call): req = SimpleController.Request() req.event = threading.Event() req.call = call self.queue.put(req) return req def configure(self, db): pass def worker_thread_run(self, q): while (req := q.get()) is not None: try: req.result = req.call() req.is_exception = False except BaseException as exc: req.result = exc req.is_exception = True req.event.set() class RawController: "This is used to find out what an API returned" class Request: call: Callable event: threading.Event result: Any is_exception: bool def __init__(self): self.queue = queue.SimpleQueue() threading.Thread(daemon=True, target=self.worker_thread_run, args=(self.queue,)).start() def close(self): self.queue.put(None) self.queue = None def send(self, call): req = RawController.Request() req.event = threading.Event() req.call = call self.queue.put(req) req.event.wait() if req.is_exception: raise req.result return req.result def worker_thread_run(self, q): while (req := q.get()) is not None: try: req.result = req.call() req.is_exception = False except BaseException as exc: req.result = exc req.is_exception = True req.event.set() # generated from tools/renames.json old_names = { ("apsw", "apswversion"), ("apsw", "enablesharedcache"), ("apsw", "exceptionfor"), ("apsw", "memoryhighwater"), ("apsw", "memoryused"), ("apsw", "releasememory"), ("apsw", "softheaplimit"), ("apsw", "sqlitelibversion"), ("apsw", "vfsnames"), ("Backup", "pagecount"), ("Blob", "readinto"), ("Connection", "blobopen"), ("Connection", "cacheflush"), ("Connection", "collationneeded"), ("Connection", "createaggregatefunction"), ("Connection", "createcollation"), ("Connection", "createmodule"), ("Connection", "createscalarfunction"), ("Connection", "enableloadextension"), ("Connection", "exectrace"), ("Connection", "filecontrol"), ("Connection", "getautocommit"), ("Connection", "getexectrace"), ("Connection", "getrowtrace"), ("Connection", "loadextension"), ("Connection", "overloadfunction"), ("Connection", "rowtrace"), ("Connection", "setauthorizer"), ("Connection", "setbusyhandler"), ("Connection", "setbusytimeout"), ("Connection", "setcommithook"), ("Connection", "setexectrace"), ("Connection", "setprofile"), ("Connection", "setprogresshandler"), ("Connection", "setrollbackhook"), ("Connection", "setrowtrace"), ("Connection", "setupdatehook"), ("Connection", "setwalhook"), ("Connection", "sqlite3pointer"), ("Connection", "totalchanges"), ("Cursor", "exectrace"), ("Cursor", "getconnection"), ("Cursor", "getdescription"), ("Cursor", "getexectrace"), ("Cursor", "getrowtrace"), ("Cursor", "rowtrace"), ("Cursor", "setexectrace"), ("Cursor", "setrowtrace"), } if __name__ == "__main__": unittest.main()