perf(testing): share one dialect and compiled-statement cache across in-memory SQLite test engines

CHANGELOG.md

@@ -16,6 +16,7 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Changed
 
+- Faster tests: `simple_module_test` gives every in-memory SQLite engine made through `init_db` with the same dialect options one shared dialect and compiled-statement cache, so a statement compiled in one test is reused by the next instead of being recompiled for every fresh engine. The databases stay separate. The patch is process-wide; set `SM_TEST_SEPARATE_COMPILE_CACHES=1` to skip it. `make test-py` drops from about 250 s to 232 s on 8 workers (test bodies 566 s to 511 s of worker time).
 - Faster test fixtures: `simple_module_test` builds the in-memory SQLite schema once per process and copies it into each test's database with SQLite's backup API, instead of running `create_all` for every `app` and `db_session` fixture. A database that already has tables is left alone. `make test-py` drops from 324 s to 256 s on 8 workers, and fixture setup time falls by about a third.
 - `simple_module_hosting.migrations.script_directory` reuses the loaded Alembic revision map while the revision files are unchanged (any added, removed or edited file reloads it). Each app boot previously loaded every migration file twice. The background-tasks Celery app also caches its entry-point scan for installed task packages.
 

docs/testing/fixtures.md

@@ -15,7 +15,7 @@ The `simple_module_test` plugin provides app-level fixtures that every test dire
 | `authenticated_client` | function | `httpx.AsyncClient` | Admin-scoped HTTP calls — carries a signed session cookie. |
 | `captured_mail` | function | `CapturedMail` | Asserting on outgoing mail: installs a recording sender (`.messages`, `.to(address)`) as the process mail sender. |
 
-All fixtures are **function-scoped** — each test gets its own. The SQLite database is in-memory; tearing down and re-creating is cheap. The first schema each process builds is kept as a template and copied into every later empty in-memory database with SQLite's backup API, so `CREATE TABLE` runs once per worker rather than once per test (`simple_module_test._sqlite_template`). A database that already has tables is never overwritten.
+All fixtures are **function-scoped** — each test gets its own. The SQLite database is in-memory; tearing down and re-creating is cheap. The first schema each process builds is kept as a template and copied into every later empty in-memory database with SQLite's backup API, so `CREATE TABLE` runs once per worker rather than once per test (`simple_module_test._sqlite_template`). A database that already has tables is never overwritten. Those engines also share one SQLAlchemy dialect and compiled-statement cache (`simple_module_test._compile_cache`), so SQL compiled in one test is reused by the next; the cache key carries no data, only statement structure. The patch applies to the whole pytest process; set `SM_TEST_SEPARATE_COMPILE_CACHES=1` to skip it.
 
 ## Postgres
 

framework/testing/simple_module_test/_compile_cache.py

@@ -0,0 +1,77 @@
+"""Share SQLAlchemy's compiled-statement cache across in-memory SQLite test engines.
+
+SQLAlchemy keeps compiled SQL per engine, keyed on the dialect object. Every
+``app`` and ``db_state`` fixture builds a new engine, so each test compiled
+every statement it ran from scratch: the boot path, the admin seed and the test
+body alike. A production process has one engine and pays this once.
+
+Engines made by ``init_db`` for an in-memory SQLite URL with the same dialect
+options are given the first such engine's dialect and compiled cache. The cache
+key is the statement's structure with its values extracted as bound parameters,
+which is what lets one production engine serve every request and tenant; the
+databases behind the engines stay separate.
+
+Patched at import, like the cheap password hashing in ``plugin``: test modules
+build engines during collection, before any fixture runs. The patch is
+process-wide, so a consumer suite that needs engines with separate dialects sets
+``SM_TEST_SEPARATE_COMPILE_CACHES`` to skip it.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from sqlalchemy.engine import URL
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.util import LRUCache
+
+from simple_module_test._sqlite_template import is_memory_sqlite
+
+# Engine options that change only how a connection is opened or logged, not the
+# dialect, so engines differing in them can share one.
+_NON_DIALECT_OPTIONS = frozenset({"echo", "connect_args", "poolclass"})
+
+_shared: dict[tuple, tuple[Any, Any]] = {}
+
+
+def _dialect_key(engine: Any, options: dict[str, Any]) -> tuple | None:
+    """The sharing key for a sync *engine*, or ``None`` when it must keep its own."""
+    if engine.url.get_backend_name() != "sqlite" or not is_memory_sqlite(engine):
+        return None
+    rest = tuple(sorted((k, repr(v)) for k, v in options.items() if k not in _NON_DIALECT_OPTIONS))
+    return (engine.url.drivername, rest)
+
+
+def share_sqlite_compile_cache() -> None:
+    if os.environ.get("SM_TEST_SEPARATE_COMPILE_CACHES"):
+        return
+    import simple_module_db.session as session
+
+    original = session.create_async_engine
+
+    def create_async_engine(url: str | URL, **kw: Any) -> AsyncEngine:
+        engine = original(url, **kw)
+        sync = engine.sync_engine
+        key = _dialect_key(sync, kw)
+        if key is None or sync._compiled_cache is None:  # None: query_cache_size=0
+            return engine
+        shared = _shared.get(key)
+        # A dialect is initialized by its own engine's first connection, so only
+        # one whose engine has connected can be handed on; until then the newest
+        # engine's dialect is the candidate.
+        if shared is None or shared[0].server_version_info is None:
+            # A cache of its own: the engine's would keep that engine (its size
+            # alert is a bound method) and its database alive for the whole run.
+            cache = LRUCache(sync._compiled_cache.capacity)
+            sync._compiled_cache = cache
+            _shared[key] = (sync.dialect, cache)
+        else:
+            # Swapped before the engine opens a connection, so the pool and every
+            # connection only ever see the shared dialect.
+            sync.dialect, sync._compiled_cache = shared
+            sync.pool._dialect = shared[0]
+        return engine
+
+    # ty types a module function by its own literal type, so any replacement is refused.
+    session.create_async_engine = create_async_engine  # ty: ignore[invalid-assignment]

framework/testing/simple_module_test/plugin.py

@@ -13,6 +13,7 @@
 
 import pytest
 
+from simple_module_test._compile_cache import share_sqlite_compile_cache
 from simple_module_test.app_factory import build_test_app as _build_test_app
 from simple_module_test.fake_events import FakeEventBus
 
@@ -142,6 +143,7 @@ def _init(self, password_hash: PasswordHash | None = None) -> None:
 
 
 _cheap_password_hashing()
+share_sqlite_compile_cache()
 
 
 @pytest.fixture

framework/testing/tests/test_shared_compile_cache.py

@@ -0,0 +1,115 @@
+"""In-memory SQLite test engines share one dialect and compiled cache (``_compile_cache``)."""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+from simple_module_db import init_db, session
+from simple_module_test import _compile_cache
+from simple_module_test.database import SQLITE_MEMORY
+from sqlalchemy import Column, Integer, MetaData, String, Table, inspect, select
+from sqlalchemy.engine import make_url
+
+_metadata = MetaData()
+_note = Table(
+    "compile_cache_probe_note",
+    _metadata,
+    Column("id", Integer, primary_key=True),
+    Column("body", String(20)),
+)
+
+
+@pytest.fixture(autouse=True)
+def _fresh_share(monkeypatch):
+    # Start from no shared dialect, and leave the suite's for the tests after this.
+    monkeypatch.setattr(_compile_cache, "_shared", {})
+
+
+@pytest.fixture
+async def states():
+    made = []
+
+    def make(url: str = SQLITE_MEMORY, **options):
+        state = init_db(url, **options)
+        made.append(state)
+        return state
+
+    yield make
+    for state in made:
+        await state.engine.dispose()
+
+
+async def _with_table(state) -> None:
+    async with state.engine.begin() as conn:
+        await conn.run_sync(_metadata.create_all)
+
+
+class TestSharing:
+    async def test_memory_engines_share_the_dialect_and_cache(self, states) -> None:
+        first = states()
+        await _with_table(first)  # connecting initializes the dialect it hands on
+        second = states().engine.sync_engine
+        assert first.engine.sync_engine.dialect is second.dialect
+        assert first.engine.sync_engine._compiled_cache is second._compiled_cache
+
+    async def test_an_engine_that_never_connected_hands_nothing_on(self, states) -> None:
+        """Its dialect was never initialized: reflection on the next engine needs
+        the SQLite version that initialization records."""
+        unused = states().engine.sync_engine
+        used = states()
+        await _with_table(used)
+        async with used.engine.connect() as conn:
+            columns = await conn.run_sync(
+                lambda sync_conn: inspect(sync_conn).get_columns("compile_cache_probe_note")
+            )
+        assert [column["name"] for column in columns] == ["id", "body"]
+        assert used.engine.sync_engine.dialect is not unused.dialect
+
+    async def test_a_statement_compiled_once_is_reused_by_the_next_engine(self, states) -> None:
+        first = states()
+        await _with_table(first)
+        second = states()
+        await _with_table(second)
+        query = select(_note.c.body).where(_note.c.id == 1)
+        async with first.engine.connect() as conn:
+            await conn.execute(query)
+        cached = len(first.engine.sync_engine._compiled_cache)
+
+        async with second.engine.connect() as conn:
+            await conn.execute(query)
+
+        assert len(second.engine.sync_engine._compiled_cache) == cached
+
+    async def test_the_databases_stay_separate(self, states) -> None:
+        first = states()
+        await _with_table(first)
+        second = states()
+        await _with_table(second)
+        async with first.engine.begin() as conn:
+            await conn.execute(_note.insert().values(body="only here"))
+        async with second.engine.connect() as conn:
+            assert (await conn.execute(select(_note.c.body))).all() == []
+
+
+class TestNotShared:
+    async def test_a_file_database_keeps_its_own(self, states, tmp_path) -> None:
+        memory = states().engine.sync_engine
+        file = states(f"sqlite+aiosqlite:///{tmp_path / 'own.db'}").engine.sync_engine
+        assert file.dialect is not memory.dialect
+
+    async def test_other_dialect_options_keep_their_own(self, states) -> None:
+        # ``init_db`` passes no dialect options today; the guard covers a caller
+        # of the patched factory that does.
+        plain = states().engine.sync_engine
+        custom = session.create_async_engine(SQLITE_MEMORY, json_serializer=json.dumps)
+        try:
+            assert custom.sync_engine.dialect is not plain.dialect
+        finally:
+            await custom.dispose()
+
+
+def test_server_databases_are_never_keyed() -> None:
+    server = SimpleNamespace(url=make_url("postgresql+asyncpg://u@h/db"))
+    assert _compile_cache._dialect_key(server, {}) is None