Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/db/backends/postgresql/creation.py: 42%

1import sys 

2 

3from psycopg2 import errorcodes 

4 

5from django.core.exceptions import ImproperlyConfigured 

6from django.db.backends.base.creation import BaseDatabaseCreation 

7from django.db.backends.utils import strip_quotes 

8 

9 

10class DatabaseCreation(BaseDatabaseCreation): 

11 def _quote_name(self, name): 

12 return self.connection.ops.quote_name(name) 

13 

14 def _get_database_create_suffix(self, encoding=None, template=None): 

15 suffix = "" 

16 if encoding: 16 ↛ 17line 16 didn't jump to line 17, because the condition on line 16 was never true

17 suffix += " ENCODING '{}'".format(encoding) 

18 if template: 18 ↛ 19line 18 didn't jump to line 19, because the condition on line 18 was never true

19 suffix += " TEMPLATE {}".format(self._quote_name(template)) 

20 return suffix and "WITH" + suffix 

21 

22 def sql_table_creation_suffix(self): 

23 test_settings = self.connection.settings_dict["TEST"] 

24 if test_settings.get("COLLATION") is not None: 24 ↛ 25line 24 didn't jump to line 25, because the condition on line 24 was never true

25 raise ImproperlyConfigured( 

26 "PostgreSQL does not support collation setting at database " 

27 "creation time." 

28 ) 

29 return self._get_database_create_suffix( 

30 encoding=test_settings["CHARSET"], 

31 template=test_settings.get("TEMPLATE"), 

32 ) 

33 

34 def _database_exists(self, cursor, database_name): 

35 cursor.execute( 

36 "SELECT 1 FROM pg_catalog.pg_database WHERE datname = %s", 

37 [strip_quotes(database_name)], 

38 ) 

39 return cursor.fetchone() is not None 

40 

41 def _execute_create_test_db(self, cursor, parameters, keepdb=False): 

42 try: 

43 if keepdb and self._database_exists(cursor, parameters["dbname"]): 43 ↛ 46line 43 didn't jump to line 46, because the condition on line 43 was never true

44 # If the database should be kept and it already exists, don't 

45 # try to create a new one. 

46 return 

47 super()._execute_create_test_db(cursor, parameters, keepdb) 

48 except Exception as e: 

49 if getattr(e.__cause__, "pgcode", "") != errorcodes.DUPLICATE_DATABASE: 

50 # All errors except "database already exists" cancel tests. 

51 self.log("Got an error creating the test database: %s" % e) 

52 sys.exit(2) 

53 elif not keepdb: 

54 # If the database should be kept, ignore "database already 

55 # exists". 

56 raise 

57 

58 def _clone_test_db(self, suffix, verbosity, keepdb=False): 

59 # CREATE DATABASE ... WITH TEMPLATE ... requires closing connections 

60 # to the template database. 

61 self.connection.close() 

62 

63 source_database_name = self.connection.settings_dict["NAME"] 

64 target_database_name = self.get_test_db_clone_settings(suffix)["NAME"] 

65 test_db_params = { 

66 "dbname": self._quote_name(target_database_name), 

67 "suffix": self._get_database_create_suffix(template=source_database_name), 

68 } 

69 with self._nodb_cursor() as cursor: 

70 try: 

71 self._execute_create_test_db(cursor, test_db_params, keepdb) 

72 except Exception: 

73 try: 

74 if verbosity >= 1: 

75 self.log( 

76 "Destroying old test database for alias %s..." 

77 % ( 

78 self._get_database_display_str( 

79 verbosity, target_database_name 

80 ), 

81 ) 

82 ) 

83 cursor.execute("DROP DATABASE %(dbname)s" % test_db_params) 

84 self._execute_create_test_db(cursor, test_db_params, keepdb) 

85 except Exception as e: 

86 self.log("Got an error cloning the test database: %s" % e) 

87 sys.exit(2)