1import re
  2
  3import pytest
  4
  5from plain.signals import request_finished, request_started
  6
  7from .. import transaction
  8from ..backends.base.base import BaseDatabaseWrapper
  9from ..db import close_old_connections, db_connection
 10from .utils import (
 11    setup_database,
 12    teardown_database,
 13)
 14
 15
 16@pytest.fixture(autouse=True)
 17def _db_disabled():
 18    """
 19    Every test should use this fixture by default to prevent
 20    access to the normal database.
 21    """
 22
 23    def cursor_disabled(self):
 24        pytest.fail("Database access not allowed without the `db` fixture")
 25
 26    BaseDatabaseWrapper._enabled_cursor = BaseDatabaseWrapper.cursor
 27    BaseDatabaseWrapper.cursor = cursor_disabled
 28
 29    yield
 30
 31    BaseDatabaseWrapper.cursor = BaseDatabaseWrapper._enabled_cursor
 32
 33
 34@pytest.fixture(scope="session")
 35def setup_db(request):
 36    """
 37    This fixture is called automatically by `db`,
 38    so a test database will only be setup if the `db` fixture is used.
 39    """
 40    verbosity = request.config.option.verbose
 41
 42    # Set up the test db across the entire session
 43    _old_db_name = setup_database(verbosity=verbosity)
 44
 45    # Keep connections open during request client / testing
 46    request_started.disconnect(close_old_connections)
 47    request_finished.disconnect(close_old_connections)
 48
 49    yield
 50
 51    # Put the signals back...
 52    request_started.connect(close_old_connections)
 53    request_finished.connect(close_old_connections)
 54
 55    # When the test session is done, tear down the test db
 56    teardown_database(_old_db_name, verbosity=verbosity)
 57
 58
 59@pytest.fixture
 60def db(setup_db, request):
 61    if "isolated_db" in request.fixturenames:
 62        pytest.fail("The 'db' and 'isolated_db' fixtures cannot be used together")
 63    # Set .cursor() back to the original implementation to unblock it
 64    BaseDatabaseWrapper.cursor = BaseDatabaseWrapper._enabled_cursor
 65
 66    if not db_connection.features.supports_transactions:
 67        pytest.fail("Database does not support transactions")
 68
 69    atomic = transaction.atomic()
 70    atomic._from_testcase = True  # TODO remove this somehow?
 71    atomic.__enter__()
 72
 73    yield
 74
 75    if (
 76        db_connection.features.can_defer_constraint_checks
 77        and not db_connection.needs_rollback
 78        and db_connection.is_usable()
 79    ):
 80        db_connection.check_constraints()
 81
 82    db_connection.set_rollback(True)
 83    atomic.__exit__(None, None, None)
 84
 85    db_connection.close()
 86
 87
 88@pytest.fixture
 89def isolated_db(request):
 90    """
 91    Create and destroy a unique test database for each test, using a prefix
 92    derived from the test function name to ensure isolation from the default
 93    test database.
 94    """
 95    if "db" in request.fixturenames:
 96        pytest.fail("The 'db' and 'isolated_db' fixtures cannot be used together")
 97    # Set .cursor() back to the original implementation to unblock it
 98    BaseDatabaseWrapper.cursor = BaseDatabaseWrapper._enabled_cursor
 99
100    verbosity = 1
101
102    # Derive a safe prefix from the test function name
103    raw_name = request.node.name
104    prefix = re.sub(r"[^0-9A-Za-z_]+", "_", raw_name)
105
106    # Set up a fresh test database for this test, using the prefix
107    _old_db_name = setup_database(verbosity=verbosity, prefix=prefix)
108
109    yield
110
111    # Tear down the test database created for this test
112    teardown_database(_old_db_name, verbosity=verbosity)