test: share unbound unit test sessions (#39690)

This commit is contained in:
Escape0707
2026-07-28 14:12:08 +00:00
committed by GitHub
parent 7083a953e1
commit 1419c7c6db
3 changed files with 51 additions and 6 deletions
+31
View File
@@ -152,6 +152,18 @@ def _sqlite_session_factory(
return 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 @pytest.fixture
def sqlite_engine(_sqlite_engine: Engine) -> Engine: def sqlite_engine(_sqlite_engine: Engine) -> Engine:
"""Expose the pristine full-schema SQLite engine to tests.""" """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 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: def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:
"""Persist the owner identity resolved by service-API app authentication. """Persist the owner identity resolved by service-API app authentication.
@@ -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: def _make_app(mode: AppMode) -> App:
return App( return App(
id="app-id", id="app-id",
@@ -5,6 +5,7 @@ from threading import Barrier
import pytest import pytest
from sqlalchemy import create_engine, inspect, text from sqlalchemy import create_engine, inspect, text
from sqlalchemy.engine import URL, Engine from sqlalchemy.engine import URL, Engine
from sqlalchemy.exc import UnboundExecutionError
from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import QueuePool 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 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( def test_sqlite_session_factory_shares_one_database_across_worker_sessions(
sqlite_session_factory: sessionmaker[Session], sqlite_session_factory: sessionmaker[Session],
) -> None: ) -> None: