diff --git a/api/tests/unit_tests/conftest.py b/api/tests/unit_tests/conftest.py index 4edf0be3193..3870d9b1d94 100644 --- a/api/tests/unit_tests/conftest.py +++ b/api/tests/unit_tests/conftest.py @@ -152,6 +152,18 @@ def _sqlite_session_factory( return factory +@pytest.fixture +def _unbound_session_factory( + _sqlite_session_factory: sessionmaker[Session], + monkeypatch: pytest.MonkeyPatch, +) -> sessionmaker[Session]: + """Create one unbound factory and install it as the global test factory.""" + + factory = sessionmaker() + 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.""" @@ -178,6 +190,25 @@ def sqlite_session(_sqlite_session_factory: sessionmaker[Session]) -> Iterator[S yield session +@pytest.fixture +def unbound_session_factory(_unbound_session_factory: sessionmaker[Session]) -> sessionmaker[Session]: + """Expose an unbound factory for paths that must not require persistence.""" + + return _unbound_session_factory + + +@pytest.fixture +def unbound_session(_unbound_session_factory: sessionmaker[Session]) -> Iterator[Session]: + """Yield an unbound Session for paths that must not require persistence. + + Bind-requiring database access fails, while bind-free Session operations can + still succeed. + """ + + with _unbound_session_factory() as session: + yield session + + def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin: """Persist the owner identity resolved by service-API app authentication. diff --git a/api/tests/unit_tests/services/test_human_input_service.py b/api/tests/unit_tests/services/test_human_input_service.py index 6d3e8d94c94..58d5d84a8e3 100644 --- a/api/tests/unit_tests/services/test_human_input_service.py +++ b/api/tests/unit_tests/services/test_human_input_service.py @@ -41,12 +41,6 @@ from services.human_input_service import ( ) -@pytest.fixture -def unbound_session_factory() -> sessionmaker[Session]: - """Supply the required constructor dependency without enabling database access.""" - return sessionmaker() - - def _make_app(mode: AppMode) -> App: return App( id="app-id", diff --git a/api/tests/unit_tests/test_sqlite_fixtures.py b/api/tests/unit_tests/test_sqlite_fixtures.py index c5a3fd660e1..11a60b407c7 100644 --- a/api/tests/unit_tests/test_sqlite_fixtures.py +++ b/api/tests/unit_tests/test_sqlite_fixtures.py @@ -5,6 +5,7 @@ from threading import Barrier import pytest from sqlalchemy import create_engine, inspect, text from sqlalchemy.engine import URL, Engine +from sqlalchemy.exc import UnboundExecutionError from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import QueuePool @@ -58,6 +59,25 @@ def test_core_session_factory_uses_the_shared_sqlite_session_factory( assert session.scalar(text("SELECT value FROM global_factory_probe")) == 42 +def test_unbound_session_factory_disables_explicit_and_global_database_access( + unbound_session_factory: sessionmaker[Session], +) -> None: + assert session_factory_module.session_factory.get_session_maker() is unbound_session_factory + + with unbound_session_factory() as session: + with pytest.raises(UnboundExecutionError): + session.get_bind() + + with session_factory_module.session_factory.create_session() as session: + with pytest.raises(UnboundExecutionError): + session.execute(text("SELECT 1")) + + +def test_unbound_session_rejects_database_access(unbound_session: Session) -> None: + with pytest.raises(UnboundExecutionError): + unbound_session.scalar(text("SELECT 1")) + + def test_sqlite_session_factory_shares_one_database_across_worker_sessions( sqlite_session_factory: sessionmaker[Session], ) -> None: