Coverage for conftest.py: 40%
201 statements
« prev ^ index » next coverage.py v7.8.0, created at 2026-02-05 06:46 -0600
1"""
2crate_anon/conftest.py
4===============================================================================
6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CRATE.
11 CRATE 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 CRATE 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 CRATE. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26pytest configuration
28"""
30import os
31from os import pardir
32from os.path import abspath, dirname, join
33import tempfile
34from typing import Any, Generator, TYPE_CHECKING
36from cardinal_pythonlib.sqlalchemy.session import (
37 make_sqlite_url,
38 SQLITE_MEMORY_URL,
39)
40import pytest
41from sqlalchemy import event, inspect
42from sqlalchemy.engine import create_engine
43from sqlalchemy.engine.base import Engine
44from sqlalchemy.orm.session import Session
46# SecretBase is used for more than just testing
47from crate_anon.anonymise import SecretBase
48from crate_anon.testing import (
49 AnonTestBase,
50 SourceTestBase,
51)
53if TYPE_CHECKING:
54 from sqlite3 import Connection
56 # Should not need to import from _pytest in later versions of pytest
57 # https://github.com/pytest-dev/pytest/issues/7469
58 from _pytest.config.argparsing import Parser
59 from _pytest.fixtures import FixtureRequest
60 from sqlalchemy.pool.base import _ConnectionRecord
62_this_directory = dirname(abspath(__file__))
63CRATE_DIRECTORY = abspath(join(_this_directory, pardir))
65ANON_DATABASE_FILENAME = os.path.join(
66 CRATE_DIRECTORY, "crate_test_anon.sqlite"
67)
68SECRET_DATABASE_FILENAME = os.path.join(
69 CRATE_DIRECTORY, "crate_test_secret.sqlite"
70)
71SOURCE_DATABASE_FILENAME = os.path.join(
72 CRATE_DIRECTORY, "crate_test_source.sqlite"
73)
76def pytest_addoption(parser: "Parser") -> None:
77 parser.addoption(
78 "--databases-in-memory",
79 action="store_false",
80 dest="databases_on_disk",
81 default=True,
82 help="Make SQLite databases in memory",
83 )
85 # create-db is used by pytest-django
86 parser.addoption(
87 "--create-test-dbs",
88 action="store_true",
89 dest="create_test_dbs",
90 default=False,
91 help="Create the test databases even if they already exist",
92 )
94 parser.addoption(
95 "--anon-db-url",
96 dest="anon_db_url",
97 help="SQLAlchemy anonymised database URL (not applicable to SQLite)",
98 )
100 parser.addoption(
101 "--secret-db-url",
102 dest="secret_db_url",
103 help="SQLAlchemy secret database URL (not applicable to SQLite)",
104 )
106 parser.addoption(
107 "--source-db-url",
108 dest="source_db_url",
109 help="SQLAlchemy source database URL (not applicable to SQLite)",
110 )
112 parser.addoption(
113 "--echo",
114 action="store_true",
115 dest="echo",
116 default=False,
117 help="Log all SQL statments to the default log handler",
118 )
121def pytest_configure(config: pytest.Config) -> None:
122 if config.option.create_db:
123 message = (
124 "--create-db is a pytest-django option which is not "
125 "currently necessary. Did you mean --create-test-dbs?"
126 )
127 pytest.exit(message)
130# noinspection PyUnusedLocal
131def set_sqlite_pragma(
132 dbapi_connection: "Connection",
133 connection_record: "_ConnectionRecord",
134) -> None:
135 cursor = dbapi_connection.cursor()
136 cursor.execute("PRAGMA foreign_keys=ON")
137 cursor.close()
140@pytest.fixture(scope="session")
141def databases_on_disk(request: "FixtureRequest") -> bool:
142 return request.config.getvalue("databases_on_disk")
145@pytest.fixture(scope="session")
146def create_test_dbs(
147 request: "FixtureRequest", databases_on_disk: bool
148) -> bool:
149 if not databases_on_disk:
150 return True
152 if not os.path.exists(ANON_DATABASE_FILENAME):
153 return True
154 if not os.path.exists(SECRET_DATABASE_FILENAME):
155 return True
156 if not os.path.exists(SOURCE_DATABASE_FILENAME):
157 return True
159 return request.config.getvalue("create_test_dbs")
162@pytest.fixture(scope="session")
163def echo(request: "FixtureRequest") -> bool:
164 return request.config.getvalue("echo")
167@pytest.fixture(scope="session")
168def anon_db_url(request: "FixtureRequest") -> bool:
169 return request.config.getvalue("anon_db_url")
172@pytest.fixture(scope="session")
173def crate_db_url(request: "FixtureRequest") -> bool:
174 return request.config.getvalue("crate_db_url")
177@pytest.fixture(scope="session")
178def nlp_db_url(request: "FixtureRequest") -> bool:
179 return request.config.getvalue("nlp_db_url")
182@pytest.fixture(scope="session")
183def secret_db_url(request: "FixtureRequest") -> bool:
184 return request.config.getvalue("secret_db_url")
187@pytest.fixture(scope="session")
188def source_db_url(request: "FixtureRequest") -> bool:
189 return request.config.getvalue("source_db_url")
192@pytest.fixture(scope="session")
193def test_db_url(request: "FixtureRequest") -> bool:
194 return request.config.getvalue("test_db_url")
197@pytest.fixture(scope="session")
198def tmpdir_obj(
199 request: "FixtureRequest",
200) -> Generator[tempfile.TemporaryDirectory, None, None]:
201 tmpdir_obj = tempfile.TemporaryDirectory()
203 yield tmpdir_obj
205 tmpdir_obj.cleanup()
208# https://gist.github.com/kissgyorgy/e2365f25a213de44b9a2
209# Author says "no [license], feel free to use it"
210# noinspection PyUnusedLocal
211@pytest.fixture(scope="session")
212def anon_engine(
213 request: "FixtureRequest",
214 anon_db_url: str,
215 create_test_dbs: bool,
216 databases_on_disk: bool,
217 echo: bool,
218) -> Generator["Engine", None, None]:
219 engine_obj = engine(
220 request,
221 anon_db_url,
222 AnonTestBase,
223 ANON_DATABASE_FILENAME,
224 create_test_dbs,
225 databases_on_disk,
226 echo,
227 )
228 yield engine_obj
230 engine_obj.dispose()
233@pytest.fixture(scope="session")
234def secret_engine(
235 request: "FixtureRequest",
236 secret_db_url: str,
237 create_test_dbs: bool,
238 databases_on_disk: bool,
239 echo: bool,
240) -> Generator["Engine", None, None]:
241 engine_obj = engine(
242 request,
243 secret_db_url,
244 SecretBase,
245 SECRET_DATABASE_FILENAME,
246 create_test_dbs,
247 databases_on_disk,
248 echo,
249 )
250 yield engine_obj
252 engine_obj.dispose()
255@pytest.fixture(scope="session")
256def source_engine(
257 request: "FixtureRequest",
258 source_db_url: str,
259 create_test_dbs: bool,
260 databases_on_disk: bool,
261 echo: bool,
262) -> Generator["Engine", None, None]:
263 engine_obj = engine(
264 request,
265 source_db_url,
266 SourceTestBase,
267 SOURCE_DATABASE_FILENAME,
268 create_test_dbs,
269 databases_on_disk,
270 echo,
271 )
272 yield engine_obj
274 engine_obj.dispose()
277def engine(
278 request: "FixtureRequest",
279 db_url: str,
280 base_class: Any,
281 filename: str,
282 create_test_dbs: bool,
283 databases_on_disk: bool,
284 echo: bool,
285) -> Engine:
287 if db_url:
288 return create_engine_from_url(
289 db_url, base_class, create_test_dbs, echo
290 )
292 return create_engine_sqlite(
293 filename, create_test_dbs, echo, databases_on_disk
294 )
297def create_engine_from_url(
298 db_url: str, base_class: Any, create_test_dbs: bool, echo: bool
299) -> Engine:
301 # The database and the user with the given password from db_url
302 # need to exist.
303 # MySQL example:
304 # mysql> CREATE DATABASE <db_name>;
305 # mysql> GRANT ALL PRIVILEGES ON <db_name>.*
306 # TO <db_user>@localhost IDENTIFIED BY '<db_password>';
307 engine = create_engine(db_url, echo=echo, pool_pre_ping=True, future=True)
309 if create_test_dbs:
310 base_class.metadata.drop_all(engine)
312 return engine
315def make_memory_sqlite_engine(echo: bool = False) -> Engine:
316 """
317 Create an SQLAlchemy :class:`Engine` for an in-memory SQLite database.
318 """
319 return create_engine(SQLITE_MEMORY_URL, echo=echo, future=True)
322def make_file_sqlite_engine(filename: str, echo: bool = False) -> Engine:
323 """
324 Create an SQLAlchemy :class:`Engine` for an on-disk SQLite database.
325 """
326 return create_engine(make_sqlite_url(filename), echo=echo, future=True)
329def create_engine_sqlite(
330 filename: str, create_test_dbs: bool, echo: bool, databases_on_disk: bool
331) -> Engine:
332 if create_test_dbs and databases_on_disk:
333 try:
334 os.remove(filename)
335 except OSError:
336 pass
338 if databases_on_disk:
339 engine = make_file_sqlite_engine(filename, echo=echo)
340 else:
341 engine = make_memory_sqlite_engine(echo=echo)
343 event.listen(engine, "connect", set_sqlite_pragma)
345 return engine
348# noinspection PyUnusedLocal
349@pytest.fixture(scope="session")
350def anon_tables(
351 request: "FixtureRequest", anon_engine: "Engine", create_test_dbs: bool
352) -> Generator[None, None, None]:
354 # Not foolproof. Will still need to pass create-test-dbs if the
355 # schema has changed.
356 database_is_empty = not inspect(anon_engine).get_table_names()
358 if create_test_dbs or database_is_empty:
359 AnonTestBase.metadata.create_all(anon_engine)
360 yield
362 # Any post-session clean up would go here
363 # Base.metadata.drop_all(engine)
364 # This would only be useful if we wanted to clean up the database
365 # after running the tests
368# noinspection PyUnusedLocal
369@pytest.fixture(scope="session")
370def secret_tables(
371 request: "FixtureRequest", secret_engine: "Engine", create_test_dbs: bool
372) -> Generator[None, None, None]:
374 # Not foolproof. Will still need to pass create-test-dbs if the
375 # schema has changed.
376 database_is_empty = not inspect(secret_engine).get_table_names()
378 if create_test_dbs or database_is_empty:
379 SecretBase.metadata.create_all(secret_engine)
380 yield
382 # Any post-session clean up would go here
383 # Base.metadata.drop_all(engine)
384 # This would only be useful if we wanted to clean up the database
385 # after running the tests
388# noinspection PyUnusedLocal
389@pytest.fixture(scope="session")
390def source_tables(
391 request: "FixtureRequest", source_engine: "Engine", create_test_dbs: bool
392) -> Generator[None, None, None]:
393 # Not foolproof. Will still need to pass create-test-dbs if the
394 # schema has changed.
395 database_is_empty = not inspect(source_engine).get_table_names()
397 if create_test_dbs or database_is_empty:
398 SourceTestBase.metadata.create_all(source_engine)
399 yield
401 # Any post-session clean up would go here
402 # Base.metadata.drop_all(engine)
403 # This would only be useful if we wanted to clean up the database
404 # after running the tests
407# noinspection PyUnusedLocal
408@pytest.fixture
409def anon_dbsession(
410 request: "FixtureRequest", anon_engine: "Engine", anon_tables: None
411) -> Generator[Session, None, None]:
412 """
413 Returns an sqlalchemy session, and after the test tears down everything
414 properly.
415 """
417 connection = anon_engine.connect()
418 transaction = connection.begin()
419 # use the connection with the already started transaction
420 session = Session(bind=connection)
422 yield session
424 session.close()
425 # If you get the warning: 'sqlalchemy.exc.SAWarning: transaction already
426 # deassociated from Connection', it might be that the code under test needs
427 # to roll back the tranaction and you need to adopt the approach taken by
428 # slow_secret_dbsession() where the tables are created and dropped for each
429 # test.
430 transaction.rollback()
431 # put back the connection to the connection pool
432 connection.close()
435# noinspection PyUnusedLocal
436@pytest.fixture
437def secret_dbsession(
438 request: "FixtureRequest", secret_engine: "Engine", secret_tables: None
439) -> Generator[Session, None, None]:
440 """
441 Returns an sqlalchemy session, and after the test tears down everything
442 properly.
443 """
445 connection = secret_engine.connect()
446 transaction = connection.begin()
447 # use the connection with the already started transaction
448 session = Session(bind=connection)
450 yield session
452 session.close()
453 # If you get the warning: 'sqlalchemy.exc.SAWarning: transaction already
454 # deassociated from Connection', it might be that the code under test needs
455 # to roll back the tranaction and you need to adopt the approach taken by
456 # slow_secret_dbsession() where the tables are created and dropped for each
457 # test.
458 transaction.rollback()
459 # put back the connection to the connection pool
460 connection.close()
463# noinspection PyUnusedLocal
464@pytest.fixture
465def source_dbsession(
466 request: "FixtureRequest", source_engine: "Engine", source_tables: None
467) -> Generator[Session, None, None]:
468 """
469 Returns an sqlalchemy session, and after the test tears down everything
470 properly.
471 """
473 connection = source_engine.connect()
474 transaction = connection.begin()
475 # use the connection with the already started transaction
476 session = Session(bind=connection, future=True)
478 yield session
480 session.close()
481 # If you get the warning: 'sqlalchemy.exc.SAWarning: transaction already
482 # deassociated from Connection', it might be that the code under test needs
483 # to roll back the tranaction and you need to adopt the approach taken by
484 # slow_secret_dbsession() where the tables are created and dropped for each
485 # test.
486 transaction.rollback()
487 # put back the connection to the connection pool
488 connection.close()
491@pytest.fixture
492def setup(
493 request: "FixtureRequest",
494 anon_engine: "Engine",
495 secret_engine: "Engine",
496 source_engine: "Engine",
497 databases_on_disk: bool,
498 anon_dbsession: Session,
499 secret_dbsession: Session,
500 source_dbsession: Session,
501 tmpdir_obj: tempfile.TemporaryDirectory,
502) -> None:
504 # Pytest prefers function-based tests over unittest.TestCase subclasses and
505 # methods, but it still supports the latter perfectly well.
506 # We use this fixture in testing/classes.py to store these values into
507 # DatabaseTestCase and its descendants.
508 request.cls.anon_engine = anon_engine
509 request.cls.secret_engine = secret_engine
510 request.cls.source_engine = source_engine
511 request.cls.databases_on_disk = databases_on_disk
512 request.cls.anon_dbsession = anon_dbsession
513 request.cls.secret_dbsession = secret_dbsession
514 request.cls.source_dbsession = source_dbsession
515 request.cls.tmpdir_obj = tmpdir_obj
516 request.cls.anon_db_filename = ANON_DATABASE_FILENAME
517 request.cls.secret_db_filename = SECRET_DATABASE_FILENAME
518 request.cls.source_db_filename = SOURCE_DATABASE_FILENAME
521@pytest.fixture
522def slow_secret_tables(
523 request: "FixtureRequest", secret_engine: "Engine", create_test_dbs: bool
524) -> Generator[None, None, None]:
525 # This has the default 'function' scope so tables are created at the
526 # beginning of each test and dropped at the end. This is for the case where
527 # we need to roll back a transaction in the code, so the trick used by
528 # secret_tables() to run each test in a transaction won't work. Potentially
529 # expensive if there are a lot of tables.
530 SecretBase.metadata.create_all(secret_engine)
532 yield
534 SecretBase.metadata.drop_all(secret_engine)
536 # in case another test that uses the normal secret_tables runs next.
537 SecretBase.metadata.create_all(secret_engine)
540# noinspection PyUnusedLocal
541@pytest.fixture
542def slow_secret_dbsession(
543 request: "FixtureRequest",
544 secret_engine: "Engine",
545 slow_secret_tables: None,
546) -> Generator[Session, None, None]:
547 """
548 As with secret_dbsession() but we don't start a transaction for each test.
549 Used in conjunction with slow_secret_tables() where the tables are dropped
550 and recreated with each test.
551 """
553 connection = secret_engine.connect()
555 session = Session(bind=connection)
557 yield session
559 session.close()
560 connection.close()
563@pytest.fixture
564def slow_secret_setup(
565 request: "FixtureRequest",
566 anon_engine: "Engine",
567 secret_engine: "Engine",
568 source_engine: "Engine",
569 databases_on_disk: bool,
570 anon_dbsession: Session,
571 slow_secret_dbsession: Session,
572 source_dbsession: Session,
573 tmpdir_obj: tempfile.TemporaryDirectory,
574) -> None:
575 request.cls.anon_engine = anon_engine
576 request.cls.secret_engine = secret_engine
577 request.cls.source_engine = source_engine
578 request.cls.databases_on_disk = databases_on_disk
579 request.cls.anon_dbsession = anon_dbsession
580 request.cls.secret_dbsession = slow_secret_dbsession
581 request.cls.source_dbsession = source_dbsession
582 request.cls.tmpdir_obj = tmpdir_obj
583 request.cls.anon_db_filename = ANON_DATABASE_FILENAME
584 request.cls.secret_db_filename = SECRET_DATABASE_FILENAME
585 request.cls.source_db_filename = SOURCE_DATABASE_FILENAME