Add more tests for async and dialect

Author: antoineleclair

tests/integration/test_async_context.py

@@ -0,0 +1,28 @@
+"""Integration tests for sync engine used inside an async context."""
+
+import asyncio
+
+import pytest
+from sqlalchemy import create_engine, text
+
+
+@pytest.mark.integration
+class TestSyncEngineInAsyncContext:
+    def test_sync_engine_inside_running_loop(self, engine_url: str) -> None:
+        """Sync engine must work when called from inside a running event loop.
+
+        This simulates the scenario where a sync SQLAlchemy engine is used
+        during app startup inside an async server like uvicorn.
+        """
+
+        async def _run_inside_loop() -> str:
+            engine = create_engine(engine_url)
+            with engine.connect() as conn:
+                result = conn.execute(text("SELECT 'ok'"))
+                row = result.fetchone()
+            engine.dispose()
+            assert row is not None
+            return str(row[0])
+
+        result = asyncio.run(_run_inside_loop())
+        assert result == "ok"

tests/test_dialect.py

@@ -58,6 +58,25 @@ def test_import_dbapi(self) -> None:
         dbapi = DqliteDialect_aio.import_dbapi()
         assert hasattr(dbapi, "aconnect")
 
+    def test_import_dbapi_has_paramstyle(self) -> None:
+        """Async dbapi module must expose paramstyle for SQLAlchemy dialect init."""
+        dbapi = DqliteDialect_aio.import_dbapi()
+        assert dbapi.paramstyle == "qmark"
+
+    def test_import_dbapi_has_module_attributes(self) -> None:
+        """Async dbapi module must expose PEP 249 attributes for SQLAlchemy."""
+        dbapi = DqliteDialect_aio.import_dbapi()
+        assert dbapi.apilevel == "2.0"
+        assert dbapi.threadsafety == 1
+
+    def test_create_async_engine(self) -> None:
+        """create_async_engine must not raise during dialect initialization."""
+        from sqlalchemy.ext.asyncio import create_async_engine
+
+        engine = create_async_engine("dqlite+aio://localhost:19001/test")
+        assert engine.dialect.name == "dqlite"
+        assert engine.dialect.driver == "dqlitedbapi_aio"
+
 
 class TestURLParsing:
     def test_parse_basic_url(self) -> None: