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)