Coverage for conftest.py: 77%
138 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 15:50 +0100
1"""
2camcops_server/conftest.py
4===============================================================================
6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CamCOPS.
11 CamCOPS is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version 3 of the License, or
14 (at your option) any later version.
16 CamCOPS is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26**Configure server self-tests for Pytest.**
28"""
30# https://gist.githubusercontent.com/kissgyorgy/e2365f25a213de44b9a2/raw/f8b5bbf06c4969bc6bbe5316defef64137c9b1e3/sqlalchemy_conftest.py
32import configparser
33from io import StringIO
34import os
35import tempfile
36from typing import Generator, TYPE_CHECKING
38import pytest
39from sqlalchemy import event, MetaData
40from sqlalchemy.engine import create_engine
41from sqlalchemy.engine.interfaces import DBAPIConnection
42from sqlalchemy.orm import Session
43from sqlalchemy.pool import ConnectionPoolEntry
45import camcops_server.cc_modules.cc_all_models # noqa: F401
47# ... import side effects (ensure all models registered)
49from camcops_server.cc_modules.cc_baseconstants import CAMCOPS_SERVER_DIRECTORY
50from camcops_server.cc_modules.cc_config import get_demo_config
51from camcops_server.cc_modules.cc_sqlalchemy import (
52 Base,
53 make_memory_sqlite_engine,
54 make_file_sqlite_engine,
55)
57if TYPE_CHECKING:
58 from sqlalchemy.engine.base import Engine
60 # Should not need to import from _pytest in later versions of pytest
61 # https://github.com/pytest-dev/pytest/issues/7469
62 from _pytest.config.argparsing import Parser
63 from _pytest.fixtures import FixtureRequest
66TEST_DATABASE_FILENAME = os.path.join(
67 CAMCOPS_SERVER_DIRECTORY, "camcops_test.sqlite"
68)
71def pytest_addoption(parser: "Parser") -> None:
72 parser.addoption(
73 "--database-in-memory",
74 action="store_false",
75 dest="database_on_disk",
76 default=True,
77 help="Make SQLite database in memory",
78 )
80 # Borrowed from pytest-django
81 parser.addoption(
82 "--create-db",
83 action="store_true",
84 dest="create_db",
85 default=False,
86 help="Create the database even if it already exists",
87 )
89 parser.addoption(
90 "--mysql",
91 action="store_true",
92 dest="mysql",
93 default=False,
94 help="Use MySQL database instead of SQLite",
95 )
97 parser.addoption(
98 "--db-url",
99 dest="db_url",
100 default=(
101 "mysql+mysqldb://camcops:camcops@localhost:3306/test_camcops"
102 "?charset=utf8"
103 ),
104 help="SQLAlchemy test database URL (MySQL only)",
105 )
107 parser.addoption(
108 "--echo",
109 action="store_true",
110 dest="echo",
111 default=False,
112 help="Log all SQL statments to the default log handler",
113 )
116# noinspection PyUnusedLocal
117def set_sqlite_pragma(
118 dbapi_connection: DBAPIConnection, connection_record: ConnectionPoolEntry
119) -> None:
120 cursor = dbapi_connection.cursor()
121 cursor.execute("PRAGMA foreign_keys=ON")
122 cursor.close()
125@pytest.fixture(scope="session")
126def database_on_disk(request: "FixtureRequest") -> bool:
127 return request.config.getvalue("database_on_disk")
130@pytest.fixture(scope="session")
131def create_db(request: "FixtureRequest", database_on_disk: bool) -> bool:
132 if not database_on_disk:
133 return True
135 if not os.path.exists(TEST_DATABASE_FILENAME):
136 return True
138 return request.config.getvalue("create_db")
141@pytest.fixture(scope="session")
142def echo(request: "FixtureRequest") -> bool:
143 return request.config.getvalue("echo")
146# noinspection PyUnusedLocal
147@pytest.fixture(scope="session")
148def mysql(request: "FixtureRequest") -> bool:
149 return request.config.getvalue("mysql")
152@pytest.fixture(scope="session")
153def db_url(request: "FixtureRequest") -> bool:
154 return request.config.getvalue("db_url")
157@pytest.fixture(scope="session")
158def tmpdir_obj(
159 request: "FixtureRequest",
160) -> Generator[tempfile.TemporaryDirectory, None, None]:
161 tmpdir_obj = tempfile.TemporaryDirectory()
163 yield tmpdir_obj
165 tmpdir_obj.cleanup()
168@pytest.fixture(scope="session")
169def config_file(
170 request: "FixtureRequest", tmpdir_obj: tempfile.TemporaryDirectory
171) -> str:
172 # We're going to be using a test (SQLite) database, but we want to
173 # be very sure that nothing writes to a real database! Also, we will
174 # want to read from this dummy config at some point.
176 tmpconfigfilename = os.path.join(tmpdir_obj.name, "dummy_config.conf")
177 with open(tmpconfigfilename, "w") as file:
178 file.write(get_config_text())
180 return tmpconfigfilename
183def get_config_text() -> str:
184 config_text = get_demo_config()
185 parser = configparser.ConfigParser()
186 parser.read_string(config_text)
188 with StringIO() as buffer:
189 parser.write(buffer)
190 config_text = buffer.getvalue()
192 return config_text
195# https://gist.github.com/kissgyorgy/e2365f25a213de44b9a2
196# Author says "no [license], feel free to use it"
197# noinspection PyUnusedLocal
198@pytest.fixture(scope="session")
199def engine(
200 request: "FixtureRequest",
201 create_db: bool,
202 database_on_disk: bool,
203 echo: bool,
204 mysql: bool,
205 db_url: str,
206) -> Generator["Engine", None, None]:
208 if mysql:
209 engine = create_engine_mysql(db_url, create_db, echo)
210 else:
211 engine = create_engine_sqlite(create_db, echo, database_on_disk)
213 yield engine
214 engine.dispose()
217def create_engine_mysql(db_url: str, create_db: bool, echo: bool) -> "Engine":
219 # The database and the user with the given password from db_url
220 # need to exist.
221 # mysql> CREATE DATABASE <db_name>;
222 # mysql> GRANT ALL PRIVILEGES ON <db_name>.*
223 # TO <db_user>@localhost IDENTIFIED BY '<db_password>';
224 engine = create_engine(db_url, echo=echo, pool_pre_ping=True)
226 if create_db:
227 Base.metadata.drop_all(engine)
229 return engine
232def create_engine_sqlite(
233 create_db: bool, echo: bool, database_on_disk: bool
234) -> "Engine":
235 if create_db and database_on_disk:
236 try:
237 os.remove(TEST_DATABASE_FILENAME)
238 except OSError:
239 pass
241 if database_on_disk:
242 engine = make_file_sqlite_engine(TEST_DATABASE_FILENAME, echo=echo)
243 else:
244 engine = make_memory_sqlite_engine(echo=echo)
246 event.listen(engine, "connect", set_sqlite_pragma)
248 return engine
251# noinspection PyUnusedLocal
252@pytest.fixture(scope="session")
253def tables(
254 request: "FixtureRequest", engine: "Engine", create_db: bool
255) -> Generator[None, None, None]:
256 if create_db:
257 Base.metadata.create_all(engine)
258 yield
260 # Any post-session clean up would go here
261 # Foreign key constraint on _security_devices prevents this:
262 # Base.metadata.drop_all(engine)
263 # This would only be useful if we wanted to clean up the database
264 # after running the tests
267# noinspection PyUnusedLocal
268@pytest.fixture
269def dbsession(
270 request: "FixtureRequest", engine: "Engine", tables: None
271) -> Generator[Session, None, None]:
272 """
273 Returns an sqlalchemy session, and after the test tears down everything
274 properly.
275 """
277 connection = engine.connect()
278 # begin the nested transaction
279 transaction = connection.begin()
280 # use the connection with the already started transaction
281 session = Session(bind=connection)
283 yield session
285 session.close()
286 # roll back the broader transaction
287 transaction.rollback()
288 # put back the connection to the connection pool
289 connection.close()
292@pytest.fixture
293def setup(
294 request: "FixtureRequest",
295 engine: "Engine",
296 database_on_disk: bool,
297 mysql: bool,
298 dbsession: Session,
299 tmpdir_obj: tempfile.TemporaryDirectory,
300 config_file: str,
301) -> None:
302 # Pytest prefers function-based tests over unittest.TestCase subclasses and
303 # methods, but it still supports the latter perfectly well.
304 # We use this fixture in cc_unittest.py to store these values into
305 # DemoRequestTestCase and its descendants.
306 request.cls.engine = engine
307 request.cls.database_on_disk = database_on_disk
308 request.cls.dbsession = dbsession
309 request.cls.tmpdir_obj = tmpdir_obj
310 request.cls.db_filename = TEST_DATABASE_FILENAME
311 request.cls.mysql = mysql
312 request.cls.config_file = config_file
315@pytest.fixture(scope="session")
316def temp_engine(
317 request: "FixtureRequest", echo: bool
318) -> Generator["Engine", None, None]:
319 """
320 An in-memory database for testing export via the temp_session fixture.
321 """
322 engine = make_memory_sqlite_engine(echo=echo)
324 yield engine
326 engine.dispose()
329# noinspection PyUnusedLocal
330@pytest.fixture
331def temp_tables(
332 request: "FixtureRequest", temp_engine: "Engine"
333) -> Generator[None, None, None]:
335 # Unlike the tables fixture, we don't create any tables as they are created
336 # in the tests themselves and the columns change between tests. So the
337 # scope here is the default 'function', which means they are dropped after
338 # each test, rather than 'session', which would only drop them at the end
339 # of the test run.
341 yield
343 metadata = MetaData()
344 metadata.reflect(temp_engine)
345 metadata.drop_all(temp_engine)
348# noinspection PyUnusedLocal
349@pytest.fixture
350def temp_session(
351 request: "FixtureRequest",
352 temp_engine: "Engine",
353 temp_tables: None,
354) -> Generator[Session, None, None]:
355 """
356 Returns an sqlalchemy session, and after the test tears down everything
357 properly.
358 """
359 connection = temp_engine.connect()
360 # begin the nested transaction
361 transaction = connection.begin()
362 # use the connection with the already started transaction
363 session = Session(bind=connection)
365 yield session
367 session.close()
368 # roll back the broader transaction
369 transaction.rollback()
370 # put back the connection to the connection pool
371 connection.close()
374@pytest.fixture
375def setup_temp_session(
376 request: "FixtureRequest",
377 temp_engine: "Engine",
378 temp_session: Session,
379) -> None:
380 """
381 Use this fixture where a second, in-memory database is required.
382 Slow, so use sparingly.
383 """
384 request.cls.temp_session = temp_session
385 request.cls.temp_engine = temp_engine