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

184 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-08-27 10:34 -0500

1""" 

2crate_anon/crateweb/research/tests/models_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 models.py. 

27 

28""" 

29 

30from unittest import TestCase 

31 

32from cardinal_pythonlib.hash import hash64 

33 

34import pytest 

35 

36from crate_anon.crateweb.research.models import PatientMultiQuery 

37from crate_anon.crateweb.research.tests.factories import ( 

38 PatientExplorerFactory, 

39 QueryFactory, 

40 SitewideQueryFactory, 

41 UserFactory, 

42) 

43 

44 

45class QueryTest(TestCase): 

46 @pytest.mark.django_db 

47 def test_sql_hash_saved_when_update_fields_none(self) -> None: 

48 query = QueryFactory() 

49 query.save() 

50 

51 sql = "SELECT foo from bar" 

52 query.sql = sql 

53 query.save() 

54 query.refresh_from_db() 

55 

56 self.assertEqual(query.sql_hash, hash64(sql)) 

57 

58 @pytest.mark.django_db 

59 def test_sql_hash_saved_when_sql_in_update_fields(self) -> None: 

60 """ 

61 Since Django 4.2 custom save() methods should update the update_fields 

62 argument. 

63 """ 

64 query = QueryFactory() 

65 query.save() 

66 

67 sql = "SELECT foo from bar" 

68 query.sql = sql 

69 query.save(update_fields=["sql"]) 

70 query.refresh_from_db() 

71 

72 self.assertEqual(query.sql_hash, hash64(sql)) 

73 

74 @pytest.mark.django_db 

75 def test_sql_hash_not_saved_when_sql_not_in_update_fields(self) -> None: 

76 """ 

77 Since Django 4.2 custom save() methods should update the update_fields 

78 argument. 

79 """ 

80 old_sql = "SELECT foo from old" 

81 query = QueryFactory(sql=old_sql, raw=False) 

82 query.save() 

83 

84 new_sql = "SELECT foo from new" 

85 query.sql = new_sql 

86 query.raw = True 

87 query.save(update_fields=["raw"]) 

88 query.refresh_from_db() 

89 

90 self.assertEqual(query.sql_hash, hash64(old_sql)) 

91 

92 @pytest.mark.django_db 

93 def test_formatted_sql_saved_when_update_fields_none(self) -> None: 

94 """ 

95 Since Django 4.2 custom save() methods should update the update_fields 

96 argument. 

97 """ 

98 query = QueryFactory() 

99 query.save() 

100 

101 sql = "SELECT foo from bar" 

102 query.sql = sql 

103 query.save() 

104 query.refresh_from_db() 

105 

106 for word in sql.split(): 

107 self.assertIn(word, query.formatted_sql) 

108 

109 @pytest.mark.django_db 

110 def test_formatted_sql_saved_when_sql_in_update_fields(self) -> None: 

111 """ 

112 Since Django 4.2 custom save() methods should update the update_fields 

113 argument. 

114 """ 

115 query = QueryFactory() 

116 query.save() 

117 

118 sql = "SELECT foo from bar" 

119 query.sql = sql 

120 query.save(update_fields=["sql"]) 

121 query.refresh_from_db() 

122 

123 for word in sql.split(): 

124 self.assertIn(word, query.formatted_sql) 

125 

126 @pytest.mark.django_db 

127 def test_formatted_sql_not_saved_when_sql_not_in_update_fields( 

128 self, 

129 ) -> None: 

130 """ 

131 Since Django 4.2 custom save() methods should update the update_fields 

132 argument. 

133 """ 

134 old_sql = "SELECT foo from old" 

135 query = QueryFactory(raw=False, sql=old_sql) 

136 query.save() 

137 

138 new_sql = "SELECT foo from new" 

139 query.sql = new_sql 

140 query.raw = True 

141 query.save(update_fields=["raw"]) 

142 query.refresh_from_db() 

143 

144 for word in old_sql.split(): 

145 self.assertIn(word, query.formatted_sql) 

146 

147 @pytest.mark.django_db 

148 def test_active_query_changes_when_object_saved(self) -> None: 

149 user = UserFactory() 

150 

151 query1 = QueryFactory(user=user, active=True) 

152 query1.save() 

153 query1.refresh_from_db() 

154 self.assertTrue(query1.active) 

155 

156 query2 = QueryFactory(user=user, active=True) 

157 query2.save() 

158 query2.refresh_from_db() 

159 query1.refresh_from_db() 

160 

161 self.assertTrue(query2.active) 

162 self.assertFalse(query1.active) 

163 

164 

165class SitewideQueryTest(TestCase): 

166 @pytest.mark.django_db 

167 def test_sql_hash_saved_when_update_fields_none(self) -> None: 

168 query = SitewideQueryFactory() 

169 query.save() 

170 

171 sql = "SELECT foo from bar" 

172 query.sql = sql 

173 query.save() 

174 query.refresh_from_db() 

175 

176 self.assertEqual(query.sql_hash, hash64(sql)) 

177 

178 @pytest.mark.django_db 

179 def test_sql_hash_saved_when_sql_in_update_fields(self) -> None: 

180 """ 

181 Since Django 4.2 custom save() methods should update the update_fields 

182 argument. 

183 """ 

184 query = SitewideQueryFactory() 

185 query.save() 

186 

187 sql = "SELECT foo from bar" 

188 query.sql = sql 

189 query.save(update_fields=["sql"]) 

190 query.refresh_from_db() 

191 

192 self.assertEqual(query.sql_hash, hash64(sql)) 

193 

194 @pytest.mark.django_db 

195 def test_sql_hash_not_saved_when_sql_not_in_update_fields(self) -> None: 

196 """ 

197 Since Django 4.2 custom save() methods should update the update_fields 

198 argument. 

199 """ 

200 old_sql = "SELECT foo from old" 

201 query = SitewideQueryFactory(sql=old_sql, raw=False) 

202 query.save() 

203 

204 new_sql = "SELECT foo from new" 

205 query.sql = new_sql 

206 query.raw = True 

207 query.save(update_fields=["raw"]) 

208 query.refresh_from_db() 

209 

210 self.assertEqual(query.sql_hash, hash64(old_sql)) 

211 

212 @pytest.mark.django_db 

213 def test_formatted_sql_saved_when_update_fields_none(self) -> None: 

214 """ 

215 Since Django 4.2 custom save() methods should update the update_fields 

216 argument. 

217 """ 

218 query = SitewideQueryFactory() 

219 query.save() 

220 

221 sql = "SELECT foo from bar" 

222 query.sql = sql 

223 query.save() 

224 query.refresh_from_db() 

225 

226 for word in sql.split(): 

227 self.assertIn(word, query.formatted_sql) 

228 

229 @pytest.mark.django_db 

230 def test_formatted_sql_saved_when_sql_in_update_fields(self) -> None: 

231 """ 

232 Since Django 4.2 custom save() methods should update the update_fields 

233 argument. 

234 """ 

235 query = SitewideQueryFactory() 

236 query.save() 

237 

238 sql = "SELECT foo from bar" 

239 query.sql = sql 

240 query.save(update_fields=["sql"]) 

241 query.refresh_from_db() 

242 

243 for word in sql.split(): 

244 self.assertIn(word, query.formatted_sql) 

245 

246 @pytest.mark.django_db 

247 def test_formatted_sql_not_saved_when_sql_not_in_update_fields( 

248 self, 

249 ) -> None: 

250 """ 

251 Since Django 4.2 custom save() methods should update the update_fields 

252 argument. 

253 """ 

254 old_sql = "SELECT foo from old" 

255 query = SitewideQueryFactory(raw=False, sql=old_sql) 

256 query.save() 

257 

258 new_sql = "SELECT foo from new" 

259 query.sql = new_sql 

260 query.raw = True 

261 query.save(update_fields=["raw"]) 

262 query.refresh_from_db() 

263 

264 for word in old_sql.split(): 

265 self.assertIn(word, query.formatted_sql) 

266 

267 

268class PatientExplorerTest(TestCase): 

269 @pytest.mark.django_db 

270 def test_pmq_hash_saved_when_update_fields_none(self) -> None: 

271 explorer = PatientExplorerFactory() 

272 explorer.save() 

273 

274 explorer.patient_multiquery = PatientMultiQuery() 

275 explorer.save() 

276 explorer.refresh_from_db() 

277 

278 self.assertEqual(explorer.pmq_hash, explorer.patient_multiquery.hash64) 

279 

280 @pytest.mark.django_db 

281 def test_pmq_hash_saved_when_patient_multiquery_in_update_fields( 

282 self, 

283 ) -> None: 

284 """ 

285 Since Django 4.2 custom save() methods should update the update_fields 

286 argument. 

287 """ 

288 explorer = PatientExplorerFactory() 

289 explorer.save() 

290 

291 explorer.patient_multiquery = PatientMultiQuery( 

292 manual_patient_id_query="SELECT foo FROM bar" 

293 ) 

294 explorer.save(update_fields=["patient_multiquery"]) 

295 explorer.refresh_from_db() 

296 

297 self.assertEqual(explorer.pmq_hash, explorer.patient_multiquery.hash64) 

298 

299 @pytest.mark.django_db 

300 def test_pmq_hash_not_saved_when_pmq_not_in_update_fields(self) -> None: 

301 """ 

302 Since Django 4.2 custom save() methods should update the update_fields 

303 argument. 

304 """ 

305 old_multiquery = PatientMultiQuery( 

306 manual_patient_id_query="SELECT foo FROM old" 

307 ) 

308 explorer = PatientExplorerFactory( 

309 patient_multiquery=old_multiquery, audited=False 

310 ) 

311 explorer.save() 

312 

313 new_multiquery = PatientMultiQuery( 

314 manual_patient_id_query="SELECT foo FROM new" 

315 ) 

316 explorer.patient_multiquery = new_multiquery 

317 explorer.audited = True 

318 

319 explorer.save(update_fields=["audited"]) 

320 explorer.refresh_from_db() 

321 

322 self.assertNotEqual(old_multiquery.hash64, new_multiquery.hash64) 

323 self.assertEqual(explorer.pmq_hash, old_multiquery.hash64) 

324 

325 @pytest.mark.django_db 

326 def test_active_patient_explorer_changes_when_object_saved(self) -> None: 

327 user = UserFactory() 

328 

329 explorer1 = PatientExplorerFactory(user=user, active=True) 

330 explorer1.save() 

331 explorer1.refresh_from_db() 

332 self.assertTrue(explorer1.active) 

333 

334 explorer2 = PatientExplorerFactory(user=user, active=True) 

335 explorer2.save() 

336 explorer2.refresh_from_db() 

337 explorer1.refresh_from_db() 

338 

339 self.assertTrue(explorer2.active) 

340 self.assertFalse(explorer1.active)