Coverage for crateweb/research/tests/research_db_info_tests.py: 100%

1""" 

2crate_anon/crateweb/research/tests/research_db_info_tests.py 

3 

4=============================================================================== 

5 

6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry. 

7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk). 

8 

9 This file is part of CRATE. 

10 

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. 

15 

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. 

20 

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/>. 

23 

24=============================================================================== 

25 

26Test research_db_info.py. 

27 

28""" 

29 

30# ============================================================================= 

31# Imports 

32# ============================================================================= 

33 

34import logging 

35import os.path 

36from tempfile import TemporaryDirectory 

37 

38from cardinal_pythonlib.dbfunc import dictfetchall 

39from cardinal_pythonlib.sql.sql_grammar import SqlGrammar 

40from django.db import connections 

41from django.test.testcases import TestCase # inherits from unittest.TestCase 

42 

43from crate_anon.crateweb.config.constants import ResearchDbInfoKeys as RDIKeys 

44from crate_anon.crateweb.core.constants import ( 

45 DJANGO_DEFAULT_CONNECTION, 

46 RESEARCH_DB_CONNECTION_NAME, 

47) 

48from crate_anon.crateweb.research.research_db_info import ( 

49 SingleResearchDatabase, 

50 ResearchDatabaseInfo, 

51) 

52 

53log = logging.getLogger(__name__) 

54 

55 

56# ============================================================================= 

57# Unit tests 

58# ============================================================================= 

59 

60 

61class ResearchDBInfoTests(TestCase): 

62 databases = {DJANGO_DEFAULT_CONNECTION, RESEARCH_DB_CONNECTION_NAME} 

63 # ... or the test framework will produce this: 

64 # 

65 # django.test.testcases.DatabaseOperationForbidden: Database queries to 

66 # 'research' are not allowed in this test. Add 'research' to 

67 # research_db_info_tests.ResearchDBInfoTests.databases to ensure proper 

68 # test isolation and silence this failure. 

69 # 

70 # It is checked by a classmethod, not an instance. 

71 

72 def setUp(self): 

73 super().setUp() 

74 

75 # crate_anon.common.constants.RUNNING_WITHOUT_CONFIG = True 

76 

77 # If we have two SQLite in-memory database (with name = ":memory:"), 

78 # they appear to be the same database. But equally, if you use a local 

79 # temporary directory, nothing is created on disk; so presumably the 

80 # Django test framework is intercepting everything? 

81 self.tempdir = TemporaryDirectory() # will be deleted on destruction 

82 self.settings( 

83 DATABASES={ 

84 DJANGO_DEFAULT_CONNECTION: { 

85 "ENGINE": "django.db.backends.sqlite3", 

86 "NAME": os.path.join(self.tempdir.name, "main.sqlite3"), 

87 }, 

88 RESEARCH_DB_CONNECTION_NAME: { 

89 "ENGINE": "django.db.backends.sqlite3", 

90 "NAME": os.path.join( 

91 self.tempdir.name, "research.sqlite3" 

92 ), 

93 }, 

94 }, 

95 # DEBUG=True, 

96 RESEARCH_DB_INFO=[ 

97 { 

98 RDIKeys.NAME: "research", 

99 RDIKeys.DESCRIPTION: "Demo research database", 

100 RDIKeys.DATABASE: "", 

101 RDIKeys.SCHEMA: "research", 

102 RDIKeys.PID_PSEUDO_FIELD: "pid", 

103 RDIKeys.MPID_PSEUDO_FIELD: "mpid", 

104 RDIKeys.TRID_FIELD: "trid", 

105 RDIKeys.RID_FIELD: "brcid", 

106 RDIKeys.RID_FAMILY: 1, 

107 RDIKeys.MRID_TABLE: "patients", 

108 RDIKeys.MRID_FIELD: "nhshash", 

109 RDIKeys.PID_DESCRIPTION: "Patient ID", 

110 RDIKeys.MPID_DESCRIPTION: "Master patient ID", 

111 RDIKeys.RID_DESCRIPTION: "Research ID", 

112 RDIKeys.MRID_DESCRIPTION: "Master research ID", 

113 RDIKeys.TRID_DESCRIPTION: "Transient research ID", 

114 RDIKeys.SECRET_LOOKUP_DB: "secret", 

115 RDIKeys.DATE_FIELDS_BY_TABLE: {}, 

116 RDIKeys.DEFAULT_DATE_FIELDS: [], 

117 RDIKeys.UPDATE_DATE_FIELD: "_when_fetched_utc", 

118 }, 

119 ], 

120 ) 

121 self.mainconn = connections[DJANGO_DEFAULT_CONNECTION] 

122 self.resconn = connections[RESEARCH_DB_CONNECTION_NAME] 

123 self.grammar = SqlGrammar() 

124 with self.resconn.cursor() as cursor: 

125 cursor.execute("CREATE TABLE t (a INT, b INT)") 

126 cursor.execute("INSERT INTO t (a, b) VALUES (1, 101)") 

127 cursor.execute("INSERT INTO t (a, b) VALUES (2, 102)") 

128 cursor.execute("COMMIT") 

129 

130 def tearDown(self) -> None: 

131 with self.resconn.cursor() as cursor: 

132 cursor.execute("DROP TABLE t") 

133 # Otherwise, you can run one test, but if you run two, you get: 

134 # 

135 # django.db.transaction.TransactionManagementError: An error occurred 

136 # in the current transaction. You can't execute queries until the end 

137 # of the 'atomic' block. 

138 # 

139 # ... no - still the problem! 

140 # Hack: combine the tests. 

141 

142 def test_django_dummy_database_and_sqlite_schema_reader(self) -> None: 

143 with self.resconn.cursor() as cursor: 

144 cursor.execute("SELECT * FROM t") 

145 results = dictfetchall(cursor) 

146 self.assertEqual(len(results), 2) 

147 self.assertEqual(results[0], dict(a=1, b=101)) 

148 self.assertEqual(results[1], dict(a=2, b=102)) 

149 

150 rdbi = ResearchDatabaseInfo(running_without_config=True) 

151 srd = SingleResearchDatabase( 

152 index=0, 

153 grammar=self.grammar, 

154 rdb_info=rdbi, 

155 connection=self.resconn, 

156 ) 

157 col_info_list = srd.schema_infodictlist # will read the database 

158 # Unfortunately it will read all the Django tables too (see above). 

159 table_t_cols = [c for c in col_info_list if c["table_name"] == "t"] 

160 self.assertTrue(len(table_t_cols) == 2) 

161 row0 = table_t_cols[0] 

162 self.assertEqual(row0["column_name"], "a") 

163 self.assertEqual(row0["column_type"], "INT") 

164 row1 = table_t_cols[1] 

165 self.assertEqual(row1["column_name"], "b") 

166 self.assertEqual(row1["column_type"], "INT")