Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1#!/usr/bin/env python 

2 

3"""camcops_server/cc_modules/tests/cc_fhir_tests.py 

4 

5=============================================================================== 

6 

7 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com). 

8 

9 This file is part of CamCOPS. 

10 

11 CamCOPS 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 CamCOPS 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 CamCOPS. If not, see <http://www.gnu.org/licenses/>. 

23 

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

25 

26""" 

27 

28import datetime 

29import json 

30from typing import Dict 

31from unittest import mock 

32 

33from camcops_server.cc_modules.cc_exportmodels import ( 

34 ExportedTask, 

35 ExportedTaskFhir, 

36) 

37 

38from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient 

39from camcops_server.cc_modules.cc_exportrecipientinfo import ExportRecipientInfo 

40from camcops_server.cc_modules.cc_fhir import FhirTaskExporter 

41from camcops_server.cc_modules.cc_unittest import DemoDatabaseTestCase 

42from camcops_server.tasks.phq9 import Phq9 

43 

44 

45# ============================================================================= 

46# Integration testing 

47# ============================================================================= 

48 

49 

50class MockFhirTaskExporter(FhirTaskExporter): 

51 pass 

52 

53 

54class MockFhirResponse(mock.Mock): 

55 def __init__(self, response_json: Dict): 

56 super().__init__( 

57 text=json.dumps(response_json), 

58 json=mock.Mock(return_value=response_json) 

59 ) 

60 

61 

62class FhirExportTestCase(DemoDatabaseTestCase): 

63 def setUp(self) -> None: 

64 super().setUp() 

65 recipientinfo = ExportRecipientInfo() 

66 

67 self.recipient = ExportRecipient(recipientinfo) 

68 self.recipient.primary_idnum = self.rio_iddef.which_idnum 

69 self.recipient.fhir_api_url = "http://www.example.com/fhir" 

70 

71 # auto increment doesn't work for BigInteger with SQLite 

72 self.recipient.id = 1 

73 self.recipient.recipient_name = "test" 

74 

75 

76class FhirTaskExporterTests(FhirExportTestCase): 

77 def create_tasks(self) -> None: 

78 self.patient = self.create_patient( 

79 forename="Gwendolyn", 

80 surname="Ryann", 

81 sex="F" 

82 ) 

83 self.patient_nhs = self.create_patient_idnum( 

84 patient_id=self.patient.id, 

85 which_idnum=self.nhs_iddef.which_idnum, 

86 idnum_value=8879736213 

87 ) 

88 self.patient_rio = self.create_patient_idnum( 

89 patient_id=self.patient.id, 

90 which_idnum=self.rio_iddef.which_idnum, 

91 idnum_value=12345 

92 ) 

93 

94 self.task = Phq9() 

95 self.apply_standard_task_fields(self.task) 

96 self.task.q1 = 0 

97 self.task.q2 = 1 

98 self.task.q3 = 2 

99 self.task.q4 = 3 

100 self.task.q5 = 0 

101 self.task.q6 = 1 

102 self.task.q7 = 2 

103 self.task.q8 = 3 

104 self.task.q9 = 0 

105 self.task.q10 = 3 

106 self.task.patient_id = self.patient.id 

107 self.task.save_with_next_available_id(self.req, self.patient._device_id) 

108 self.dbsession.commit() 

109 

110 def test_patient_exported_with_phq9(self) -> None: 

111 exported_task = ExportedTask(task=self.task, recipient=self.recipient) 

112 exported_task_fhir = ExportedTaskFhir(exported_task) 

113 

114 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

115 

116 response_json = { 

117 'type': 'transaction-response', 

118 } 

119 

120 with mock.patch.object( 

121 exporter.client.server, "post_json", 

122 return_value=MockFhirResponse(response_json)) as mock_post: 

123 exporter.export_task() 

124 

125 args, kwargs = mock_post.call_args 

126 

127 sent_json = args[1] 

128 

129 self.assertEqual(sent_json["resourceType"], "Bundle") 

130 self.assertEqual(sent_json["type"], "transaction") 

131 

132 patient = sent_json["entry"][0]["resource"] 

133 self.assertEqual(patient["resourceType"], "Patient") 

134 

135 identifier = patient["identifier"] 

136 which_idnum = self.patient_rio.which_idnum 

137 idnum_value = self.patient_rio.idnum_value 

138 

139 iddef_url = f"http://127.0.0.1:8000/fhir_patient_id/{which_idnum}" 

140 

141 self.assertEqual(identifier[0]["system"], iddef_url) 

142 self.assertEqual(identifier[0]["value"], str(idnum_value)) 

143 

144 self.assertEqual(patient["name"][0]["family"], self.patient.surname) 

145 self.assertEqual(patient["name"][0]["given"], [self.patient.forename]) 

146 self.assertEqual(patient["gender"], "female") 

147 

148 request = sent_json["entry"][0]["request"] 

149 self.assertEqual(request["method"], "POST") 

150 self.assertEqual(request["url"], "Patient") 

151 self.assertEqual( 

152 request["ifNoneExist"], 

153 (f"identifier={iddef_url}|{idnum_value}") 

154 ) 

155 

156 def test_questionnaire_exported_with_phq9(self) -> None: 

157 exported_task = ExportedTask(task=self.task, recipient=self.recipient) 

158 exported_task_fhir = ExportedTaskFhir(exported_task) 

159 

160 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

161 

162 response_json = { 

163 'type': 'transaction-response', 

164 } 

165 

166 with mock.patch.object( 

167 exporter.client.server, "post_json", 

168 return_value=MockFhirResponse(response_json)) as mock_post: 

169 exporter.export_task() 

170 

171 args, kwargs = mock_post.call_args 

172 

173 sent_json = args[1] 

174 

175 questionnaire = sent_json["entry"][1]["resource"] 

176 self.assertEqual(questionnaire["resourceType"], "Questionnaire") 

177 self.assertEqual(questionnaire["status"], "active") 

178 

179 identifier = questionnaire["identifier"] 

180 

181 questionnaire_url = "http://127.0.0.1:8000/fhir_questionnaire_id" 

182 self.assertEqual(identifier[0]["system"], questionnaire_url) 

183 self.assertEqual(identifier[0]["value"], "phq9") 

184 

185 question_1 = questionnaire["item"][0] 

186 question_10 = questionnaire["item"][9] 

187 self.assertEqual(question_1["linkId"], "q1") 

188 self.assertEqual(question_1["text"], 

189 "1. Little interest or pleasure in doing things") 

190 self.assertEqual(question_1["type"], "choice") 

191 

192 self.assertEqual(question_10["linkId"], "q10") 

193 self.assertEqual( 

194 question_10["text"], 

195 ("10. If you checked off any problems, how difficult have these " 

196 "problems made it for you to do your work, take care of things " 

197 "at home, or get along with other people?") 

198 ) 

199 self.assertEqual(question_10["type"], "choice") 

200 self.assertEqual(len(questionnaire["item"]), 10) 

201 

202 request = sent_json["entry"][1]["request"] 

203 self.assertEqual(request["method"], "POST") 

204 self.assertEqual(request["url"], "Questionnaire") 

205 self.assertEqual( 

206 request["ifNoneExist"], 

207 (f"identifier={questionnaire_url}|phq9") 

208 ) 

209 

210 def test_questionnaire_response_exported_with_phq9(self) -> None: 

211 exported_task = ExportedTask(task=self.task, recipient=self.recipient) 

212 exported_task_fhir = ExportedTaskFhir(exported_task) 

213 

214 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

215 

216 response_json = { 

217 'type': 'transaction-response', 

218 } 

219 

220 with mock.patch.object( 

221 exporter.client.server, "post_json", 

222 return_value=MockFhirResponse(response_json)) as mock_post: 

223 exporter.export_task() 

224 

225 args, kwargs = mock_post.call_args 

226 

227 sent_json = args[1] 

228 

229 response = sent_json["entry"][2]["resource"] 

230 self.assertEqual(response["resourceType"], "QuestionnaireResponse") 

231 self.assertEqual( 

232 response["questionnaire"], 

233 "http://127.0.0.1:8000/fhir_questionnaire_id|phq9" 

234 ) 

235 self.assertEqual(response["status"], "completed") 

236 

237 subject = response["subject"] 

238 identifier = subject["identifier"] 

239 self.assertEqual(subject["type"], "Patient") 

240 which_idnum = self.patient_rio.which_idnum 

241 idnum_value = self.patient_rio.idnum_value 

242 

243 iddef_url = f"http://127.0.0.1:8000/fhir_patient_id/{which_idnum}" 

244 self.assertEqual(identifier["system"], iddef_url) 

245 self.assertEqual(identifier["value"], str(idnum_value)) 

246 

247 request = sent_json["entry"][2]["request"] 

248 self.assertEqual(request["method"], "POST") 

249 self.assertEqual(request["url"], "QuestionnaireResponse") 

250 response_url = "http://127.0.0.1:8000/fhir_questionnaire_response_id/phq9" # noqa E501 

251 self.assertEqual( 

252 request["ifNoneExist"], 

253 (f"identifier={response_url}|{self.task._pk}") 

254 ) 

255 

256 item_1 = response["item"][0] 

257 item_10 = response["item"][9] 

258 self.assertEqual(item_1["linkId"], "q1") 

259 self.assertEqual(item_1["text"], 

260 "1. Little interest or pleasure in doing things") 

261 answer_1 = item_1["answer"][0] 

262 self.assertEqual(answer_1["valueInteger"], self.task.q1) 

263 

264 self.assertEqual(item_10["linkId"], "q10") 

265 self.assertEqual( 

266 item_10["text"], 

267 ("10. If you checked off any problems, how difficult have these " 

268 "problems made it for you to do your work, take care of things " 

269 "at home, or get along with other people?") 

270 ) 

271 answer_10 = item_10["answer"][0] 

272 self.assertEqual(answer_10["valueInteger"], self.task.q10) 

273 

274 self.assertEqual(len(response["item"]), 10) 

275 

276 def test_exported_task_saved(self) -> None: 

277 exported_task = ExportedTask(task=self.task, recipient=self.recipient) 

278 # auto increment doesn't work for BigInteger with SQLite 

279 exported_task.id = 1 

280 self.dbsession.add(exported_task) 

281 

282 exported_task_fhir = ExportedTaskFhir(exported_task) 

283 self.dbsession.add(exported_task_fhir) 

284 

285 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

286 

287 response_json = { 

288 'resourceType': 'Bundle', 

289 'id': 'cae48957-e7e6-4649-97f8-0a882076ad0a', 

290 'type': 'transaction-response', 

291 'link': [ 

292 { 

293 'relation': 'self', 

294 'url': 'http://localhost:8080/fhir' 

295 } 

296 ], 

297 'entry': [ 

298 { 

299 'response': { 

300 'status': '200 OK', 

301 'location': 'Patient/1/_history/1', 

302 'etag': '1' 

303 } 

304 }, 

305 { 

306 'response': { 

307 'status': '200 OK', 

308 'location': 'Questionnaire/26/_history/1', 

309 'etag': '1' 

310 } 

311 }, 

312 { 

313 'response': { 

314 'status': '201 Created', 

315 'location': 'QuestionnaireResponse/42/_history/1', 

316 'etag': '1', 

317 'lastModified': '2021-05-24T09:30:11.098+00:00' 

318 } 

319 } 

320 ] 

321 } 

322 

323 with mock.patch.object(exporter.client.server, "post_json", 

324 return_value=MockFhirResponse(response_json)): 

325 exporter.export_task() 

326 

327 self.dbsession.commit() 

328 

329 entries = exported_task_fhir.entries 

330 

331 entries.sort(key=lambda e: e.location) 

332 

333 self.assertEqual(entries[0].status, "200 OK") 

334 self.assertEqual(entries[0].location, "Patient/1/_history/1") 

335 self.assertEqual(entries[0].etag, "1") 

336 

337 self.assertEqual(entries[1].status, "200 OK") 

338 self.assertEqual(entries[1].location, "Questionnaire/26/_history/1") 

339 self.assertEqual(entries[1].etag, "1") 

340 

341 self.assertEqual(entries[2].status, "201 Created") 

342 self.assertEqual(entries[2].location, 

343 "QuestionnaireResponse/42/_history/1") 

344 self.assertEqual(entries[2].etag, "1") 

345 self.assertEqual(entries[2].last_modified, 

346 datetime.datetime(2021, 5, 24, 9, 30, 11, 98000))