Coverage for fastapi_restly / testing / _fixtures.py: 89%
117 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-02 09:54 +0000
1import traceback
2from contextlib import asynccontextmanager
3from pathlib import Path
4from typing import AsyncIterator, Iterator
5from unittest.mock import AsyncMock, MagicMock, patch
7import alembic
8import alembic.command
9import alembic.config
10import pytest
11import pytest_asyncio
12from fastapi import FastAPI
13from sqlalchemy.ext.asyncio import AsyncConnection
14from sqlalchemy.ext.asyncio import AsyncSession as SA_AsyncSession
15from sqlalchemy.orm import Session as SA_Session
17from ..db import activate_savepoint_only_mode, fr_globals, get_fr_globals
18from ._client import RestlyTestClient
21@pytest.fixture(scope="session")
22def project_root() -> Path:
23 """Return the project root directory."""
24 # Try to find the project root by looking for pyproject.toml
25 current = Path.cwd()
26 while current != current.parent:
27 if (current / "pyproject.toml").exists():
28 return current
29 current = current.parent
30 raise Exception("Could not find a pyproject.toml to establish project root")
33@pytest.fixture(autouse=True, scope="session")
34def autouse_alembic_upgrade(project_root):
35 # Only run alembic migrations if the alembic directory exists
36 alembic_dir = project_root / "alembic"
37 if not alembic_dir.exists():
38 return # Skip if no alembic directory
40 # TODO: Move project_root to Settings?
41 alembic_cfg = alembic.config.Config(project_root / "alembic.ini")
42 alembic_cfg.set_main_option("script_location", str(alembic_dir))
43 try:
44 alembic.command.upgrade(alembic_cfg, "head")
45 except Exception as exc:
46 tb = traceback.format_exc()
47 pytest.exit(
48 f"Alembic migrations failed: {exc}\n\nTraceback:\n{tb}", returncode=1
49 )
52@pytest.fixture(autouse=True, scope="session")
53def autouse_savepoint_only_mode_sessions() -> None:
54 # Only run if database connections are set up
55 if not fr_globals.async_make_session and not fr_globals.make_session:
56 return # Skip if no database connections
58 if fr_globals.async_make_session: 58 ↛ 60line 58 didn't jump to line 60 because the condition on line 58 was always true
59 activate_savepoint_only_mode(fr_globals.async_make_session)
60 if fr_globals.make_session:
61 activate_savepoint_only_mode(fr_globals.make_session)
64@pytest.fixture
65def _shared_connection():
66 # Sync tests need a sync sessionmaker, but async-only projects should still
67 # be able to use the async_session fixture without one.
68 if not fr_globals.make_session:
69 yield None
70 return
72 engine = fr_globals.make_session.kw["bind"]
73 with engine.connect() as conn:
74 yield conn
77@pytest_asyncio.fixture
78async def async_session(_shared_connection) -> AsyncIterator[SA_AsyncSession]:
79 """
80 Pytest fixture providing a database session with savepoint-based isolation.
82 Each test runs inside a savepoint. At the end of the test, the savepoint is
83 rolled back, leaving the database clean for the next test.
85 NOTE: Calling session.rollback() inside a test rolls back to the last savepoint
86 (created by each patched commit()), NOT to the start of the test. This differs
87 from production behavior. To undo all changes in a test, use session.rollback()
88 after each commit(), but be aware that data added before the last commit() is
89 still visible.
90 """
91 # Only run if database connections are set up
92 if not fr_globals.async_make_session: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 pytest.skip("Database connection not set up")
95 async_engine = fr_globals.async_make_session.kw["bind"]
97 @asynccontextmanager
98 async def get_bound_async_connection():
99 if _shared_connection is None: 99 ↛ 104line 99 didn't jump to line 104 because the condition on line 99 was always true
100 async with async_engine.connect() as async_conn:
101 yield async_conn
102 return
104 async_conn = AsyncConnection(async_engine, sync_connection=_shared_connection)
105 async with async_conn:
106 yield async_conn
108 async with get_bound_async_connection() as async_conn:
109 async with fr_globals.async_make_session(bind=async_conn) as sess:
110 async def begin_nested():
111 await sess.begin_nested()
112 return sess
114 mock_sessionmaker = AsyncMock()
115 mock_sessionmaker.side_effect = begin_nested
116 # session.begin() is used as a context manager (async with session.begin():)
117 # We need it to also return our savepoint session so explicit transaction
118 # blocks work correctly with our isolation mechanism
119 mock_sessionmaker.begin.return_value.__aenter__.side_effect = begin_nested
120 # FIXME: begin().__aexit__ should flush pending changes to make them visible
121 # within the test, but currently does not. This may cause visibility issues
122 # when using `async with session.begin(): ...` blocks inside tests.
123 # Impact: changes inside explicit begin() blocks may not be visible after exit.
125 async def passthrough_exit(self, exc_type, exc_value, traceback):
126 await sess.flush()
127 return False # re-raise any exception
129 async def patched_commit(self):
130 await sess.flush()
131 await sess.begin_nested()
133 globals_obj = get_fr_globals()
134 original_async_make_session = globals_obj.async_make_session
135 globals_obj.async_make_session = mock_sessionmaker
136 try:
137 with (
138 patch.object(SA_AsyncSession, "__aexit__", passthrough_exit),
139 patch.object(SA_AsyncSession, "commit", patched_commit),
140 ):
141 yield sess
142 finally:
143 globals_obj.async_make_session = original_async_make_session
146@pytest.fixture
147def session(_shared_connection) -> Iterator[SA_Session]:
148 """
149 Pytest fixture providing a database session with savepoint-based isolation.
151 Each test runs inside a savepoint. At the end of the test, the savepoint is
152 rolled back, leaving the database clean for the next test.
154 NOTE: Calling session.rollback() inside a test rolls back to the last savepoint
155 (created by each patched commit()), NOT to the start of the test. This differs
156 from production behavior. To undo all changes in a test, use session.rollback()
157 after each commit(), but be aware that data added before the last commit() is
158 still visible.
159 """
160 # Only run if database connections are set up
161 if not fr_globals.make_session: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 pytest.skip("Database connection not set up")
164 with fr_globals.make_session(bind=_shared_connection) as sess:
166 def begin_nested():
167 sess.begin_nested()
168 return sess
170 mock_sessionmaker = MagicMock()
171 mock_sessionmaker.side_effect = begin_nested
172 # session.begin() is used as a context manager (with session.begin():)
173 # We need it to also return our savepoint session so explicit transaction
174 # blocks work correctly with our isolation mechanism
175 mock_sessionmaker.begin.return_value.__enter__.side_effect = begin_nested
176 # FIXME: begin().__exit__ should flush pending changes to make them visible
177 # within the test, but currently does not. This may cause visibility issues
178 # when using `with session.begin(): ...` blocks inside tests.
179 # Impact: changes inside explicit begin() blocks may not be visible after exit.
181 def passthrough_exit(self, exc_type, exc_value, traceback):
182 sess.flush()
183 return False # re-raise any exception
185 def patched_commit(self):
186 sess.flush()
187 sess.begin_nested()
189 globals_obj = get_fr_globals()
190 original_make_session = globals_obj.make_session
191 globals_obj.make_session = mock_sessionmaker
192 try:
193 with (
194 patch.object(SA_Session, "__exit__", passthrough_exit),
195 patch.object(SA_Session, "commit", patched_commit),
196 ):
197 yield sess
198 finally:
199 globals_obj.make_session = original_make_session
202@pytest.fixture
203def app() -> FastAPI:
204 """Create a FastAPI app instance for testing."""
205 return FastAPI()
208@pytest.fixture
209def client(app) -> RestlyTestClient:
210 """Create a RestlyTestClient instance for testing."""
211 return RestlyTestClient(app)