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

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
3"""camcops_server/cc_modules/tests/cc_fhir_tests.py
5===============================================================================
7 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
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 <http://www.gnu.org/licenses/>.
24===============================================================================
26"""
28import datetime
29import json
30from typing import Dict
31from unittest import mock
33import pendulum
34from requests.exceptions import HTTPError
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
51# =============================================================================
52# Integration testing
53# =============================================================================
56class MockFhirTaskExporter(FhirTaskExporter):
57 pass
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 )
68class FhirExportTestCase(DemoDatabaseTestCase):
69 def setUp(self) -> None:
70 super().setUp()
71 recipientinfo = ExportRecipientInfo()
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"
77 # auto increment doesn't work for BigInteger with SQLite
78 self.recipient.id = 1
79 self.recipient.recipient_name = "test"
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 )
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()
116 def test_patient_exported(self) -> None:
117 exported_task = ExportedTask(task=self.task, recipient=self.recipient)
118 exported_task_fhir = ExportedTaskFhir(exported_task)
120 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
122 response_json = {
123 'type': 'transaction-response',
124 }
126 with mock.patch.object(
127 exporter.client.server, "post_json",
128 return_value=MockFhirResponse(response_json)) as mock_post:
129 exporter.export_task()
131 args, kwargs = mock_post.call_args
133 sent_json = args[1]
135 self.assertEqual(sent_json["resourceType"], "Bundle")
136 self.assertEqual(sent_json["type"], "transaction")
138 patient = sent_json["entry"][0]["resource"]
139 self.assertEqual(patient["resourceType"], "Patient")
141 identifier = patient["identifier"]
142 which_idnum = self.patient_rio.which_idnum
143 idnum_value = self.patient_rio.idnum_value
145 iddef_url = f"http://127.0.0.1:8000/fhir_patient_id/{which_idnum}"
147 self.assertEqual(identifier[0]["system"], iddef_url)
148 self.assertEqual(identifier[0]["value"], str(idnum_value))
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")
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 )
162 def test_questionnaire_exported(self) -> None:
163 exported_task = ExportedTask(task=self.task, recipient=self.recipient)
164 exported_task_fhir = ExportedTaskFhir(exported_task)
166 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
168 response_json = {
169 'type': 'transaction-response',
170 }
172 with mock.patch.object(
173 exporter.client.server, "post_json",
174 return_value=MockFhirResponse(response_json)) as mock_post:
175 exporter.export_task()
177 args, kwargs = mock_post.call_args
179 sent_json = args[1]
181 questionnaire = sent_json["entry"][1]["resource"]
182 self.assertEqual(questionnaire["resourceType"], "Questionnaire")
183 self.assertEqual(questionnaire["status"], "active")
185 identifier = questionnaire["identifier"]
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")
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")
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)
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 )
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)
220 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
222 response_json = {
223 'type': 'transaction-response',
224 }
226 with mock.patch.object(
227 exporter.client.server, "post_json",
228 return_value=MockFhirResponse(response_json)) as mock_post:
229 exporter.export_task()
231 args, kwargs = mock_post.call_args
233 sent_json = args[1]
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")
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
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))
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 )
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)
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)
280 self.assertEqual(len(response["item"]), 10)
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)
288 exported_task_fhir = ExportedTaskFhir(exported_task)
289 self.dbsession.add(exported_task_fhir)
291 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
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 }
329 with mock.patch.object(exporter.client.server, "post_json",
330 return_value=MockFhirResponse(response_json)):
331 exporter.export_task()
333 self.dbsession.commit()
335 entries = exported_task_fhir.entries
337 entries.sort(key=lambda e: e.location)
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")
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")
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))
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)
358 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
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 )
366 with self.assertRaises(FhirExportException) as cm:
367 exporter.export_task()
369 message = str(cm.exception)
370 self.assertIn("Something is not implemented", message)
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)
376 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
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()
387 message = str(cm.exception)
388 self.assertIn("Something bad happened", message)
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)
394 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
396 exporter.client.server = None
397 with self.assertRaises(FhirExportException) as cm:
398 exporter.export_task()
400 message = str(cm.exception)
401 self.assertIn("Cannot create a resource without a server", message)
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)
408 with self.assertRaises(FhirExportException) as cm:
409 FhirTaskExporter(self.req, exported_task_fhir)
411 message = str(cm.exception)
412 self.assertIn("must be initialized with `base_uri`", message)
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"
426 self.task.save_with_next_available_id(self.req,
427 self.server_device.id)
428 self.dbsession.commit()
430 def test_questionnaire_exported(self) -> None:
431 exported_task = ExportedTask(task=self.task, recipient=self.recipient)
432 exported_task_fhir = ExportedTaskFhir(exported_task)
434 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
436 response_json = {
437 'type': 'transaction-response',
438 }
440 with mock.patch.object(
441 exporter.client.server, "post_json",
442 return_value=MockFhirResponse(response_json)) as mock_post:
443 exporter.export_task()
445 args, kwargs = mock_post.call_args
447 sent_json = args[1]
449 questionnaire = sent_json["entry"][0]["resource"]
450 self.assertEqual(questionnaire["resourceType"], "Questionnaire")
451 self.assertEqual(questionnaire["status"], "active")
453 identifier = questionnaire["identifier"]
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")
459 q_datetime = questionnaire["item"][0]
461 q1_choice = questionnaire["item"][1]
462 q2_choice = questionnaire["item"][2]
463 q3_choice = questionnaire["item"][3]
465 q1_satisfaction = questionnaire["item"][4]
466 q2_satisfaction = questionnaire["item"][5]
468 # q_datetime
469 self.assertEqual(q_datetime["linkId"], "q_datetime")
470 self.assertEqual(q_datetime["text"],
471 "Date & Time the Assessment Tool was Completed")
472 self.assertEqual(q_datetime["type"], "dateTime")
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")
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")
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")
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")
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")
516 self.assertEqual(len(questionnaire["item"]), 6)
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 )
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)
530 exporter = MockFhirTaskExporter(self.req, exported_task_fhir)
532 response_json = {
533 'type': 'transaction-response',
534 }
536 with mock.patch.object(
537 exporter.client.server, "post_json",
538 return_value=MockFhirResponse(response_json)) as mock_post:
539 exporter.export_task()
541 args, kwargs = mock_post.call_args
543 sent_json = args[1]
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")
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 )
562 q_datetime = response["item"][0]
564 q1_choice = response["item"][1]
565 q2_choice = response["item"][2]
566 q3_choice = response["item"][3]
568 q1_satisfaction = response["item"][4]
569 q2_satisfaction = response["item"][5]
571 # q_datetime
572 self.assertEqual(q_datetime["linkId"], "q_datetime")
573 self.assertEqual(q_datetime["text"],
574 "Date & 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())
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)
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)
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)
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)
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)
628 self.assertEqual(len(response["item"]), 6)