Coverage for cc_modules/tests/cc_fhir

1"""

2camcops_server/cc_modules/tests/cc_fhir_tests.py

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

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

9 This file is part of CamCOPS.

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.

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.

21 You should have received a copy of the GNU General Public License

22 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.

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

26"""

28import datetime

29import json

30import logging

31from typing import Dict, List

32from unittest import mock

34from cardinal_pythonlib.httpconst import HttpMethod

35import pendulum

36from requests.exceptions import HTTPError

38from camcops_server.cc_modules.cc_constants import FHIRConst as Fc

39from camcops_server.cc_modules.cc_exportmodels import (

40 ExportedTask,

41 ExportedTaskFhir,

42 ExportedTaskFhirEntry,

43)

44from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient

45from camcops_server.cc_modules.cc_exportrecipientinfo import (

46 ExportRecipientInfo,

47)

48from camcops_server.cc_modules.cc_fhir import (

49 fhir_reference_from_identifier,

50 fhir_sysval_from_id,

51 FhirExportException,

52 FhirTaskExporter,

53)

54from camcops_server.cc_modules.cc_pyramid import Routes

55from camcops_server.cc_modules.cc_testfactories import (

56 NHSPatientIdNumFactory,

57 PatientFactory,

58 RioPatientIdNumFactory,

59)

61from camcops_server.cc_modules.cc_unittest import DemoRequestTestCase

62from camcops_server.cc_modules.cc_version_string import (

63 CAMCOPS_SERVER_VERSION_STRING,

64)

65from camcops_server.tasks.tests.factories import (

66 ApeqptFactory,

67 BmiFactory,

68 DiagnosisIcd10Factory,

69 DiagnosisIcd10ItemFactory,

70 DiagnosisIcd9CMFactory,

71 DiagnosisIcd9CMItemFactory,

72 Gad7Factory,

73 Phq9Factory,

74)

77log = logging.getLogger()

80# =============================================================================

81# Helper classes

82# =============================================================================

85class MockFhirTaskExporter(FhirTaskExporter):

86 pass

89class MockFhirResponse(mock.Mock):

90 def __init__(self, response_json: Dict):

91 super().__init__(

92 text=json.dumps(response_json),

93 json=mock.Mock(return_value=response_json),

94 )

97class FhirExportTestCase(DemoRequestTestCase):

98 def setUp(self) -> None:

99 super().setUp()

100 recipientinfo = ExportRecipientInfo()

101

102 self.recipient = ExportRecipient(recipientinfo)

103 self.recipient.fhir_api_url = "https://www.example.com/fhir"

104

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

106 self.recipient.id = 1

107 self.recipient.recipient_name = "test"

108

109 self.camcops_root_url = self.req.route_url(Routes.HOME).rstrip("/")

110 # ... no trailing slash

111

112

113class FhirExportPatientTestCase(FhirExportTestCase):

114 def setUp(self) -> None:

115 super().setUp()

116

117 self.patient = PatientFactory()

118 self.patient_nhs_idnum = NHSPatientIdNumFactory(patient=self.patient)

119 self.patient_rio_idnum = RioPatientIdNumFactory(patient=self.patient)

120

121 self.recipient.primary_idnum = self.patient_rio_idnum.which_idnum

122

123

124# =============================================================================

125# A generic patient-based task: PHQ9

126# =============================================================================

127

128

129class FhirTaskExporterPhq9Tests(FhirExportPatientTestCase):

130 def setUp(self) -> None:

131 super().setUp()

132

133 self.task = Phq9Factory(

134 patient=self.patient,

135 q1=0,

136 q2=1,

137 q3=2,

138 q4=3,

139 q5=0,

140 q6=1,

141 q7=2,

142 q8=3,

143 q9=0,

144 q10=3,

145 )

146

147 def test_patient_exported(self) -> None:

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

149 exported_task_fhir = ExportedTaskFhir(exported_task)

150

151 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

152

153 response_json = {Fc.TYPE: Fc.TRANSACTION_RESPONSE}

154

155 with mock.patch.object(

156 exporter.client.server,

157 "post_json",

158 return_value=MockFhirResponse(response_json),

159 ) as mock_post:

160 exporter.export_task()

161

162 args, kwargs = mock_post.call_args

163

164 sent_json = args[1]

165

166 self.assertEqual(sent_json[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_BUNDLE)

167 self.assertEqual(sent_json[Fc.TYPE], Fc.TRANSACTION)

168

169 patient = sent_json[Fc.ENTRY][0][Fc.RESOURCE]

170 self.assertEqual(patient[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_PATIENT)

171

172 identifier = patient[Fc.IDENTIFIER]

173 idnum_value = self.patient_rio_idnum.idnum_value

174

175 patient_id = self.patient.get_fhir_identifier(self.req, self.recipient)

176

177 self.assertEqual(identifier[0][Fc.SYSTEM], patient_id.system)

178 self.assertEqual(identifier[0][Fc.VALUE], str(idnum_value))

179

180 self.assertEqual(

181 patient[Fc.NAME][0][Fc.NAME_FAMILY], self.patient.surname

182 )

183 self.assertEqual(

184 patient[Fc.NAME][0][Fc.NAME_GIVEN], [self.patient.forename]

185 )

186 self.assertEqual(patient[Fc.GENDER], Fc.GENDER_FEMALE)

187

188 request = sent_json[Fc.ENTRY][0][Fc.REQUEST]

189 self.assertEqual(request[Fc.METHOD], HttpMethod.POST)

190 self.assertEqual(request[Fc.URL], Fc.RESOURCE_TYPE_PATIENT)

191 self.assertEqual(

192 request[Fc.IF_NONE_EXIST],

193 fhir_reference_from_identifier(patient_id),

194 )

195

196 def test_questionnaire_exported(self) -> None:

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

198 exported_task_fhir = ExportedTaskFhir(exported_task)

199

200 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

201

202 response_json = {Fc.TYPE: Fc.TRANSACTION_RESPONSE}

203

204 with mock.patch.object(

205 exporter.client.server,

206 "post_json",

207 return_value=MockFhirResponse(response_json),

208 ) as mock_post:

209 exporter.export_task()

210

211 args, kwargs = mock_post.call_args

212

213 sent_json = args[1]

214

215 questionnaire = sent_json[Fc.ENTRY][1][Fc.RESOURCE]

216 self.assertEqual(

217 questionnaire[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_QUESTIONNAIRE

218 )

219 self.assertEqual(questionnaire[Fc.STATUS], Fc.QSTATUS_ACTIVE)

220

221 identifier = questionnaire[Fc.IDENTIFIER]

222

223 questionnaire_url = (

224 f"{self.camcops_root_url}/{Routes.FHIR_QUESTIONNAIRE_SYSTEM}"

225 )

226 self.assertEqual(identifier[0][Fc.SYSTEM], questionnaire_url)

227 self.assertEqual(

228 identifier[0][Fc.VALUE], f"phq9/{CAMCOPS_SERVER_VERSION_STRING}"

229 )

230

231 question_1 = questionnaire[Fc.ITEM][0]

232 question_10 = questionnaire[Fc.ITEM][9]

233 self.assertEqual(question_1[Fc.LINK_ID], "q1")

234 self.assertEqual(

235 question_1[Fc.TEXT],

236 "1. Little interest or pleasure in doing things",

237 )

238 self.assertEqual(question_1[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

239

240 options = question_1[Fc.ANSWER_OPTION]

241 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

242 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.DISPLAY], "Not at all")

243

244 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

245 self.assertEqual(

246 options[1][Fc.VALUE_CODING][Fc.DISPLAY], "Several days"

247 )

248

249 self.assertEqual(options[2][Fc.VALUE_CODING][Fc.CODE], "2")

250 self.assertEqual(

251 options[2][Fc.VALUE_CODING][Fc.DISPLAY], "More than half the days"

252 )

253

254 self.assertEqual(options[3][Fc.VALUE_CODING][Fc.CODE], "3")

255 self.assertEqual(

256 options[3][Fc.VALUE_CODING][Fc.DISPLAY], "Nearly every day"

257 )

258

259 self.assertEqual(question_10[Fc.LINK_ID], "q10")

260 self.assertEqual(

261 question_10[Fc.TEXT],

262 (

263 "10. If you checked off any problems, how difficult have "

264 "these problems made it for you to do your work, take care of "

265 "things at home, or get along with other people?"

266 ),

267 )

268 self.assertEqual(question_10[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

269 options = question_10[Fc.ANSWER_OPTION]

270 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

271 self.assertEqual(

272 options[0][Fc.VALUE_CODING][Fc.DISPLAY], "Not difficult at all"

273 )

274

275 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

276 self.assertEqual(

277 options[1][Fc.VALUE_CODING][Fc.DISPLAY], "Somewhat difficult"

278 )

279

280 self.assertEqual(options[2][Fc.VALUE_CODING][Fc.CODE], "2")

281 self.assertEqual(

282 options[2][Fc.VALUE_CODING][Fc.DISPLAY], "Very difficult"

283 )

284

285 self.assertEqual(options[3][Fc.VALUE_CODING][Fc.CODE], "3")

286 self.assertEqual(

287 options[3][Fc.VALUE_CODING][Fc.DISPLAY], "Extremely difficult"

288 )

289

290 self.assertEqual(len(questionnaire[Fc.ITEM]), 10)

291

292 request = sent_json[Fc.ENTRY][1][Fc.REQUEST]

293 self.assertEqual(request[Fc.METHOD], HttpMethod.POST)

294 self.assertEqual(request[Fc.URL], Fc.RESOURCE_TYPE_QUESTIONNAIRE)

295 q_id = self.task._get_fhir_questionnaire_id(self.req)

296 self.assertEqual(

297 request[Fc.IF_NONE_EXIST], fhir_reference_from_identifier(q_id)

298 )

299

300 def test_questionnaire_response_exported(self) -> None:

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

302 exported_task_fhir = ExportedTaskFhir(exported_task)

303

304 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

305

306 response_json = {Fc.TYPE: Fc.TRANSACTION_RESPONSE}

307

308 with mock.patch.object(

309 exporter.client.server,

310 "post_json",

311 return_value=MockFhirResponse(response_json),

312 ) as mock_post:

313 exporter.export_task()

314

315 args, kwargs = mock_post.call_args

316

317 sent_json = args[1]

318

319 response = sent_json[Fc.ENTRY][2][Fc.RESOURCE]

320 self.assertEqual(

321 response[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_QUESTIONNAIRE_RESPONSE

322 )

323

324 q_id = self.task._get_fhir_questionnaire_id(self.req)

325 self.assertEqual(response[Fc.QUESTIONNAIRE], fhir_sysval_from_id(q_id))

326 self.assertEqual(

327 response[Fc.AUTHORED], self.task.when_created.isoformat()

328 )

329 self.assertEqual(response[Fc.STATUS], Fc.QSTATUS_COMPLETED)

330

331 subject = response[Fc.SUBJECT]

332 identifier = subject[Fc.IDENTIFIER]

333 self.assertEqual(subject[Fc.TYPE], Fc.RESOURCE_TYPE_PATIENT)

334 idnum_value = self.patient_rio_idnum.idnum_value

335

336 patient_id = self.patient.get_fhir_identifier(self.req, self.recipient)

337 if isinstance(identifier, list):

338 test_identifier = identifier[0]

339 else: # only one

340 test_identifier = identifier

341 self.assertEqual(test_identifier[Fc.SYSTEM], patient_id.system)

342 self.assertEqual(test_identifier[Fc.VALUE], str(idnum_value))

343

344 request = sent_json[Fc.ENTRY][2][Fc.REQUEST]

345 self.assertEqual(request[Fc.METHOD], HttpMethod.POST)

346 self.assertEqual(

347 request[Fc.URL], Fc.RESOURCE_TYPE_QUESTIONNAIRE_RESPONSE

348 )

349 qr_id = self.task._get_fhir_questionnaire_response_id(self.req)

350 self.assertEqual(

351 request[Fc.IF_NONE_EXIST], fhir_reference_from_identifier(qr_id)

352 )

353

354 item_1 = response[Fc.ITEM][0]

355 item_10 = response[Fc.ITEM][9]

356 self.assertEqual(item_1[Fc.LINK_ID], "q1")

357 self.assertEqual(

358 item_1[Fc.TEXT], "1. Little interest or pleasure in doing things"

359 )

360 answer_1 = item_1[Fc.ANSWER][0]

361 # noinspection PyUnresolvedReferences

362 self.assertEqual(answer_1[Fc.VALUE_INTEGER], self.task.q1)

363

364 self.assertEqual(item_10[Fc.LINK_ID], "q10")

365 self.assertEqual(

366 item_10[Fc.TEXT],

367 (

368 "10. If you checked off any problems, how difficult have "

369 "these problems made it for you to do your work, take care of "

370 "things at home, or get along with other people?"

371 ),

372 )

373 answer_10 = item_10[Fc.ANSWER][0]

374 self.assertEqual(answer_10[Fc.VALUE_INTEGER], self.task.q10)

375

376 self.assertEqual(len(response[Fc.ITEM]), 10)

377

378 # noinspection PyUnresolvedReferences

379 def test_exported_task_saved(self) -> None:

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

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

382 exported_task.id = 1

383 self.dbsession.add(exported_task)

384

385 exported_task_fhir = ExportedTaskFhir(exported_task)

386 self.dbsession.add(exported_task_fhir)

387

388 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

389

390 response_json = {

391 Fc.RESOURCE_TYPE: Fc.RESOURCE_TYPE_BUNDLE,

392 Fc.ID: "cae48957-e7e6-4649-97f8-0a882076ad0a",

393 Fc.TYPE: Fc.TRANSACTION_RESPONSE,

394 Fc.LINK: [

395 {Fc.RELATION: Fc.SELF, Fc.URL: "http://localhost:8080/fhir"}

396 ],

397 Fc.ENTRY: [

398 {

399 Fc.RESPONSE: {

400 Fc.STATUS: Fc.RESPONSE_STATUS_200_OK,

401 Fc.LOCATION: "Patient/1/_history/1",

402 Fc.ETAG: "1",

403 }

404 },

405 {

406 Fc.RESPONSE: {

407 Fc.STATUS: Fc.RESPONSE_STATUS_200_OK,

408 Fc.LOCATION: "Questionnaire/26/_history/1",

409 Fc.ETAG: "1",

410 }

411 },

412 {

413 Fc.RESPONSE: {

414 Fc.STATUS: Fc.RESPONSE_STATUS_201_CREATED,

415 Fc.LOCATION: "QuestionnaireResponse/42/_history/1",

416 Fc.ETAG: "1",

417 Fc.LAST_MODIFIED: "2021-05-24T09:30:11.098+00:00",

418 }

419 },

420 ],

421 }

422

423 with mock.patch.object(

424 exporter.client.server,

425 "post_json",

426 return_value=MockFhirResponse(response_json),

427 ):

428 exporter.export_task()

429

430 self.dbsession.commit()

431

432 entries = (

433 exported_task_fhir.entries

434 ) # type: List[ExportedTaskFhirEntry]

435

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

437

438 self.assertEqual(entries[0].status, Fc.RESPONSE_STATUS_200_OK)

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

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

441

442 self.assertEqual(entries[1].status, Fc.RESPONSE_STATUS_200_OK)

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

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

445

446 self.assertEqual(entries[2].status, Fc.RESPONSE_STATUS_201_CREATED)

447 self.assertEqual(

448 entries[2].location, "QuestionnaireResponse/42/_history/1"

449 )

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

451 self.assertEqual(

452 entries[2].last_modified,

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

454 )

455

456 def test_raises_when_http_error(self) -> None:

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

458 exported_task_fhir = ExportedTaskFhir(exported_task)

459

460 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

461

462 errmsg = "Something bad happened"

463 with mock.patch.object(

464 exporter.client.server,

465 "post_json",

466 side_effect=HTTPError(response=mock.Mock(text=errmsg)),

467 ):

468 with self.assertRaises(FhirExportException) as cm:

469 exporter.export_task()

470

471 message = str(cm.exception)

472 self.assertIn(errmsg, message)

473

474 def test_raises_when_fhirclient_raises(self) -> None:

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

476 exported_task_fhir = ExportedTaskFhir(exported_task)

477

478 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

479

480 exporter.client.server = None

481 with self.assertRaises(FhirExportException) as cm:

482 exporter.export_task()

483

484 message = str(cm.exception)

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

486

487 def test_raises_for_missing_api_url(self) -> None:

488 self.recipient.fhir_api_url = ""

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

490 exported_task_fhir = ExportedTaskFhir(exported_task)

491

492 with self.assertRaises(FhirExportException) as cm:

493 FhirTaskExporter(self.req, exported_task_fhir)

494

495 message = str(cm.exception)

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

497

498

499# =============================================================================

500# A generic anonymous task: APEQPT

501# =============================================================================

502

503APEQPT_Q_WHEN = "Date and time the assessment tool was completed"

504OFFERED_PREFERENCE = "Have you been offered your preference?"

505SATISFIED_ASSESSMENT = "How satisfied were you with your assessment?"

506TELL_US = (

507 "Please use this space to tell us about your experience of our service."

508)

509PREFER_ANY = "Do you prefer any of the treatments among the options available?"

510GIVEN_INFO = (

511 "Were you given information about options for choosing a "

512 "treatment that is appropriate for your problems?"

513)

514APEQ_SATIS_A4 = "Completely satisfied"

515APEQ_SATIS_A3 = "Mostly satisfied"

516APEQ_SATIS_A2 = "Neither satisfied nor dissatisfied"

517APEQ_SATIS_A1 = "Not satisfied"

518APEQ_SATIS_A0 = "Not at all satisfied"

519

520

521class FhirTaskExporterAnonymousTests(FhirExportTestCase):

522 def setUp(self) -> None:

523 super().setUp()

524

525 self.task = ApeqptFactory(

526 q_datetime=pendulum.now(),

527 q1_choice=0,

528 q2_choice=1,

529 q3_choice=2,

530 q1_satisfaction=3,

531 q2_satisfaction="Service experience",

532 )

533

534 def test_questionnaire_exported(self) -> None:

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

536 exported_task_fhir = ExportedTaskFhir(exported_task)

537

538 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

539

540 response_json = {Fc.TYPE: Fc.TRANSACTION_RESPONSE}

541

542 with mock.patch.object(

543 exporter.client.server,

544 "post_json",

545 return_value=MockFhirResponse(response_json),

546 ) as mock_post:

547 exporter.export_task()

548

549 args, kwargs = mock_post.call_args

550

551 sent_json = args[1]

552

553 questionnaire = sent_json[Fc.ENTRY][0][Fc.RESOURCE]

554 self.assertEqual(

555 questionnaire[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_QUESTIONNAIRE

556 )

557 self.assertEqual(questionnaire[Fc.STATUS], Fc.QSTATUS_ACTIVE)

558

559 identifier = questionnaire[Fc.IDENTIFIER]

560

561 questionnaire_url = (

562 f"{self.camcops_root_url}/{Routes.FHIR_QUESTIONNAIRE_SYSTEM}"

563 )

564 self.assertEqual(identifier[0][Fc.SYSTEM], questionnaire_url)

565 self.assertEqual(

566 identifier[0][Fc.VALUE], f"apeqpt/{CAMCOPS_SERVER_VERSION_STRING}"

567 )

568

569 self.assertEqual(len(questionnaire[Fc.ITEM]), 5)

570 (

571 q1_choice,

572 q2_choice,

573 q3_choice,

574 q1_satisfaction,

575 q2_satisfaction,

576 ) = questionnaire[Fc.ITEM]

577

578 # q1_choice

579 self.assertEqual(q1_choice[Fc.LINK_ID], "q1_choice")

580 self.assertEqual(q1_choice[Fc.TEXT], GIVEN_INFO)

581 self.assertEqual(q1_choice[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

582

583 options = q1_choice[Fc.ANSWER_OPTION]

584 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

585 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.DISPLAY], "No")

586

587 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

588 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.DISPLAY], "Yes")

589

590 # q2_choice

591 self.assertEqual(q2_choice[Fc.LINK_ID], "q2_choice")

592 self.assertEqual(q2_choice[Fc.TEXT], PREFER_ANY)

593 self.assertEqual(q2_choice[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

594 options = q2_choice[Fc.ANSWER_OPTION]

595 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

596 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.DISPLAY], "No")

597

598 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

599 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.DISPLAY], "Yes")

600

601 # q3_choice

602 self.assertEqual(q3_choice[Fc.LINK_ID], "q3_choice")

603 self.assertEqual(q3_choice[Fc.TEXT], OFFERED_PREFERENCE)

604 self.assertEqual(q3_choice[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

605 options = q3_choice[Fc.ANSWER_OPTION]

606 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

607 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.DISPLAY], "No")

608

609 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

610 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.DISPLAY], "Yes")

611

612 self.assertEqual(options[2][Fc.VALUE_CODING][Fc.CODE], "2")

613 self.assertEqual(options[2][Fc.VALUE_CODING][Fc.DISPLAY], "N/A")

614

615 # q1_satisfaction

616 self.assertEqual(q1_satisfaction[Fc.LINK_ID], "q1_satisfaction")

617 self.assertEqual(q1_satisfaction[Fc.TEXT], SATISFIED_ASSESSMENT)

618 self.assertEqual(q1_satisfaction[Fc.TYPE], Fc.QITEM_TYPE_CHOICE)

619 options = q1_satisfaction[Fc.ANSWER_OPTION]

620 self.assertEqual(options[0][Fc.VALUE_CODING][Fc.CODE], "0")

621 self.assertEqual(

622 options[0][Fc.VALUE_CODING][Fc.DISPLAY], APEQ_SATIS_A0

623 )

624

625 self.assertEqual(options[1][Fc.VALUE_CODING][Fc.CODE], "1")

626 self.assertEqual(

627 options[1][Fc.VALUE_CODING][Fc.DISPLAY], APEQ_SATIS_A1

628 )

629

630 self.assertEqual(options[2][Fc.VALUE_CODING][Fc.CODE], "2")

631 self.assertEqual(

632 options[2][Fc.VALUE_CODING][Fc.DISPLAY], APEQ_SATIS_A2

633 )

634

635 self.assertEqual(options[3][Fc.VALUE_CODING][Fc.CODE], "3")

636 self.assertEqual(

637 options[3][Fc.VALUE_CODING][Fc.DISPLAY], APEQ_SATIS_A3

638 )

639

640 self.assertEqual(options[4][Fc.VALUE_CODING][Fc.CODE], "4")

641 self.assertEqual(

642 options[4][Fc.VALUE_CODING][Fc.DISPLAY], APEQ_SATIS_A4

643 )

644

645 # q2 satisfaction

646 self.assertEqual(q2_satisfaction[Fc.LINK_ID], "q2_satisfaction")

647 self.assertEqual(q2_satisfaction[Fc.TEXT], TELL_US)

648 self.assertEqual(q2_satisfaction[Fc.TYPE], Fc.QITEM_TYPE_STRING)

649

650 request = sent_json[Fc.ENTRY][0][Fc.REQUEST]

651 self.assertEqual(request[Fc.METHOD], HttpMethod.POST)

652 self.assertEqual(request[Fc.URL], Fc.RESOURCE_TYPE_QUESTIONNAIRE)

653 q_id = self.task._get_fhir_questionnaire_id(self.req)

654 self.assertEqual(

655 request[Fc.IF_NONE_EXIST], fhir_reference_from_identifier(q_id)

656 )

657

658 def test_questionnaire_response_exported(self) -> None:

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

660 exported_task_fhir = ExportedTaskFhir(exported_task)

661

662 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)

663

664 response_json = {Fc.TYPE: Fc.TRANSACTION_RESPONSE}

665

666 with mock.patch.object(

667 exporter.client.server,

668 "post_json",

669 return_value=MockFhirResponse(response_json),

670 ) as mock_post:

671 exporter.export_task()

672

673 args, kwargs = mock_post.call_args

674

675 sent_json = args[1]

676

677 response = sent_json[Fc.ENTRY][1][Fc.RESOURCE]

678 self.assertEqual(

679 response[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_QUESTIONNAIRE_RESPONSE

680 )

681 q_id = self.task._get_fhir_questionnaire_id(self.req)

682 self.assertEqual(response[Fc.QUESTIONNAIRE], fhir_sysval_from_id(q_id))

683 self.assertEqual(

684 response[Fc.AUTHORED], self.task.when_created.isoformat()

685 )

686 self.assertEqual(response[Fc.STATUS], Fc.QSTATUS_COMPLETED)

687

688 request = sent_json[Fc.ENTRY][1][Fc.REQUEST]

689 self.assertEqual(request[Fc.METHOD], HttpMethod.POST)

690 self.assertEqual(request[Fc.URL], "QuestionnaireResponse")

691 qr_id = self.task._get_fhir_questionnaire_response_id(self.req)

692 self.assertEqual(

693 request[Fc.IF_NONE_EXIST], fhir_reference_from_identifier(qr_id)

694 )

695

696 self.assertEqual(len(response[Fc.ITEM]), 5)

697 (

698 q1_choice,

699 q2_choice,

700 q3_choice,

701 q1_satisfaction,

702 q2_satisfaction,

703 ) = response[Fc.ITEM]

704

705 # q1_choice

706 self.assertEqual(q1_choice[Fc.LINK_ID], "q1_choice")

707 self.assertEqual(q1_choice[Fc.TEXT], GIVEN_INFO)

708 q1_choice_answer = q1_choice[Fc.ANSWER][0]

709 self.assertEqual(

710 q1_choice_answer[Fc.VALUE_INTEGER], self.task.q1_choice

711 )

712

713 # q2_choice

714 self.assertEqual(q2_choice[Fc.LINK_ID], "q2_choice")

715 self.assertEqual(q2_choice[Fc.TEXT], PREFER_ANY)

716 q2_choice_answer = q2_choice[Fc.ANSWER][0]

717 self.assertEqual(

718 q2_choice_answer[Fc.VALUE_INTEGER], self.task.q2_choice

719 )

720

721 # q3_choice

722 self.assertEqual(q3_choice[Fc.LINK_ID], "q3_choice")

723 self.assertEqual(q3_choice[Fc.TEXT], OFFERED_PREFERENCE)

724 q3_choice_answer = q3_choice[Fc.ANSWER][0]

725 self.assertEqual(

726 q3_choice_answer[Fc.VALUE_INTEGER], self.task.q3_choice

727 )

728

729 # q1_satisfaction

730 self.assertEqual(q1_satisfaction[Fc.LINK_ID], "q1_satisfaction")

731 self.assertEqual(q1_satisfaction[Fc.TEXT], SATISFIED_ASSESSMENT)

732 q1_satisfaction_answer = q1_satisfaction[Fc.ANSWER][0]

733 self.assertEqual(

734 q1_satisfaction_answer[Fc.VALUE_INTEGER], self.task.q1_satisfaction

735 )

736

737 # q2 satisfaction

738 self.assertEqual(q2_satisfaction[Fc.LINK_ID], "q2_satisfaction")

739 self.assertEqual(q2_satisfaction[Fc.TEXT], TELL_US)

740 q2_satisfaction_answer = q2_satisfaction[Fc.ANSWER][0]

741 self.assertEqual(

742 q2_satisfaction_answer[Fc.VALUE_STRING], self.task.q2_satisfaction

743 )

744

745

746# =============================================================================

747# Tasks that add their own special details

748# =============================================================================

749

750

751class FhirTaskExporterBMITests(FhirExportPatientTestCase):

752 def setUp(self) -> None:

753 super().setUp()

754

755 self.task = BmiFactory(patient=self.patient)

756

757 def test_observations(self) -> None:

758 bundle = self.task.get_fhir_bundle(

759 self.req, self.recipient, skip_docs_if_other_content=True

760 )

761

762 bundle_json = bundle.as_json()

763

764 height_entry = bundle_json[Fc.ENTRY][3]

765 mass_entry = bundle_json[Fc.ENTRY][4]

766 bmi_entry = bundle_json[Fc.ENTRY][5]

767 waist_entry = bundle_json[Fc.ENTRY][6]

768

769 height_resource = height_entry[Fc.RESOURCE]

770 mass_resource = mass_entry[Fc.RESOURCE]

771 bmi_resource = bmi_entry[Fc.RESOURCE]

772 waist_resource = waist_entry[Fc.RESOURCE]

773

774 self.assertEqual(

775 height_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_OBSERVATION

776 )

777 self.assertEqual(

778 height_resource[Fc.VALUE_QUANTITY][Fc.VALUE], self.task.height_m

779 )

780

781 self.assertEqual(

782 mass_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_OBSERVATION

783 )

784 self.assertAlmostEqual(

785 mass_resource[Fc.VALUE_QUANTITY][Fc.VALUE],

786 self.task.mass_kg,

787 places=2,

788 )

789

790 self.assertEqual(

791 bmi_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_OBSERVATION

792 )

793 self.assertAlmostEqual(

794 bmi_resource[Fc.VALUE_QUANTITY][Fc.VALUE],

795 self.task.bmi(),

796 places=2,

797 )

798

799 self.assertEqual(

800 waist_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_OBSERVATION

801 )

802 self.assertAlmostEqual(

803 waist_resource[Fc.VALUE_QUANTITY][Fc.VALUE],

804 self.task.waist_cm,

805 places=2,

806 )

807

808

809class FhirTaskExporterDiagnosisIcd10Tests(FhirExportPatientTestCase):

810 def setUp(self) -> None:

811 super().setUp()

812

813 self.task = DiagnosisIcd10Factory(patient=self.patient)

814

815 # noinspection PyArgumentList

816 self.item1 = DiagnosisIcd10ItemFactory(

817 diagnosis_icd10=self.task,

818 seqnum=1,

819 code="F33.30",

820 description="Recurrent depressive disorder, current episode "

821 "severe with psychotic symptoms: "

822 "with mood-congruent psychotic symptoms",

823 comment="Cotard's syndrome",

824 )

825 # noinspection PyArgumentList

826 self.item2 = DiagnosisIcd10ItemFactory(

827 diagnosis_icd10=self.task,

828 seqnum=2,

829 code="F43.1",

830 description="Post-traumatic stress disorder",

831 )

832

833 def test_observations(self) -> None:

834 bundle = self.task.get_fhir_bundle(

835 self.req, self.recipient, skip_docs_if_other_content=True

836 )

837

838 bundle_json = bundle.as_json()

839

840 cotard_resource = bundle_json[Fc.ENTRY][4][Fc.RESOURCE]

841 ptsd_resource = bundle_json[Fc.ENTRY][5][Fc.RESOURCE]

842

843 self.assertEqual(

844 cotard_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_CONDITION

845 )

846 self.assertEqual(

847 cotard_resource[Fc.CODE][Fc.CODING][0][Fc.CODE], "F33.30"

848 )

849 self.assertIn(

850 "Cotard's syndrome",

851 cotard_resource[Fc.CODE][Fc.CODING][0][Fc.DISPLAY],

852 )

853 self.assertIn(

854 "Recurrent depressive",

855 cotard_resource[Fc.CODE][Fc.CODING][0][Fc.DISPLAY],

856 )

857

858 self.assertEqual(

859 ptsd_resource[Fc.CODE][Fc.CODING][0][Fc.CODE], "F43.1"

860 )

861

862

863class FhirTaskExporterDiagnosisIcd9CMTests(FhirExportPatientTestCase):

864 def setUp(self) -> None:

865 super().setUp()

866

867 self.task = DiagnosisIcd9CMFactory(patient=self.patient)

868

869 # noinspection PyArgumentList

870 self.item1 = DiagnosisIcd9CMItemFactory(

871 diagnosis_icd9cm=self.task,

872 seqnum=1,

873 code="290.4",

874 description="Vascular dementia",

875 comment="or perhaps mixed dementia",

876 )

877 # noinspection PyArgumentList

878 self.item2 = DiagnosisIcd9CMItemFactory(

879 diagnosis_icd9cm=self.task,

880 seqnum=2,

881 code="303.0",

882 description="Acute alcoholic intoxication",

883 )

884

885 def test_observations(self) -> None:

886 bundle = self.task.get_fhir_bundle(

887 self.req, self.recipient, skip_docs_if_other_content=True

888 )

889 bundle_json = bundle.as_json()

890 dementia_resource = bundle_json[Fc.ENTRY][4][Fc.RESOURCE]

891 intoxication_resource = bundle_json[Fc.ENTRY][5][Fc.RESOURCE]

892

893 self.assertEqual(

894 dementia_resource[Fc.RESOURCE_TYPE], Fc.RESOURCE_TYPE_CONDITION

895 )

896 self.assertEqual(

897 dementia_resource[Fc.CODE][Fc.CODING][0][Fc.CODE], "290.4"

898 )

899 self.assertIn(

900 "Vascular dementia",

901 dementia_resource[Fc.CODE][Fc.CODING][0][Fc.DISPLAY],

902 )

903 self.assertIn(

904 "or perhaps mixed dementia",

905 dementia_resource[Fc.CODE][Fc.CODING][0][Fc.DISPLAY],

906 )

907

908 self.assertEqual(

909 intoxication_resource[Fc.CODE][Fc.CODING][0][Fc.CODE], "303.0"

910 )

911

912

913class FhirTaskExporterGad7Tests(FhirExportPatientTestCase):

914 """

915 The GAD7 is a standard questionnaire that we don't provide any special

916 FHIR support for; we rely on autodiscovery. This is essentially a high

917 level test for _fhir_autodiscover() in cc_task.py, albeit not a

918 particularly thorough one.

919 """

920

921 def setUp(self) -> None:

922 super().setUp()

923

924 self.task = Gad7Factory(

925 patient=self.patient,

926 q1=0,

927 q2=1,

928 q3=2,

929 q4=3,

930 q5=0,

931 q6=1,

932 q7=2,

933 )

934

935 def test_observations(self) -> None:

936 bundle = self.task.get_fhir_bundle(

937 self.req, self.recipient, skip_docs_if_other_content=True

938 )

939 bundle_json = bundle.as_json()

940 questions = bundle_json[Fc.ENTRY][1][Fc.RESOURCE][Fc.ITEM]

941 answers = bundle_json[Fc.ENTRY][2][Fc.RESOURCE][Fc.ITEM]

942

943 # Question text

944 self.assertIn(

945 "1. Feeling nervous, anxious or on edge", questions[0][Fc.TEXT]

946 )

947 # Comment string

948 self.assertIn(

949 "Q1, nervous/anxious/on edge (0 not at all - 3 nearly every day)",

950 questions[0][Fc.TEXT],

951 )

952

953 self.assertIn(

954 "1. Feeling nervous, anxious or on edge", answers[0][Fc.TEXT]

955 )

956 self.assertIn(

957 "Q1, nervous/anxious/on edge (0 not at all - 3 nearly every day)",

958 answers[0][Fc.TEXT],

959 )

960

961 self.assertEqual(answers[0][Fc.ANSWER][0][Fc.VALUE_INTEGER], 0)

962 self.assertEqual(answers[1][Fc.ANSWER][0][Fc.VALUE_INTEGER], 1)

963 self.assertEqual(answers[2][Fc.ANSWER][0][Fc.VALUE_INTEGER], 2)

964 self.assertEqual(answers[3][Fc.ANSWER][0][Fc.VALUE_INTEGER], 3)

Coverage for cc_modules/tests/cc_fhir_tests.py: 16%

403 statements