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 

33import pendulum 

34from requests.exceptions import HTTPError 

35 

36from camcops_server.cc_modules.cc_exportmodels import ( 

37 ExportedTask, 

38 ExportedTaskFhir, 

39) 

40from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient 

41from camcops_server.cc_modules.cc_exportrecipientinfo import ExportRecipientInfo 

42from camcops_server.cc_modules.cc_fhir import ( 

43 FhirExportException, 

44 FhirTaskExporter, 

45) 

46from camcops_server.cc_modules.cc_unittest import DemoDatabaseTestCase 

47from camcops_server.tasks.apeqpt import Apeqpt 

48from camcops_server.tasks.phq9 import Phq9 

49 

50 

51# ============================================================================= 

52# Integration testing 

53# ============================================================================= 

54 

55 

56class MockFhirTaskExporter(FhirTaskExporter): 

57 pass 

58 

59 

60class MockFhirResponse(mock.Mock): 

61 def __init__(self, response_json: Dict): 

62 super().__init__( 

63 text=json.dumps(response_json), 

64 json=mock.Mock(return_value=response_json) 

65 ) 

66 

67 

68class FhirExportTestCase(DemoDatabaseTestCase): 

69 def setUp(self) -> None: 

70 super().setUp() 

71 recipientinfo = ExportRecipientInfo() 

72 

73 self.recipient = ExportRecipient(recipientinfo) 

74 self.recipient.primary_idnum = self.rio_iddef.which_idnum 

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

76 

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

78 self.recipient.id = 1 

79 self.recipient.recipient_name = "test" 

80 

81 

82class FhirTaskExporterPhq9Tests(FhirExportTestCase): 

83 def create_tasks(self) -> None: 

84 self.patient = self.create_patient( 

85 forename="Gwendolyn", 

86 surname="Ryann", 

87 sex="F" 

88 ) 

89 self.patient_nhs = self.create_patient_idnum( 

90 patient_id=self.patient.id, 

91 which_idnum=self.nhs_iddef.which_idnum, 

92 idnum_value=8879736213 

93 ) 

94 self.patient_rio = self.create_patient_idnum( 

95 patient_id=self.patient.id, 

96 which_idnum=self.rio_iddef.which_idnum, 

97 idnum_value=12345 

98 ) 

99 

100 self.task = Phq9() 

101 self.apply_standard_task_fields(self.task) 

102 self.task.q1 = 0 

103 self.task.q2 = 1 

104 self.task.q3 = 2 

105 self.task.q4 = 3 

106 self.task.q5 = 0 

107 self.task.q6 = 1 

108 self.task.q7 = 2 

109 self.task.q8 = 3 

110 self.task.q9 = 0 

111 self.task.q10 = 3 

112 self.task.patient_id = self.patient.id 

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

114 self.dbsession.commit() 

115 

116 def test_patient_exported(self) -> None: 

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

118 exported_task_fhir = ExportedTaskFhir(exported_task) 

119 

120 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

121 

122 response_json = { 

123 'type': 'transaction-response', 

124 } 

125 

126 with mock.patch.object( 

127 exporter.client.server, "post_json", 

128 return_value=MockFhirResponse(response_json)) as mock_post: 

129 exporter.export_task() 

130 

131 args, kwargs = mock_post.call_args 

132 

133 sent_json = args[1] 

134 

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

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

137 

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

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

140 

141 identifier = patient["identifier"] 

142 which_idnum = self.patient_rio.which_idnum 

143 idnum_value = self.patient_rio.idnum_value 

144 

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

146 

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

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

149 

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

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

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

153 

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

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

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

157 self.assertEqual( 

158 request["ifNoneExist"], 

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

160 ) 

161 

162 def test_questionnaire_exported(self) -> None: 

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

164 exported_task_fhir = ExportedTaskFhir(exported_task) 

165 

166 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

167 

168 response_json = { 

169 'type': 'transaction-response', 

170 } 

171 

172 with mock.patch.object( 

173 exporter.client.server, "post_json", 

174 return_value=MockFhirResponse(response_json)) as mock_post: 

175 exporter.export_task() 

176 

177 args, kwargs = mock_post.call_args 

178 

179 sent_json = args[1] 

180 

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

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

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

184 

185 identifier = questionnaire["identifier"] 

186 

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

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

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

190 

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

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

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

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

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

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

197 

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

199 self.assertEqual( 

200 question_10["text"], 

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

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

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

204 ) 

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

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

207 

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

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

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

211 self.assertEqual( 

212 request["ifNoneExist"], 

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

214 ) 

215 

216 def test_questionnaire_response_exported(self) -> None: 

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

218 exported_task_fhir = ExportedTaskFhir(exported_task) 

219 

220 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

221 

222 response_json = { 

223 'type': 'transaction-response', 

224 } 

225 

226 with mock.patch.object( 

227 exporter.client.server, "post_json", 

228 return_value=MockFhirResponse(response_json)) as mock_post: 

229 exporter.export_task() 

230 

231 args, kwargs = mock_post.call_args 

232 

233 sent_json = args[1] 

234 

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

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

237 self.assertEqual( 

238 response["questionnaire"], 

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

240 ) 

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

242 

243 subject = response["subject"] 

244 identifier = subject["identifier"] 

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

246 which_idnum = self.patient_rio.which_idnum 

247 idnum_value = self.patient_rio.idnum_value 

248 

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

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

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

252 

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

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

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

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

257 self.assertEqual( 

258 request["ifNoneExist"], 

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

260 ) 

261 

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

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

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

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

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

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

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

269 

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

271 self.assertEqual( 

272 item_10["text"], 

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

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

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

276 ) 

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

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

279 

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

281 

282 def test_exported_task_saved(self) -> None: 

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

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

285 exported_task.id = 1 

286 self.dbsession.add(exported_task) 

287 

288 exported_task_fhir = ExportedTaskFhir(exported_task) 

289 self.dbsession.add(exported_task_fhir) 

290 

291 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

292 

293 response_json = { 

294 'resourceType': 'Bundle', 

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

296 'type': 'transaction-response', 

297 'link': [ 

298 { 

299 'relation': 'self', 

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

301 } 

302 ], 

303 'entry': [ 

304 { 

305 'response': { 

306 'status': '200 OK', 

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

308 'etag': '1' 

309 } 

310 }, 

311 { 

312 'response': { 

313 'status': '200 OK', 

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

315 'etag': '1' 

316 } 

317 }, 

318 { 

319 'response': { 

320 'status': '201 Created', 

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

322 'etag': '1', 

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

324 } 

325 } 

326 ] 

327 } 

328 

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

330 return_value=MockFhirResponse(response_json)): 

331 exporter.export_task() 

332 

333 self.dbsession.commit() 

334 

335 entries = exported_task_fhir.entries 

336 

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

338 

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

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

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

342 

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

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

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

346 

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

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

349 "QuestionnaireResponse/42/_history/1") 

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

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

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

353 

354 def test_raises_when_task_does_not_support(self) -> None: 

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

356 exported_task_fhir = ExportedTaskFhir(exported_task) 

357 

358 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

359 

360 with mock.patch.object(self.task, 

361 "get_fhir_bundle_entries") as mock_task: 

362 mock_task.side_effect = NotImplementedError( 

363 "Something is not implemented" 

364 ) 

365 

366 with self.assertRaises(FhirExportException) as cm: 

367 exporter.export_task() 

368 

369 message = str(cm.exception) 

370 self.assertIn("Something is not implemented", message) 

371 

372 def test_raises_when_http_error(self) -> None: 

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

374 exported_task_fhir = ExportedTaskFhir(exported_task) 

375 

376 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

377 

378 with mock.patch.object( 

379 exporter.client.server, "post_json", 

380 side_effect=HTTPError( 

381 response=mock.Mock(text="Something bad happened") 

382 ) 

383 ): 

384 with self.assertRaises(FhirExportException) as cm: 

385 exporter.export_task() 

386 

387 message = str(cm.exception) 

388 self.assertIn("Something bad happened", message) 

389 

390 def test_raises_when_fhirclient_raises(self) -> None: 

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

392 exported_task_fhir = ExportedTaskFhir(exported_task) 

393 

394 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

395 

396 exporter.client.server = None 

397 with self.assertRaises(FhirExportException) as cm: 

398 exporter.export_task() 

399 

400 message = str(cm.exception) 

401 self.assertIn("Cannot create a resource without a server", message) 

402 

403 def test_raises_for_missing_api_url(self) -> None: 

404 self.recipient.fhir_api_url = "" 

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

406 exported_task_fhir = ExportedTaskFhir(exported_task) 

407 

408 with self.assertRaises(FhirExportException) as cm: 

409 FhirTaskExporter(self.req, exported_task_fhir) 

410 

411 message = str(cm.exception) 

412 self.assertIn("must be initialized with `base_uri`", message) 

413 

414 

415class FhirTaskExporterAnonymousTests(FhirExportTestCase): 

416 def create_tasks(self) -> None: 

417 self.task = Apeqpt() 

418 self.apply_standard_task_fields(self.task) 

419 self.task.q_datetime = pendulum.now() 

420 self.task.q1_choice = 0 

421 self.task.q2_choice = 1 

422 self.task.q3_choice = 2 

423 self.task.q1_satisfaction = 3 

424 self.task.q2_satisfaction = "Service experience" 

425 

426 self.task.save_with_next_available_id(self.req, 

427 self.server_device.id) 

428 self.dbsession.commit() 

429 

430 def test_questionnaire_exported(self) -> None: 

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

432 exported_task_fhir = ExportedTaskFhir(exported_task) 

433 

434 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

435 

436 response_json = { 

437 'type': 'transaction-response', 

438 } 

439 

440 with mock.patch.object( 

441 exporter.client.server, "post_json", 

442 return_value=MockFhirResponse(response_json)) as mock_post: 

443 exporter.export_task() 

444 

445 args, kwargs = mock_post.call_args 

446 

447 sent_json = args[1] 

448 

449 questionnaire = sent_json["entry"][0]["resource"] 

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

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

452 

453 identifier = questionnaire["identifier"] 

454 

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

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

457 self.assertEqual(identifier[0]["value"], "apeqpt") 

458 

459 q_datetime = questionnaire["item"][0] 

460 

461 q1_choice = questionnaire["item"][1] 

462 q2_choice = questionnaire["item"][2] 

463 q3_choice = questionnaire["item"][3] 

464 

465 q1_satisfaction = questionnaire["item"][4] 

466 q2_satisfaction = questionnaire["item"][5] 

467 

468 # q_datetime 

469 self.assertEqual(q_datetime["linkId"], "q_datetime") 

470 self.assertEqual(q_datetime["text"], 

471 "Date &amp; Time the Assessment Tool was Completed") 

472 self.assertEqual(q_datetime["type"], "dateTime") 

473 

474 # q1_choice 

475 self.assertEqual(q1_choice["linkId"], "q1_choice") 

476 self.assertEqual( 

477 q1_choice["text"], 

478 ("Were you given information about options for choosing a " 

479 "treatment that is appropriate for your problems?") 

480 ) 

481 self.assertEqual(q1_choice["type"], "choice") 

482 

483 # q2_choice 

484 self.assertEqual(q2_choice["linkId"], "q2_choice") 

485 self.assertEqual( 

486 q2_choice["text"], 

487 ("Do you prefer any of the treatments among the options available?") 

488 ) 

489 self.assertEqual(q2_choice["type"], "choice") 

490 

491 # q3_choice 

492 self.assertEqual(q3_choice["linkId"], "q3_choice") 

493 self.assertEqual( 

494 q3_choice["text"], 

495 ("Have you been offered your preference?") 

496 ) 

497 self.assertEqual(q3_choice["type"], "choice") 

498 

499 # q1_satisfaction 

500 self.assertEqual(q1_satisfaction["linkId"], "q1_satisfaction") 

501 self.assertEqual( 

502 q1_satisfaction["text"], 

503 ("How satisfied were you with your assessment") 

504 ) 

505 self.assertEqual(q1_satisfaction["type"], "choice") 

506 

507 # q2 satisfaction 

508 self.assertEqual(q2_satisfaction["linkId"], "q2_satisfaction") 

509 self.assertEqual( 

510 q2_satisfaction["text"], 

511 ("Please use this space to tell us about your experience of our " 

512 "service") 

513 ) 

514 self.assertEqual(q2_satisfaction["type"], "choice") 

515 

516 self.assertEqual(len(questionnaire["item"]), 6) 

517 

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

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

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

521 self.assertEqual( 

522 request["ifNoneExist"], 

523 (f"identifier={questionnaire_url}|apeqpt") 

524 ) 

525 

526 def test_questionnaire_response_exported(self) -> None: 

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

528 exported_task_fhir = ExportedTaskFhir(exported_task) 

529 

530 exporter = MockFhirTaskExporter(self.req, exported_task_fhir) 

531 

532 response_json = { 

533 'type': 'transaction-response', 

534 } 

535 

536 with mock.patch.object( 

537 exporter.client.server, "post_json", 

538 return_value=MockFhirResponse(response_json)) as mock_post: 

539 exporter.export_task() 

540 

541 args, kwargs = mock_post.call_args 

542 

543 sent_json = args[1] 

544 

545 response = sent_json["entry"][1]["resource"] 

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

547 self.assertEqual( 

548 response["questionnaire"], 

549 "http://127.0.0.1:8000/fhir_questionnaire_id|apeqpt" 

550 ) 

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

552 

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

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

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

556 response_url = "http://127.0.0.1:8000/fhir_questionnaire_response_id/apeqpt" # noqa E501 

557 self.assertEqual( 

558 request["ifNoneExist"], 

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

560 ) 

561 

562 q_datetime = response["item"][0] 

563 

564 q1_choice = response["item"][1] 

565 q2_choice = response["item"][2] 

566 q3_choice = response["item"][3] 

567 

568 q1_satisfaction = response["item"][4] 

569 q2_satisfaction = response["item"][5] 

570 

571 # q_datetime 

572 self.assertEqual(q_datetime["linkId"], "q_datetime") 

573 self.assertEqual(q_datetime["text"], 

574 "Date &amp; Time the Assessment Tool was Completed") 

575 q_datetime_answer = q_datetime["answer"][0] 

576 self.assertEqual(q_datetime_answer["valueDateTime"], 

577 self.task.q_datetime.isoformat()) 

578 

579 # q1_choice 

580 self.assertEqual(q1_choice["linkId"], "q1_choice") 

581 self.assertEqual( 

582 q1_choice["text"], 

583 ("Were you given information about options for choosing a " 

584 "treatment that is appropriate for your problems?") 

585 ) 

586 q1_choice_answer = q1_choice["answer"][0] 

587 self.assertEqual(q1_choice_answer["valueInteger"], self.task.q1_choice) 

588 

589 # q2_choice 

590 self.assertEqual(q2_choice["linkId"], "q2_choice") 

591 self.assertEqual( 

592 q2_choice["text"], 

593 ("Do you prefer any of the treatments among the options available?") 

594 ) 

595 q2_choice_answer = q2_choice["answer"][0] 

596 self.assertEqual(q2_choice_answer["valueInteger"], self.task.q2_choice) 

597 

598 # q3_choice 

599 self.assertEqual(q3_choice["linkId"], "q3_choice") 

600 self.assertEqual( 

601 q3_choice["text"], 

602 ("Have you been offered your preference?") 

603 ) 

604 q3_choice_answer = q3_choice["answer"][0] 

605 self.assertEqual(q3_choice_answer["valueInteger"], self.task.q3_choice) 

606 

607 # q1_satisfaction 

608 self.assertEqual(q1_satisfaction["linkId"], "q1_satisfaction") 

609 self.assertEqual( 

610 q1_satisfaction["text"], 

611 ("How satisfied were you with your assessment") 

612 ) 

613 q1_satisfaction_answer = q1_satisfaction["answer"][0] 

614 self.assertEqual(q1_satisfaction_answer["valueInteger"], 

615 self.task.q1_satisfaction) 

616 

617 # q2 satisfaction 

618 self.assertEqual(q2_satisfaction["linkId"], "q2_satisfaction") 

619 self.assertEqual( 

620 q2_satisfaction["text"], 

621 ("Please use this space to tell us about your experience of our " 

622 "service") 

623 ) 

624 q2_satisfaction_answer = q2_satisfaction["answer"][0] 

625 self.assertEqual(q2_satisfaction_answer["valueString"], 

626 self.task.q2_satisfaction) 

627 

628 self.assertEqual(len(response["item"]), 6)