Plain is headed towards 1.0! Subscribe for development updates →

 1import os
 2import sys
 3from pathlib import Path
 4
 5from plain.models.backends.base.creation import BaseDatabaseCreation
 6
 7
 8class DatabaseCreation(BaseDatabaseCreation):
 9    @staticmethod
10    def is_in_memory_db(database_name):
11        return not isinstance(database_name, Path) and (
12            database_name == ":memory:" or "mode=memory" in database_name
13        )
14
15    def _get_test_db_name(self, prefix=""):
16        raw_name = self.connection.settings_dict["TEST"]["NAME"] or ":memory:"
17        # Special in-memory case
18        if raw_name == ":memory:":
19            return "file:memorydb?mode=memory&cache=shared"
20
21        test_database_name = raw_name
22
23        if prefix:
24            test_database_name = f"{prefix}_{test_database_name}"
25
26        return test_database_name
27
28    def _create_test_db(self, *, test_database_name, verbosity, autoclobber):
29        """
30        Internal implementation - delete existing SQLite test DB file if needed.
31        """
32        if not self.is_in_memory_db(test_database_name):
33            # Erase the old test database file.
34            if verbosity >= 1:
35                self.log(f"Destroying old test database '{test_database_name}'...")
36            if os.access(test_database_name, os.F_OK):
37                if not autoclobber:
38                    confirm = input(
39                        "Type 'yes' if you would like to try deleting the test "
40                        f"database '{test_database_name}', or 'no' to cancel: "
41                    )
42                if autoclobber or confirm == "yes":
43                    try:
44                        os.remove(test_database_name)
45                    except Exception as e:
46                        self.log(f"Got an error deleting the old test database: {e}")
47                        sys.exit(2)
48                else:
49                    self.log("Tests cancelled.")
50                    sys.exit(1)
51        return test_database_name
52
53    def _destroy_test_db(self, test_database_name, verbosity):
54        if test_database_name and not self.is_in_memory_db(test_database_name):
55            # Remove the SQLite database file
56            os.remove(test_database_name)
57
58    def test_db_signature(self, prefix=""):
59        """
60        Return a tuple that uniquely identifies a test database.
61
62        This takes into account the special cases of ":memory:" and "" for
63        SQLite since the databases will be distinct despite having the same
64        TEST NAME. See https://www.sqlite.org/inmemorydb.html
65        """
66        test_database_name = self._get_test_db_name(prefix)
67        sig = [self.connection.settings_dict["NAME"]]
68        if self.is_in_memory_db(test_database_name):
69            sig.append(":memory:")
70        else:
71            sig.append(test_database_name)
72        return tuple(sig)