test: use pristine file-backed SQLite fixtures (#39624)

Co-authored-by: Asuka Minato <i@asukaminato.eu.org>

api/models/model.py

@@ -1117,14 +1117,14 @@ class ExporleBanner(TypeBase):
     status: Mapped[BannerStatus] = mapped_column(
         EnumText(BannerStatus, length=255),
         nullable=False,
-        server_default=sa.text("'enabled'::character varying"),
+        server_default=sa.text("'enabled'"),
         default=BannerStatus.ENABLED,
     )
     created_at: Mapped[datetime] = mapped_column(
         sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
     )
     language: Mapped[str] = mapped_column(
-        String(255), nullable=False, server_default=sa.text("'en-US'::character varying"), default="en-US"
+        String(255), nullable=False, server_default=sa.text("'en-US'"), default="en-US"
     )
 
 

api/tests/unit_tests/conftest.py

@@ -1,11 +1,13 @@
 import os
+import shutil
 from collections.abc import Iterator
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 from flask import Flask
 from sqlalchemy import create_engine
-from sqlalchemy.engine import Engine
+from sqlalchemy.engine import URL, Engine
 from sqlalchemy.orm import Session, sessionmaker
 
 # Getting the absolute path of the current file's directory
@@ -35,7 +37,7 @@ os.environ.setdefault("OPENDAL_SCHEME", "fs")
 os.environ.setdefault("OPENDAL_FS_ROOT", "/tmp/dify-storage")
 os.environ.setdefault("STORAGE_TYPE", "opendal")
 
-from core.db.session_factory import configure_session_factory, session_factory
+import core.db.session_factory as session_factory_module
 from extensions import ext_redis
 from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
 from models.base import TypeBase
@@ -111,42 +113,70 @@ def reset_secret_key():
         dify_config.SECRET_KEY = original
 
 
-@pytest.fixture(scope="session")
-def _unit_test_engine():
-    engine = create_engine("sqlite:///:memory:")
-    yield engine
-    engine.dispose()
-
-
 @pytest.fixture
-def sqlite_engine() -> Iterator[Engine]:
-    """Create an isolated in-memory SQLite engine for tests that need a disposable database."""
+def _sqlite_engine(_sqlite_database_template: Path, tmp_path: Path) -> Iterator[Engine]:
+    """Create an engine over a pristine per-test copy of the SQLite schema."""
+
+    database_path = tmp_path / "unit-tests.sqlite3"
+    shutil.copyfile(_sqlite_database_template, database_path)
+    engine = create_engine(URL.create("sqlite", database=str(database_path)))
 
-    engine = create_engine("sqlite:///:memory:")
     try:
         yield engine
     finally:
         engine.dispose()
+        database_path.unlink(missing_ok=True)
 
 
-@pytest.fixture
-def sqlite_session(request: pytest.FixtureRequest, sqlite_engine: Engine) -> Iterator[Session]:
-    """Yield a SQLite session after creating the model tables passed through ``request.param``."""
+@pytest.fixture(scope="session")
+def _sqlite_database_template(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Create one empty full-schema SQLite database per pytest worker."""
 
-    models: tuple[type[TypeBase], ...] = request.param
-    tables = [model.metadata.tables[model.__tablename__] for model in models]
-    TypeBase.metadata.create_all(sqlite_engine, tables=tables)
-    session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
-    with session_factory() as session:
-        yield session
+    database_path = tmp_path_factory.mktemp("sqlite-template") / "unit-tests.sqlite3"
+    engine = create_engine(URL.create("sqlite", database=str(database_path)))
+    try:
+        TypeBase.metadata.create_all(engine)
+    finally:
+        engine.dispose()
+    return database_path
 
 
 @pytest.fixture(autouse=True)
-def _configure_session_factory(_unit_test_engine):
-    try:
-        session_factory.get_session_maker()
-    except RuntimeError:
-        configure_session_factory(_unit_test_engine, expire_on_commit=False)
+def _sqlite_session_factory(
+    _sqlite_engine: Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> sessionmaker[Session]:
+    """Bind all unit-test Sessions to the pristine full-schema SQLite database."""
+
+    factory = sessionmaker(bind=_sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr(session_factory_module, "_session_maker", factory)
+    return factory
+
+
+@pytest.fixture
+def sqlite_engine(_sqlite_engine: Engine) -> Engine:
+    """Expose the pristine full-schema SQLite engine to tests."""
+
+    return _sqlite_engine
+
+
+@pytest.fixture
+def sqlite_session_factory(_sqlite_session_factory: sessionmaker[Session]) -> sessionmaker[Session]:
+    """Expose the shared SQLite session factory to tests."""
+
+    return _sqlite_session_factory
+
+
+@pytest.fixture
+def sqlite_session(_sqlite_session_factory: sessionmaker[Session]) -> Iterator[Session]:
+    """Yield a session over the pristine full-schema SQLite database.
+
+    Legacy indirect model parameters remain accepted by pytest but are ignored.
+    Remove those decorators as their test files receive individual review.
+    """
+
+    with _sqlite_session_factory() as session:
+        yield session
 
 
 def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:

api/tests/unit_tests/events/test_update_provider_when_message_created.py

@@ -18,7 +18,6 @@ from models.provider import ProviderType
 @pytest.fixture
 def credit_pool_session_factory(sqlite_engine: Engine) -> Iterator[sessionmaker[Session]]:
     """Bind message-created accounting to fixture-owned SQLite sessions."""
-    TenantCreditPool.__table__.create(sqlite_engine)
     session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
     with patch("events.event_handlers.update_provider_when_message_created.db.session", session_factory):
         yield session_factory

api/tests/unit_tests/services/test_account_service.py

@@ -1,5 +1,4 @@
 import json
-from collections.abc import Iterator
 from datetime import datetime, timedelta
 from unittest.mock import MagicMock, patch
 from uuid import UUID
@@ -11,7 +10,6 @@ from sqlalchemy.orm import Session
 from configs import dify_config
 from models.account import (
     Account,
-    AccountIntegrate,
     AccountStatus,
     Tenant,
     TenantAccountJoin,
@@ -114,22 +112,6 @@ class TestAccountService:
     - Error conditions and edge cases
     """
 
-    @pytest.fixture
-    def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
-        """SQLite session with the account/workspace tables these service tests touch."""
-        tables = [
-            model.metadata.tables[model.__tablename__]
-            for model in (
-                Account,
-                Tenant,
-                TenantAccountJoin,
-                TenantPluginAutoUpgradeStrategy,
-            )
-        ]
-        Account.metadata.create_all(sqlite_engine, tables=tables)
-        with Session(sqlite_engine, expire_on_commit=False) as session:
-            yield session
-
     @pytest.fixture
     def mock_password_dependencies(self):
         """Mock setup for password-related functions."""
@@ -1264,24 +1246,6 @@ class TestRegisterService:
     - Error conditions and edge cases
     """
 
-    @pytest.fixture
-    def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
-        """SQLite session with the account/workspace tables registration flows touch."""
-        tables = [
-            model.metadata.tables[model.__tablename__]
-            for model in (
-                Account,
-                AccountIntegrate,
-                Tenant,
-                TenantAccountJoin,
-                TenantPluginAutoUpgradeStrategy,
-                DifySetup,
-            )
-        ]
-        Account.metadata.create_all(sqlite_engine, tables=tables)
-        with Session(sqlite_engine, expire_on_commit=False) as session:
-            yield session
-
     @pytest.fixture
     def mock_redis_dependencies(self):
         """Mock setup for Redis-related functions."""

api/tests/unit_tests/test_sqlite_fixtures.py

@@ -0,0 +1,83 @@
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from threading import Barrier
+
+import pytest
+from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.engine import URL, Engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import QueuePool
+
+import core.db.session_factory as session_factory_module
+from models.account import Account
+from models.base import TypeBase
+from models.model import ExporleBanner
+
+
+def test_sqlite_session_contains_the_full_registered_schema(sqlite_session: Session) -> None:
+    table_names = set(inspect(sqlite_session.get_bind()).get_table_names())
+
+    assert table_names == set(TypeBase.metadata.tables)
+
+
+@pytest.mark.parametrize("sqlite_session", [(Account,)], indirect=True)
+def test_sqlite_session_accepts_deferred_legacy_indirect_parameters(sqlite_session: Session) -> None:
+    """Prove legacy model parameters no longer limit the copied schema."""
+
+    assert inspect(sqlite_session.get_bind()).has_table(ExporleBanner.__tablename__)
+
+
+def test_sqlite_engine_is_a_pristine_file_copy(
+    sqlite_engine: Engine,
+    request: pytest.FixtureRequest,
+) -> None:
+    sqlite_database_template: Path = request.getfixturevalue("_sqlite_database_template")
+    assert isinstance(sqlite_engine.pool, QueuePool)
+    assert sqlite_engine.url.database != str(sqlite_database_template)
+
+    with sqlite_engine.begin() as connection:
+        connection.execute(text("CREATE TABLE per_test_mutation (value INTEGER NOT NULL)"))
+
+    template_engine = create_engine(URL.create("sqlite", database=str(sqlite_database_template)))
+    try:
+        assert not inspect(template_engine).has_table("per_test_mutation")
+    finally:
+        template_engine.dispose()
+
+
+def test_core_session_factory_uses_the_shared_sqlite_session_factory(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    assert session_factory_module.session_factory.get_session_maker() is sqlite_session_factory
+
+    with sqlite_session_factory.begin() as session:
+        session.execute(text("CREATE TABLE global_factory_probe (value INTEGER NOT NULL)"))
+        session.execute(text("INSERT INTO global_factory_probe (value) VALUES (42)"))
+
+    with session_factory_module.session_factory.create_session() as session:
+        assert session.scalar(text("SELECT value FROM global_factory_probe")) == 42
+
+
+def test_sqlite_session_factory_shares_one_database_across_worker_sessions(
+    sqlite_session_factory: sessionmaker[Session],
+) -> None:
+    with sqlite_session_factory.begin() as session:
+        session.execute(text("CREATE TABLE thread_probe (value INTEGER NOT NULL)"))
+        session.execute(text("INSERT INTO thread_probe (value) VALUES (42)"))
+
+    worker_barrier = Barrier(2)
+
+    def read_value() -> tuple[int, int]:
+        with sqlite_session_factory() as session:
+            connection = session.connection()
+            worker_barrier.wait(timeout=1)
+            value = session.scalar(text("SELECT value FROM thread_probe"))
+            connection_id = id(connection.connection.dbapi_connection)
+            return connection_id, value
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        futures = [executor.submit(read_value) for _ in range(2)]
+        results = [future.result() for future in futures]
+
+    assert {value for _, value in results} == {42}
+    assert len({connection_id for connection_id, _ in results}) == 2