Coverage for cc_modules/tests/client_api_tests.py: 99%
1032 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 15:55 +0100
1"""
2camcops_server/cc_modules/tests/client_api_tests.py
4===============================================================================
6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry.
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 json
29import string
30from typing import Dict
31from unittest import mock, TestCase
33from cardinal_pythonlib.classes import class_attribute_names
34from cardinal_pythonlib.convert import (
35 base64_64format_encode,
36 hex_xformat_encode,
37)
38from cardinal_pythonlib.nhs import generate_random_nhs_number
39from cardinal_pythonlib.sql.literals import sql_quote_string
40from cardinal_pythonlib.text import escape_newlines, unescape_newlines
41from pendulum import DateTime as Pendulum, Duration, local, now, parse
42from pyramid.response import Response
43from semantic_version import Version
44from sqlalchemy import select
46from camcops_server.cc_modules.cc_all_models import CLIENT_TABLE_MAP
47from camcops_server.cc_modules.cc_cache import cache_region_static
48from camcops_server.cc_modules.cc_client_api_core import (
49 AllowedTablesFieldNames,
50 ExtraStringFieldNames,
51 fail_server_error,
52 fail_unsupported_operation,
53 fail_user_error,
54 ServerErrorException,
55 TabletParam,
56 UserErrorException,
57)
58from camcops_server.cc_modules.cc_constants import ERA_NOW
59from camcops_server.cc_modules.cc_convert import decode_values
60from camcops_server.cc_modules.cc_device import Device
61from camcops_server.cc_modules.cc_dirtytables import DirtyTable
62from camcops_server.cc_modules.cc_patientidnum import PatientIdNum
63from camcops_server.cc_modules.cc_proquint import uuid_from_proquint
64from camcops_server.cc_modules.cc_taskindex import (
65 PatientIdNumIndexEntry,
66 TaskIndexEntry,
67)
68from camcops_server.cc_modules.cc_testfactories import (
69 DeviceFactory,
70 DirtyTableFactory,
71 Fake,
72 GroupFactory,
73 NHSIdNumDefinitionFactory,
74 NHSPatientIdNumFactory,
75 PatientFactory,
76 PatientTaskScheduleFactory,
77 ServerCreatedNHSPatientIdNumFactory,
78 ServerCreatedPatientFactory,
79 TaskScheduleFactory,
80 TaskScheduleItemFactory,
81 UserFactory,
82 UserGroupMembershipFactory,
83)
84from camcops_server.cc_modules.cc_unittest import DemoRequestTestCase
85from camcops_server.cc_modules.cc_user import User
86from camcops_server.cc_modules.cc_version import (
87 FIRST_TABLET_VER_WITH_EXPLICIT_PKNAME_IN_UPLOAD_TABLE,
88)
89from camcops_server.cc_modules.cc_validators import (
90 validate_alphanum_underscore,
91)
92from camcops_server.cc_modules.client_api import (
93 client_api,
94 FAILURE_CODE,
95 get_or_create_single_user,
96 make_single_user_mode_username,
97 Operations,
98 SUCCESS_CODE,
99)
100from camcops_server.tasks import Bmi
101from camcops_server.tasks.tests.factories import BmiFactory, Phq9Factory
103TEST_NHS_NUMBER = generate_random_nhs_number()
106def get_reply_dict_from_response(response: Response) -> Dict[str, str]:
107 """
108 For unit testing: convert the text in a :class:`Response` back to a
109 dictionary, so we can check it was correct.
110 """
111 txt = str(response)
112 d = {} # type: Dict[str, str]
113 # Format is: "200 OK\r\n<other headers>\r\n\r\n<content>"
114 # There's a blank line between the heads and the body.
115 http_gap = "\r\n\r\n"
116 camcops_linesplit = "\n"
117 camcops_k_v_sep = ":"
118 try:
119 start_of_content = txt.index(http_gap) + len(http_gap)
120 txt = txt[start_of_content:]
121 for line in txt.split(camcops_linesplit):
122 if not line:
123 continue
124 colon_pos = line.index(camcops_k_v_sep)
125 key = line[:colon_pos]
126 value = line[colon_pos + len(camcops_k_v_sep) :]
127 key = key.strip()
128 value = value.strip()
129 d[key] = value
130 return d
131 except ValueError:
132 return {}
135class ExceptionTests(TestCase):
136 def test_fail_user_error_raises(self) -> None:
137 with self.assertRaises(UserErrorException):
138 fail_user_error("testmsg")
140 def test_fail_server_error_raises(self) -> None:
141 with self.assertRaises(ServerErrorException):
142 fail_server_error("testmsg")
144 def test_fail_unsupported_operation_raises(self) -> None:
145 with self.assertRaises(UserErrorException):
146 fail_unsupported_operation("duffop")
149class EncodeDecodeValuesTests(TestCase):
150 def test_values_decoded_correctly(self) -> None:
151 # Encoding/decoding tests
152 # data = bytearray("hello")
153 data = b"hello"
154 enc_b64data = base64_64format_encode(data)
155 enc_hexdata = hex_xformat_encode(data)
156 not_enc_1 = "X'012345'"
157 not_enc_2 = "64'aGVsbG8='"
158 teststring = """one, two, 3, 4.5, NULL, 'hello "hi
159 with linebreak"', 'NULL', 'quote''s here', {b}, {h}, {s1}, {s2}"""
160 sql_csv_testdict = {
161 teststring.format(
162 b=enc_b64data,
163 h=enc_hexdata,
164 s1=sql_quote_string(not_enc_1),
165 s2=sql_quote_string(not_enc_2),
166 ): [
167 "one",
168 "two",
169 3,
170 4.5,
171 None,
172 'hello "hi\n with linebreak"',
173 "NULL",
174 "quote's here",
175 data,
176 data,
177 not_enc_1,
178 not_enc_2,
179 ],
180 "": [],
181 }
182 for k, v in sql_csv_testdict.items():
183 r = decode_values(k)
184 self.assertEqual(
185 r,
186 v,
187 "Mismatch! Result: {r!s}\n"
188 "Should have been: {v!s}\n"
189 "Key was: {k!s}".format(r=r, v=v, k=k),
190 )
193class EscapeUnescapeNewlinesTests(TestCase):
194 def test_escapes_and_unescapes_correctly(self) -> None:
196 # Newline encoding/decodine
197 test_string = (
198 "slash \\ newline \n ctrl_r \r special \\n other special \\r "
199 "quote ' doublequote \" "
200 )
201 self.assertEqual(
202 unescape_newlines(escape_newlines(test_string)),
203 test_string,
204 "Bug in escape_newlines() or unescape_newlines()",
205 )
208class ValidateAlphanumUnderscoreTests(TestCase):
209 def test_class_attribute_names_validate(self) -> None:
210 for x in class_attribute_names(Operations):
211 try:
212 request = None
213 validate_alphanum_underscore(x, request)
214 except ValueError:
215 self.fail(f"Operations.{x} fails validate_alphanum_underscore")
218class ClientApiTestCase(DemoRequestTestCase):
219 def setUp(self) -> None:
220 super().setUp()
222 self.group = GroupFactory()
223 self.user = self.req._debugging_user = UserFactory(
224 upload_group_id=self.group.id,
225 )
227 UserGroupMembershipFactory(
228 user_id=self.user.id,
229 group_id=self.group.id,
230 may_upload=True,
231 may_register_devices=True,
232 )
233 # Ensure the server device exists so that we don't get ID clashes
234 Device.get_server_device(self.dbsession)
235 self.device = DeviceFactory()
237 self.post_dict = {
238 TabletParam.CAMCOPS_VERSION: (
239 FIRST_TABLET_VER_WITH_EXPLICIT_PKNAME_IN_UPLOAD_TABLE
240 ),
241 TabletParam.DEVICE: self.device.name,
242 }
243 # To prevent test failures when mocking cached functions
244 cache_region_static.configure(
245 backend="dogpile.cache.null", replace_existing_backend=True
246 )
248 def call_api(self) -> Dict[str, str]:
249 self.req.fake_request_post_from_dict(self.post_dict)
250 response = client_api(self.req)
251 return get_reply_dict_from_response(response)
254# class MalformedRequestTests(ClientApiTestCase):
255# def test_returns_400_bad_request(self) -> None:
256# self.req.set_post_body(b"rubbish")
257# response = client_api(self.req)
259# self.assertEqual(response.status, "400 Bad Request")
260# self.assertIn(b"Not a valid CamCOPS API request", response.body)
263class OpRegisterPatientTests(ClientApiTestCase):
264 def setUp(self) -> None:
265 super().setUp()
267 self.patient = ServerCreatedPatientFactory()
268 self.idnum = ServerCreatedNHSPatientIdNumFactory(patient=self.patient)
269 PatientIdNumIndexEntry.index_idnum(self.idnum, self.dbsession)
271 proquint = self.patient.uuid_as_proquint
273 self.post_dict[TabletParam.OPERATION] = Operations.REGISTER_PATIENT
274 self.post_dict[TabletParam.PATIENT_PROQUINT] = proquint
276 def test_returns_patient_info(self) -> None:
277 reply_dict = self.call_api()
279 self.assertEqual(
280 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
281 )
283 patient_dict = json.loads(reply_dict[TabletParam.PATIENT_INFO])[0]
285 self.assertEqual(
286 patient_dict[TabletParam.SURNAME], self.patient.surname
287 )
288 self.assertEqual(
289 patient_dict[TabletParam.FORENAME], self.patient.forename
290 )
291 self.assertEqual(patient_dict[TabletParam.SEX], self.patient.sex)
292 self.assertEqual(
293 patient_dict[TabletParam.DOB], self.patient.dob.isoformat()
294 )
295 self.assertEqual(
296 patient_dict[TabletParam.ADDRESS], self.patient.address
297 )
298 self.assertEqual(patient_dict[TabletParam.GP], self.patient.gp)
299 self.assertEqual(patient_dict[TabletParam.OTHER], self.patient.other)
300 self.assertEqual(
301 patient_dict[f"idnum{self.idnum.which_idnum}"],
302 self.idnum.idnum_value,
303 )
305 def test_creates_user(self) -> None:
306 reply_dict = self.call_api()
307 self.assertEqual(
308 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
309 )
311 username = reply_dict[TabletParam.USER]
312 self.assertEqual(
313 username,
314 make_single_user_mode_username(self.device.name, self.patient._pk),
315 )
316 password = reply_dict[TabletParam.PASSWORD]
317 self.assertEqual(len(password), 32)
319 valid_chars = string.ascii_letters + string.digits + string.punctuation
320 self.assertTrue(all(c in valid_chars for c in password))
322 user = self.dbsession.execute(
323 select(User).where(User.username == username)
324 ).scalar_one()
325 self.assertEqual(user.upload_group, self.patient.group)
326 self.assertTrue(user.auto_generated)
327 self.assertTrue(user.may_register_devices)
328 self.assertTrue(user.may_upload)
330 def test_does_not_create_user_when_name_exists(self) -> None:
331 single_user_username = make_single_user_mode_username(
332 self.device.name, self.patient._pk
333 )
335 UserFactory(
336 username=single_user_username,
337 password="old password",
338 password__request=self.req,
339 )
341 reply_dict = self.call_api()
343 self.assertEqual(
344 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
345 )
347 username = reply_dict[TabletParam.USER]
348 self.assertEqual(
349 username,
350 make_single_user_mode_username(self.device.name, self.patient._pk),
351 )
352 password = reply_dict[TabletParam.PASSWORD]
353 self.assertEqual(len(password), 32)
355 valid_chars = string.ascii_letters + string.digits + string.punctuation
356 self.assertTrue(all(c in valid_chars for c in password))
358 user = self.dbsession.execute(
359 select(User).where(User.username == username)
360 ).scalar_one()
361 self.assertEqual(user.upload_group, self.patient.group)
362 self.assertTrue(user.auto_generated)
363 self.assertTrue(user.may_register_devices)
364 self.assertTrue(user.may_upload)
366 def test_raises_for_invalid_proquint(self) -> None:
367 self.post_dict[TabletParam.PATIENT_PROQUINT] = "invalid"
368 reply_dict = self.call_api()
369 self.assertEqual(
370 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
371 )
372 self.assertIn(
373 "no patient with access key 'invalid'",
374 reply_dict[TabletParam.ERROR],
375 )
377 def test_raises_for_missing_valid_proquint(self) -> None:
378 valid_proquint = "sazom-diliv-navol-hubot-mufur-mamuv-kojus-loluv-v"
380 # Error message is same as for invalid proquint so make sure our
381 # test proquint really is valid (should not raise)
382 uuid_from_proquint(valid_proquint)
384 self.post_dict[TabletParam.PATIENT_PROQUINT] = valid_proquint
385 reply_dict = self.call_api()
387 self.assertEqual(
388 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
389 )
390 self.assertIn(
391 f"no patient with access key '{valid_proquint}'",
392 reply_dict[TabletParam.ERROR],
393 )
395 def test_raises_when_no_patient_idnums(self) -> None:
396 # In theory this shouldn't be possible in normal operation as the
397 # patient cannot be created without any idnums
398 patient = ServerCreatedPatientFactory()
400 proquint = patient.uuid_as_proquint
401 self.post_dict[TabletParam.PATIENT_PROQUINT] = proquint
402 reply_dict = self.call_api()
403 self.assertEqual(
404 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
405 )
406 self.assertIn(
407 "Patient has no ID numbers", reply_dict[TabletParam.ERROR]
408 )
410 def test_raises_when_patient_not_created_on_server(self) -> None:
411 patient = PatientFactory()
413 proquint = patient.uuid_as_proquint
414 self.post_dict[TabletParam.PATIENT_PROQUINT] = proquint
415 reply_dict = self.call_api()
416 self.assertEqual(
417 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
418 )
419 self.assertIn(
420 f"no patient with access key '{proquint}'",
421 reply_dict[TabletParam.ERROR],
422 )
424 def test_returns_ip_use_flags(self) -> None:
425 ip_use = self.patient.group.ip_use
427 reply_dict = self.call_api()
429 self.assertEqual(
430 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
431 )
433 ip_use_info = json.loads(reply_dict[TabletParam.IP_USE_INFO])
435 self.assertEqual(
436 ip_use_info[TabletParam.IP_USE_COMMERCIAL], ip_use.commercial
437 )
438 self.assertEqual(
439 ip_use_info[TabletParam.IP_USE_CLINICAL], ip_use.clinical
440 )
441 self.assertEqual(
442 ip_use_info[TabletParam.IP_USE_EDUCATIONAL], ip_use.educational
443 )
444 self.assertEqual(
445 ip_use_info[TabletParam.IP_USE_RESEARCH], ip_use.research
446 )
449class OpGetTaskSchedulesTests(ClientApiTestCase):
450 def test_returns_task_schedules(self) -> None:
451 schedule1 = TaskScheduleFactory(group=self.group)
452 schedule2 = TaskScheduleFactory(group=self.group)
454 TaskScheduleItemFactory(
455 task_schedule=schedule1,
456 task_table_name="phq9",
457 due_from=Duration(days=0),
458 due_by=Duration(days=7),
459 )
460 TaskScheduleItemFactory(
461 task_schedule=schedule1,
462 task_table_name="bmi",
463 due_from=Duration(days=0),
464 due_by=Duration(days=8),
465 )
466 TaskScheduleItemFactory(
467 task_schedule=schedule1,
468 task_table_name="phq9",
469 due_from=Duration(days=30),
470 due_by=Duration(days=37),
471 )
472 TaskScheduleItemFactory(
473 task_schedule=schedule1,
474 task_table_name="gmcpq",
475 due_from=Duration(days=30),
476 due_by=Duration(days=38),
477 )
479 # This is the patient originally created om the server
480 server_patient = ServerCreatedPatientFactory(_group=self.group)
481 server_idnum = ServerCreatedNHSPatientIdNumFactory(
482 patient=server_patient
483 )
485 # This is the same patient but from the device
486 patient = PatientFactory(_group=self.group)
487 idnum = NHSPatientIdNumFactory(
488 patient=patient,
489 which_idnum=server_idnum.which_idnum,
490 idnum_value=server_idnum.idnum_value,
491 )
492 PatientIdNumIndexEntry.index_idnum(idnum, self.dbsession)
494 PatientTaskScheduleFactory(
495 patient=server_patient,
496 task_schedule=schedule1,
497 settings={
498 "bmi": {"bmi_key": "bmi_value"},
499 "phq9": {"phq9_key": "phq9_value"},
500 },
501 start_datetime=local(2020, 7, 31),
502 )
504 PatientTaskScheduleFactory(
505 patient=server_patient,
506 task_schedule=schedule2,
507 )
509 bmi = BmiFactory(
510 patient=patient,
511 when_created=local(2020, 8, 1),
512 )
513 self.assertTrue(bmi.is_complete())
515 TaskIndexEntry.index_task(
516 bmi, self.dbsession, indexed_at_utc=Pendulum.utcnow()
517 )
519 proquint = server_patient.uuid_as_proquint
521 # For type checker
522 assert proquint is not None
524 self.post_dict[TabletParam.OPERATION] = Operations.GET_TASK_SCHEDULES
525 self.post_dict[TabletParam.PATIENT_PROQUINT] = proquint
527 reply_dict = self.call_api()
529 self.assertEqual(
530 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
531 )
533 task_schedules = json.loads(reply_dict[TabletParam.TASK_SCHEDULES])
535 self.assertEqual(len(task_schedules), 2)
537 s = task_schedules[0]
538 self.assertEqual(s[TabletParam.TASK_SCHEDULE_NAME], schedule1.name)
540 schedule_items = s[TabletParam.TASK_SCHEDULE_ITEMS]
541 self.assertEqual(len(schedule_items), 4)
543 phq9_1_sched = schedule_items[0]
544 self.assertEqual(phq9_1_sched[TabletParam.TABLE], "phq9")
545 self.assertEqual(
546 phq9_1_sched[TabletParam.SETTINGS], {"phq9_key": "phq9_value"}
547 )
548 self.assertEqual(
549 parse(phq9_1_sched[TabletParam.DUE_FROM]), local(2020, 7, 31)
550 )
551 self.assertEqual(
552 parse(phq9_1_sched[TabletParam.DUE_BY]), local(2020, 8, 7)
553 )
554 self.assertFalse(phq9_1_sched[TabletParam.COMPLETE])
555 self.assertFalse(phq9_1_sched[TabletParam.ANONYMOUS])
557 bmi_sched = schedule_items[1]
558 self.assertEqual(bmi_sched[TabletParam.TABLE], "bmi")
559 self.assertEqual(
560 bmi_sched[TabletParam.SETTINGS], {"bmi_key": "bmi_value"}
561 )
562 self.assertEqual(
563 parse(bmi_sched[TabletParam.DUE_FROM]), local(2020, 7, 31)
564 )
565 self.assertEqual(
566 parse(bmi_sched[TabletParam.DUE_BY]), local(2020, 8, 8)
567 )
568 self.assertTrue(bmi_sched[TabletParam.COMPLETE])
569 self.assertFalse(bmi_sched[TabletParam.ANONYMOUS])
571 phq9_2_sched = schedule_items[2]
572 self.assertEqual(phq9_2_sched[TabletParam.TABLE], "phq9")
573 self.assertEqual(
574 phq9_2_sched[TabletParam.SETTINGS], {"phq9_key": "phq9_value"}
575 )
576 self.assertEqual(
577 parse(phq9_2_sched[TabletParam.DUE_FROM]), local(2020, 8, 30)
578 )
579 self.assertEqual(
580 parse(phq9_2_sched[TabletParam.DUE_BY]), local(2020, 9, 6)
581 )
582 self.assertFalse(phq9_2_sched[TabletParam.COMPLETE])
583 self.assertFalse(phq9_2_sched[TabletParam.ANONYMOUS])
585 # GMCPQ
586 gmcpq_sched = schedule_items[3]
587 self.assertTrue(gmcpq_sched[TabletParam.ANONYMOUS])
590class OpGetOrCreateSingleUserTests(DemoRequestTestCase):
591 def setUp(self) -> None:
592 super().setUp()
594 self.patient = PatientFactory()
595 self.req._debugging_user = UserFactory()
597 def test_user_is_added_to_patient_group(self) -> None:
598 user, _ = get_or_create_single_user(self.req, "test", self.patient)
599 self.dbsession.flush()
601 self.assertIn(self.patient.group.id, user.group_ids)
603 def test_user_is_created_with_username(self) -> None:
604 user, _ = get_or_create_single_user(self.req, "test", self.patient)
605 self.dbsession.flush()
607 self.assertEqual(user.username, "test")
609 def test_user_is_assigned_password(self) -> None:
610 _, password = get_or_create_single_user(self.req, "test", self.patient)
611 self.dbsession.flush()
613 valid_chars = string.ascii_letters + string.digits + string.punctuation
614 self.assertTrue(all(c in valid_chars for c in password))
616 def test_user_upload_group_set(self) -> None:
617 user, _ = get_or_create_single_user(self.req, "test", self.patient)
618 self.dbsession.flush()
620 self.assertEqual(user.upload_group, self.patient.group)
622 def test_user_auto_generated_flag_set(self) -> None:
623 user, _ = get_or_create_single_user(self.req, "test", self.patient)
624 self.dbsession.flush()
626 self.assertTrue(user.auto_generated)
628 def test_user_is_not_superuser(self) -> None:
629 user, _ = get_or_create_single_user(self.req, "test", self.patient)
630 self.dbsession.flush()
632 self.assertFalse(user.superuser)
634 def test_single_patient_pk_set(self) -> None:
635 user, _ = get_or_create_single_user(self.req, "test", self.patient)
636 self.dbsession.flush()
638 self.assertEqual(user.single_patient_pk, self.patient._pk)
640 def test_user_may_register_devices(self) -> None:
641 user, _ = get_or_create_single_user(self.req, "test", self.patient)
642 self.dbsession.flush()
644 self.assertTrue(user.user_group_memberships[0].may_register_devices)
646 def test_user_may_upload(self) -> None:
647 user, _ = get_or_create_single_user(self.req, "test", self.patient)
648 self.dbsession.flush()
650 self.assertTrue(user.user_group_memberships[0].may_upload)
652 def test_existing_user_is_updated(self) -> None:
653 existing_user = UserFactory(username="test")
655 user, _ = get_or_create_single_user(self.req, "test", self.patient)
656 self.dbsession.flush()
658 self.assertEqual(user, existing_user)
661class OpUploadEntireDatabaseTests(ClientApiTestCase):
662 def setUp(self) -> None:
663 super().setUp()
664 self.post_dict[TabletParam.OPERATION] = (
665 Operations.UPLOAD_ENTIRE_DATABASE
666 )
667 self.post_dict[TabletParam.FINALIZING] = 1
669 def test_fails_if_pknameinfo_is_not_a_dict(self) -> None:
670 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(
671 [{"key": "valid JSON but list not dict"}]
672 )
674 reply_dict = self.call_api()
676 self.assertEqual(
677 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
678 )
679 self.assertIn(
680 "PK name info JSON is not a dict", reply_dict[TabletParam.ERROR]
681 )
683 def test_fails_if_databasedata_is_not_a_dict(self) -> None:
684 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(
685 {"key": "valid JSON"}
686 )
687 self.post_dict[TabletParam.DBDATA] = json.dumps(
688 [{"key": "valid JSON but list not dict"}]
689 )
691 reply_dict = self.call_api()
693 self.assertEqual(
694 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
695 )
696 self.assertIn(
697 "Database data JSON is not a dict", reply_dict[TabletParam.ERROR]
698 )
700 def test_fails_if_table_names_do_not_match(self) -> None:
701 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(
702 {"table1": "", "table2": "", "table3": ""}
703 )
704 self.post_dict[TabletParam.DBDATA] = json.dumps(
705 {"table4": "", "table5": "", "table6": ""}
706 )
708 reply_dict = self.call_api()
710 self.assertEqual(
711 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
712 )
713 self.assertIn("Table names don't match", reply_dict[TabletParam.ERROR])
715 def test_fails_if_table_names_do_not_exist(self) -> None:
716 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(
717 {"table1": "", "table2": "", "table3": ""}
718 )
719 self.post_dict[TabletParam.DBDATA] = json.dumps(
720 {"table1": "", "table2": "", "table3": ""}
721 )
723 reply_dict = self.call_api()
725 self.assertEqual(
726 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
727 )
728 self.assertIn(
729 "Attempt to upload nonexistent tables",
730 reply_dict[TabletParam.ERROR],
731 )
733 def test_empty_upload_succeeds(self) -> None:
734 pknameinfo = {key: "" for key in CLIENT_TABLE_MAP.keys()}
735 dbdata = {key: "" for key in CLIENT_TABLE_MAP.keys()}
737 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(pknameinfo)
738 self.post_dict[TabletParam.DBDATA] = json.dumps(dbdata)
740 reply_dict = self.call_api()
741 self.assertEqual(
742 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
743 )
745 def test_upload_row_succeeds(self) -> None:
746 now_utc_string = now("UTC").isoformat()
747 patient = PatientFactory(_device=self.device)
748 bmi_data = {
749 "id": "1",
750 "height_m": "1.83",
751 "mass_kg": "67",
752 "when_created": now_utc_string,
753 "when_last_modified": now_utc_string,
754 "_move_off_tablet": "1",
755 "patient_id": str(patient.id),
756 }
758 pknameinfo = {"bmi": "id"}
759 dbdata = {"bmi": [bmi_data]}
761 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(pknameinfo)
762 self.post_dict[TabletParam.DBDATA] = json.dumps(dbdata)
764 reply_dict = self.call_api()
765 self.assertEqual(
766 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
767 )
768 bmi = self.dbsession.execute(
769 select(Bmi).where(Bmi.id == 1)
770 ).scalar_one()
772 self.assertAlmostEqual(bmi.height_m, 1.83)
773 self.assertAlmostEqual(bmi.mass_kg, 67)
775 def test_upload_row_fails_with_no_pkname(self) -> None:
776 now_utc_string = now("UTC").isoformat()
777 patient = PatientFactory(_device=self.device)
778 bmi_data = {
779 "id": "1",
780 "height_m": "1.83",
781 "mass_kg": "67",
782 "when_created": now_utc_string,
783 "when_last_modified": now_utc_string,
784 "_move_off_tablet": "1",
785 "patient_id": str(patient.id),
786 }
788 pknameinfo = {"bmi": ""}
789 dbdata = {"bmi": [bmi_data]}
791 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(pknameinfo)
792 self.post_dict[TabletParam.DBDATA] = json.dumps(dbdata)
794 reply_dict = self.call_api()
795 self.assertEqual(
796 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
797 )
798 self.assertIn(
799 "Client-side PK name not specified by client for non-empty table",
800 reply_dict[TabletParam.ERROR],
801 msg=reply_dict,
802 )
804 def test_empty_upload_flags_existing_for_deletion(self) -> None:
805 patient = PatientFactory(_device=self.device)
806 bmi = BmiFactory(
807 patient=patient,
808 _removal_pending=True,
809 _era=ERA_NOW,
810 _current=True,
811 )
813 pknameinfo = {key: "" for key in CLIENT_TABLE_MAP.keys()}
814 dbdata = {key: "" for key in CLIENT_TABLE_MAP.keys()}
816 self.post_dict[TabletParam.PKNAMEINFO] = json.dumps(pknameinfo)
817 self.post_dict[TabletParam.DBDATA] = json.dumps(dbdata)
819 reply_dict = self.call_api()
820 self.assertEqual(
821 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
822 )
823 self.dbsession.commit()
825 self.assertFalse(bmi._current)
826 self.assertFalse(bmi._removal_pending)
827 self.assertTrue(bmi._removing_user_id, self.user.id)
828 self.assertIsNotNone(bmi._when_removed_exact)
829 self.assertIsNotNone(bmi._when_removed_batch_utc)
832class OpValidatePatientsTests(ClientApiTestCase):
833 def setUp(self) -> None:
834 super().setUp()
836 self.post_dict[TabletParam.OPERATION] = Operations.VALIDATE_PATIENTS
838 def test_fails_if_patient_info_is_not_a_list(self) -> None:
839 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps({})
841 reply_dict = self.call_api()
843 self.assertEqual(
844 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
845 )
846 self.assertIn(
847 "Top-level JSON is not a list", reply_dict[TabletParam.ERROR]
848 )
850 def test_succeeds_for_empty_list(self) -> None:
851 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps([])
853 reply_dict = self.call_api()
855 self.assertEqual(
856 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
857 )
859 def test_fails_if_one_patients_info_is_not_a_dict(self) -> None:
860 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps([[]])
862 reply_dict = self.call_api()
864 self.assertEqual(
865 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
866 )
867 self.assertIn(
868 "Patient JSON is not a dict", reply_dict[TabletParam.ERROR]
869 )
871 def test_fails_if_one_patients_info_is_empty(self) -> None:
872 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps([{}])
874 reply_dict = self.call_api()
876 self.assertEqual(
877 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
878 )
879 self.assertIn("Patient JSON is empty", reply_dict[TabletParam.ERROR])
881 def test_fails_if_forename_is_not_a_string(self) -> None:
882 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
883 [{TabletParam.FORENAME: 1}]
884 )
886 reply_dict = self.call_api()
888 self.assertEqual(
889 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
890 )
891 self.assertIn("non-string: 1", reply_dict[TabletParam.ERROR])
893 def test_fails_if_surname_is_not_a_string(self) -> None:
894 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
895 [{TabletParam.SURNAME: 2}]
896 )
898 reply_dict = self.call_api()
900 self.assertEqual(
901 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
902 )
903 self.assertIn("non-string: 2", reply_dict[TabletParam.ERROR])
905 def test_fails_if_sex_is_not_valid(self) -> None:
906 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
907 [{TabletParam.SEX: "Q"}]
908 )
910 reply_dict = self.call_api()
912 self.assertEqual(
913 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
914 )
915 self.assertIn("Bad sex value: 'Q'", reply_dict[TabletParam.ERROR])
917 def test_fails_if_dob_is_not_a_string(self) -> None:
918 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
919 [{TabletParam.DOB: 3}]
920 )
922 reply_dict = self.call_api()
924 self.assertEqual(
925 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
926 )
927 self.assertIn("non-string: 3", reply_dict[TabletParam.ERROR])
929 def test_fails_if_dob_fails_to_parse(self) -> None:
930 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
931 [{TabletParam.DOB: "Yesterday"}]
932 )
934 reply_dict = self.call_api()
936 self.assertEqual(
937 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
938 )
939 self.assertIn(
940 "Invalid DOB: 'Yesterday'", reply_dict[TabletParam.ERROR]
941 )
943 def test_fails_if_email_is_not_a_string(self) -> None:
944 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
945 [{TabletParam.EMAIL: 4}]
946 )
948 reply_dict = self.call_api()
950 self.assertEqual(
951 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
952 )
953 self.assertIn("non-string: 4", reply_dict[TabletParam.ERROR])
955 def test_fails_if_email_invalid(self) -> None:
956 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
957 [{TabletParam.EMAIL: "email"}]
958 )
960 reply_dict = self.call_api()
962 self.assertEqual(
963 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
964 )
965 self.assertIn(
966 "Bad e-mail address: 'email'", reply_dict[TabletParam.ERROR]
967 )
969 def test_fails_if_address_is_not_a_string(self) -> None:
970 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
971 [{TabletParam.ADDRESS: 5}]
972 )
974 reply_dict = self.call_api()
976 self.assertEqual(
977 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
978 )
979 self.assertIn("non-string: 5", reply_dict[TabletParam.ERROR])
981 def test_fails_if_gp_is_not_a_string(self) -> None:
982 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
983 [{TabletParam.GP: 6}]
984 )
986 reply_dict = self.call_api()
988 self.assertEqual(
989 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
990 )
991 self.assertIn("non-string: 6", reply_dict[TabletParam.ERROR])
993 def test_fails_if_other_is_not_a_string(self) -> None:
994 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
995 [{TabletParam.OTHER: 7}]
996 )
998 reply_dict = self.call_api()
1000 self.assertEqual(
1001 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1002 )
1003 self.assertIn("non-string: 7", reply_dict[TabletParam.ERROR])
1005 def test_fails_if_which_idnum_is_not_an_int(self) -> None:
1006 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1007 [{f"{TabletParam.IDNUM_PREFIX}foo": 12345}]
1008 )
1010 reply_dict = self.call_api()
1012 self.assertEqual(
1013 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1014 )
1015 self.assertIn(
1016 "Bad idnum key: 'idnumfoo'", reply_dict[TabletParam.ERROR]
1017 )
1019 def test_fails_if_which_idnum_is_not_valid(self) -> None:
1020 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1021 [{f"{TabletParam.IDNUM_PREFIX}2": 12345}]
1022 )
1024 self.req.valid_which_idnums = [1]
1026 reply_dict = self.call_api()
1028 self.assertEqual(
1029 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1030 )
1031 self.assertIn("Bad ID number type: 2", reply_dict[TabletParam.ERROR])
1033 def test_fails_if_which_idnum_already_seen(self) -> None:
1034 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1035 [
1036 {
1037 f"{TabletParam.IDNUM_PREFIX}1": 12345,
1038 f"{TabletParam.IDNUM_PREFIX}01": 12345,
1039 }
1040 ]
1041 )
1043 self.req.valid_which_idnums = [1]
1045 reply_dict = self.call_api()
1047 self.assertEqual(
1048 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1049 )
1050 self.assertIn(
1051 "More than one ID number supplied for ID number type 1",
1052 reply_dict[TabletParam.ERROR],
1053 )
1055 def test_fails_if_idnum_not_an_int(self) -> None:
1056 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1057 [{f"{TabletParam.IDNUM_PREFIX}1": "foo"}]
1058 )
1060 self.req.valid_which_idnums = [1]
1062 reply_dict = self.call_api()
1064 self.assertEqual(
1065 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1066 )
1067 self.assertIn(
1068 "Bad ID number value: 'foo'", reply_dict[TabletParam.ERROR]
1069 )
1071 def test_fails_if_idref_invalid(self) -> None:
1072 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1073 [{f"{TabletParam.IDNUM_PREFIX}1": 0}]
1074 )
1076 self.req.valid_which_idnums = [1]
1078 reply_dict = self.call_api()
1080 self.assertEqual(
1081 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1082 )
1083 self.assertIn(
1084 "Bad ID number: IdNumReference(idnum_value=0, which_idnum=1)",
1085 reply_dict[TabletParam.ERROR],
1086 )
1088 def test_fails_if_finalizing_is_not_a_bool(self) -> None:
1089 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1090 [{TabletParam.FINALIZING: 123}]
1091 )
1093 reply_dict = self.call_api()
1095 self.assertEqual(
1096 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1097 )
1098 self.assertIn(
1099 "Bad 'finalizing' value: 123", reply_dict[TabletParam.ERROR]
1100 )
1102 def test_fails_for_unknown_json_key(self) -> None:
1103 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1104 [{"foobar": 123}]
1105 )
1107 reply_dict = self.call_api()
1109 self.assertEqual(
1110 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1111 )
1112 self.assertIn(
1113 "Unknown JSON key: 'foobar'", reply_dict[TabletParam.ERROR]
1114 )
1116 def test_fails_for_missing_finalizing_key(self) -> None:
1117 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1118 [{TabletParam.SURNAME: "Valid"}] # Needs to have something
1119 )
1121 reply_dict = self.call_api()
1123 self.assertEqual(
1124 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1125 )
1126 self.assertIn(
1127 "Missing 'finalizing' JSON key", reply_dict[TabletParam.ERROR]
1128 )
1130 def test_fails_when_candidate_invalid_for_group(self) -> None:
1131 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1132 [
1133 {
1134 TabletParam.SURNAME: "Valid", # Needs to have something
1135 TabletParam.FINALIZING: True,
1136 }
1137 ]
1138 )
1140 mock_invalid = mock.Mock(return_value=(False, "Mock reason"))
1142 with mock.patch.multiple(
1143 "camcops_server.cc_modules.client_api",
1144 is_candidate_patient_valid_for_group=mock_invalid,
1145 ):
1146 reply_dict = self.call_api()
1148 self.assertEqual(
1149 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1150 )
1151 self.assertIn("Invalid patient", reply_dict[TabletParam.ERROR])
1152 self.assertIn("Mock reason", reply_dict[TabletParam.ERROR])
1154 def test_fails_when_candidate_invalid_for_restricted_user(self) -> None:
1155 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1156 [
1157 {
1158 TabletParam.SURNAME: "Valid", # Needs to have something
1159 TabletParam.FINALIZING: True,
1160 }
1161 ]
1162 )
1164 mock_valid = mock.Mock(return_value=(True, ""))
1165 mock_invalid = mock.Mock(return_value=(False, "Mock reason"))
1167 with mock.patch.multiple(
1168 "camcops_server.cc_modules.client_api",
1169 is_candidate_patient_valid_for_group=mock_valid,
1170 is_candidate_patient_valid_for_restricted_user=mock_invalid,
1171 ):
1172 reply_dict = self.call_api()
1174 self.assertEqual(
1175 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1176 )
1177 self.assertIn("Invalid patient", reply_dict[TabletParam.ERROR])
1178 self.assertIn("Mock reason", reply_dict[TabletParam.ERROR])
1180 def test_succeeds_for_valid_patient(self) -> None:
1181 sex = Fake.en_gb.sex()
1182 dob = Fake.en_gb.consistent_date_of_birth().isoformat()
1184 # All values set for maximum test coverage
1185 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1186 [
1187 {
1188 TabletParam.FORENAME: Fake.en_gb.forename(sex),
1189 TabletParam.SURNAME: Fake.en_gb.last_name(),
1190 TabletParam.SEX: sex,
1191 TabletParam.DOB: dob,
1192 TabletParam.ADDRESS: Fake.en_gb.address(),
1193 TabletParam.GP: Fake.en_gb.name(),
1194 TabletParam.OTHER: Fake.en_us.paragraph(),
1195 TabletParam.EMAIL: Fake.en_gb.email(),
1196 TabletParam.FINALIZING: True,
1197 }
1198 ]
1199 )
1201 mock_valid = mock.Mock(return_value=(True, ""))
1203 with mock.patch.multiple(
1204 "camcops_server.cc_modules.client_api",
1205 is_candidate_patient_valid_for_group=mock_valid,
1206 is_candidate_patient_valid_for_restricted_user=mock_valid,
1207 ):
1208 reply_dict = self.call_api()
1210 self.assertEqual(
1211 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1212 )
1214 def test_succeeds_for_empty_dob(self) -> None:
1215 # All values set for maximum test coverage
1216 self.post_dict[TabletParam.PATIENT_INFO] = json.dumps(
1217 [
1218 {
1219 TabletParam.DOB: "",
1220 TabletParam.FINALIZING: True,
1221 }
1222 ]
1223 )
1225 mock_valid = mock.Mock(return_value=(True, ""))
1227 with mock.patch.multiple(
1228 "camcops_server.cc_modules.client_api",
1229 is_candidate_patient_valid_for_group=mock_valid,
1230 is_candidate_patient_valid_for_restricted_user=mock_valid,
1231 ):
1232 reply_dict = self.call_api()
1234 self.assertEqual(
1235 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1236 )
1239class OpWhichKeysToSendTests(ClientApiTestCase):
1240 def setUp(self) -> None:
1241 super().setUp()
1243 self.post_dict[TabletParam.OPERATION] = Operations.WHICH_KEYS_TO_SEND
1244 self.post_dict[TabletParam.PKNAME] = "id"
1246 def test_non_existent_table_rejected(self) -> None:
1247 self.post_dict[TabletParam.TABLE] = "nonexistent_table"
1248 reply_dict = self.call_api()
1249 self.assertEqual(
1250 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1251 )
1252 self.assertEqual(
1253 reply_dict[TabletParam.ERROR],
1254 "Invalid client table name: nonexistent_table",
1255 msg=reply_dict,
1256 )
1258 def test_table_rejected_if_client_too_old(self) -> None:
1259 self.post_dict[TabletParam.CAMCOPS_VERSION] = "2.0.0"
1260 self.post_dict[TabletParam.TABLE] = "table_1"
1261 mock_tables = mock.Mock(
1262 return_value={
1263 "table_1": Version("2.0.1"),
1264 }
1265 )
1266 mock_client_table_map = {"table_1": mock.Mock()}
1267 with mock.patch.multiple(
1268 "camcops_server.cc_modules.client_api",
1269 all_tables_with_min_client_version=mock_tables,
1270 CLIENT_TABLE_MAP=mock_client_table_map,
1271 ):
1272 reply_dict = self.call_api()
1274 self.assertEqual(
1275 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1276 )
1277 self.assertIn(
1278 "Client CamCOPS version 2.0.0 is less than the version (2.0.1)",
1279 reply_dict[TabletParam.ERROR],
1280 msg=reply_dict,
1281 )
1283 def test_fails_for_pk_value_date_count_mismatch(self) -> None:
1284 self.post_dict[TabletParam.TABLE] = "bmi"
1285 self.post_dict[TabletParam.PKVALUES] = "1"
1286 self.post_dict[TabletParam.DATEVALUES] = ""
1288 reply_dict = self.call_api()
1290 self.assertEqual(
1291 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1292 )
1293 self.assertIn("Number of PK values", reply_dict[TabletParam.ERROR])
1294 self.assertIn(
1295 "doesn't match number of dates", reply_dict[TabletParam.ERROR]
1296 )
1298 def test_fails_for_pk_value_move_off_tablet_count_mismatch(self) -> None:
1299 self.post_dict[TabletParam.TABLE] = "bmi"
1300 self.post_dict[TabletParam.PKVALUES] = "1,2"
1301 self.post_dict[TabletParam.DATEVALUES] = "2025-01-23,2025-01-24"
1302 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1304 reply_dict = self.call_api()
1306 self.assertEqual(
1307 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1308 )
1309 self.assertIn(
1310 "Number of move-off-tablet values", reply_dict[TabletParam.ERROR]
1311 )
1312 self.assertIn(
1313 "doesn't match number of PKs", reply_dict[TabletParam.ERROR]
1314 )
1316 def test_fails_for_non_integer_client_pk(self) -> None:
1317 self.post_dict[TabletParam.TABLE] = "bmi"
1318 self.post_dict[TabletParam.PKVALUES] = "1,strawberry"
1319 self.post_dict[TabletParam.DATEVALUES] = "2025-01-23,2025-01-24"
1320 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1,1"
1322 reply_dict = self.call_api()
1324 self.assertEqual(
1325 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1326 )
1327 self.assertIn(
1328 "Bad (non-integer) client PK", reply_dict[TabletParam.ERROR]
1329 )
1331 def test_fails_for_missing_date_time(self) -> None:
1332 self.post_dict[TabletParam.TABLE] = "bmi"
1333 self.post_dict[TabletParam.PKVALUES] = "1"
1334 self.post_dict[TabletParam.DATEVALUES] = "null"
1335 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1337 reply_dict = self.call_api()
1339 self.assertEqual(
1340 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1341 )
1342 self.assertIn("Missing date/time", reply_dict[TabletParam.ERROR])
1344 def test_fails_for_bad_date_time(self) -> None:
1345 self.post_dict[TabletParam.TABLE] = "bmi"
1346 self.post_dict[TabletParam.PKVALUES] = "1"
1347 self.post_dict[TabletParam.DATEVALUES] = "Tuesday"
1348 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1350 reply_dict = self.call_api()
1352 self.assertEqual(
1353 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1354 )
1355 self.assertIn("Bad date/time", reply_dict[TabletParam.ERROR])
1357 def test_succeeds_for_valid_values(self) -> None:
1358 self.post_dict[TabletParam.TABLE] = "bmi"
1359 self.post_dict[TabletParam.PKVALUES] = "123"
1360 self.post_dict[TabletParam.DATEVALUES] = "2025-01-23"
1361 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1363 reply_dict = self.call_api()
1365 self.assertEqual(
1366 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1367 )
1368 self.assertEqual(reply_dict[TabletParam.RESULT], "123", msg=reply_dict)
1370 def test_succeeds_for_existing_record(self) -> None:
1371 self.post_dict[TabletParam.TABLE] = "bmi"
1372 patient = PatientFactory(_device=self.device)
1373 bmi = BmiFactory(id=123, patient=patient, _era=ERA_NOW)
1375 self.post_dict[TabletParam.PKVALUES] = f"{bmi.id}"
1376 self.post_dict[TabletParam.DATEVALUES] = "2025-01-23"
1377 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1379 reply_dict = self.call_api()
1381 self.assertEqual(
1382 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1383 )
1384 self.assertEqual(reply_dict[TabletParam.RESULT], "123", msg=reply_dict)
1386 def test_succeeds_for_unmodified_record_marked_for_preservation(
1387 self,
1388 ) -> None:
1389 time_now = local(2025, 1, 26)
1391 patient = PatientFactory(_device=self.device)
1392 bmi = BmiFactory(
1393 id=123,
1394 patient=patient,
1395 _era=ERA_NOW,
1396 when_last_modified=time_now,
1397 _move_off_tablet=False,
1398 )
1400 self.post_dict[TabletParam.TABLE] = "bmi"
1401 self.post_dict[TabletParam.PKVALUES] = f"{bmi.id}"
1402 self.post_dict[TabletParam.DATEVALUES] = time_now.isoformat()
1403 self.post_dict[TabletParam.MOVE_OFF_TABLET_VALUES] = "1"
1405 reply_dict = self.call_api()
1407 self.assertEqual(
1408 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1409 )
1410 self.assertEqual(reply_dict[TabletParam.RESULT], "", msg=reply_dict)
1411 self.dbsession.commit()
1413 self.assertTrue(bmi._move_off_tablet)
1416class OpDeleteWhereKeyNotTests(ClientApiTestCase):
1417 def setUp(self) -> None:
1418 super().setUp()
1420 self.post_dict[TabletParam.OPERATION] = Operations.DELETE_WHERE_KEY_NOT
1421 self.post_dict[TabletParam.TABLE] = "bmi"
1422 self.post_dict[TabletParam.PKNAME] = "id"
1424 def test_records_not_specified_marked_for_removal(self) -> None:
1425 patient = PatientFactory(_device=self.device)
1426 bmis = BmiFactory.create_batch(
1427 3,
1428 patient=patient,
1429 _removal_pending=False,
1430 _era=ERA_NOW,
1431 )
1433 self.post_dict[TabletParam.PKVALUES] = f"{bmis[0].id},{bmis[1].id}"
1434 reply_dict = self.call_api()
1436 self.assertEqual(
1437 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1438 )
1439 self.assertEqual(
1440 reply_dict[TabletParam.RESULT], "Trimmed", msg=reply_dict
1441 )
1443 self.dbsession.commit()
1445 self.assertFalse(bmis[0]._removal_pending)
1446 self.assertFalse(bmis[1]._removal_pending)
1447 self.assertTrue(bmis[2]._removal_pending)
1450class OpStartPreservationTests(ClientApiTestCase):
1451 def setUp(self) -> None:
1452 super().setUp()
1454 self.post_dict[TabletParam.OPERATION] = Operations.START_PRESERVATION
1455 self.post_dict[TabletParam.TABLE] = "bmi"
1456 self.post_dict[TabletParam.PKNAME] = "id"
1458 def test_device_currently_preserving(self) -> None:
1459 self.assertFalse(self.device.currently_preserving)
1461 patient = PatientFactory(_device=self.device)
1462 bmi = BmiFactory(
1463 id=123,
1464 patient=patient,
1465 _era=ERA_NOW,
1466 )
1467 self.post_dict[TabletParam.PKVALUES] = f"{bmi.id}"
1468 reply_dict = self.call_api()
1470 self.assertEqual(
1471 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1472 )
1473 self.assertEqual(
1474 reply_dict[TabletParam.RESULT], "STARTPRESERVATION", msg=reply_dict
1475 )
1477 self.dbsession.commit()
1478 self.assertTrue(self.device.currently_preserving)
1480 def test_marks_table_dirty(self) -> None:
1481 self.assertIsNone(
1482 self.dbsession.execute(
1483 select(DirtyTable).where(DirtyTable.tablename == "bmi")
1484 ).scalar_one_or_none()
1485 )
1487 patient = PatientFactory(_device=self.device)
1488 bmi = BmiFactory(
1489 id=123,
1490 patient=patient,
1491 _era=ERA_NOW,
1492 )
1493 self.post_dict[TabletParam.PKVALUES] = f"{bmi.id}"
1494 reply_dict = self.call_api()
1496 self.assertEqual(
1497 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1498 )
1499 self.assertEqual(
1500 reply_dict[TabletParam.RESULT], "STARTPRESERVATION", msg=reply_dict
1501 )
1503 self.dbsession.commit()
1505 self.assertIsNotNone(
1506 self.dbsession.execute(
1507 select(DirtyTable).where(DirtyTable.tablename == "bmi")
1508 ).scalar_one_or_none()
1509 )
1512class OpUploadEmptyTablesTests(ClientApiTestCase):
1513 def setUp(self) -> None:
1514 super().setUp()
1516 self.post_dict[TabletParam.OPERATION] = Operations.UPLOAD_EMPTY_TABLES
1517 self.post_dict[TabletParam.TABLES] = "bmi,phq9"
1519 def test_all_records_flagged_as_deleted(self) -> None:
1520 patient = PatientFactory(_device=self.device)
1521 bmi = BmiFactory(
1522 patient=patient,
1523 _era=ERA_NOW,
1524 _current=True,
1525 _removal_pending=False,
1526 )
1527 phq9 = Phq9Factory(
1528 patient=patient,
1529 _era=ERA_NOW,
1530 _current=True,
1531 _removal_pending=False,
1532 )
1533 reply_dict = self.call_api()
1535 self.assertEqual(
1536 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1537 )
1538 self.assertEqual(
1539 reply_dict[TabletParam.RESULT],
1540 "UPLOAD-EMPTY-TABLES",
1541 msg=reply_dict,
1542 )
1543 self.dbsession.commit()
1545 self.assertTrue(bmi._removal_pending)
1546 self.assertTrue(phq9._removal_pending)
1548 def test_tables_marked_dirty_if_records_in_current_era(self) -> None:
1549 self.assertIsNone(
1550 self.dbsession.execute(
1551 select(DirtyTable).where(DirtyTable.tablename == "bmi")
1552 ).scalar_one_or_none()
1553 )
1554 self.assertIsNone(
1555 self.dbsession.execute(
1556 select(DirtyTable).where(DirtyTable.tablename == "phq9")
1557 ).scalar_one_or_none()
1558 )
1560 patient = PatientFactory(_device=self.device)
1561 BmiFactory(
1562 patient=patient,
1563 _era=ERA_NOW,
1564 _current=True,
1565 )
1566 Phq9Factory(
1567 patient=patient,
1568 _era=ERA_NOW,
1569 _current=True,
1570 )
1571 reply_dict = self.call_api()
1573 self.assertEqual(
1574 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1575 )
1576 self.dbsession.commit()
1577 self.assertIsNotNone(
1578 self.dbsession.execute(
1579 select(DirtyTable).where(DirtyTable.tablename == "bmi")
1580 ).scalar_one_or_none()
1581 )
1582 self.assertIsNotNone(
1583 self.dbsession.execute(
1584 select(DirtyTable).where(DirtyTable.tablename == "phq9")
1585 ).scalar_one_or_none()
1586 )
1588 def test_tables_marked_clean_if_no_records_in_current_era(self) -> None:
1589 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
1590 DirtyTableFactory(tablename="phq9", device_id=self.device.id)
1591 self.device.currently_preserving = True
1592 self.device.ongoing_upload_batch_utc = now("UTC")
1593 self.dbsession.add(self.device)
1594 self.dbsession.commit()
1596 reply_dict = self.call_api()
1598 self.assertEqual(
1599 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1600 )
1601 self.dbsession.commit()
1602 self.assertIsNone(
1603 self.dbsession.execute(
1604 select(DirtyTable).where(DirtyTable.tablename == "bmi")
1605 ).scalar_one_or_none()
1606 )
1607 self.assertIsNone(
1608 self.dbsession.execute(
1609 select(DirtyTable).where(DirtyTable.tablename == "phq9")
1610 ).scalar_one_or_none()
1611 )
1614class OpUploadRecordTests(ClientApiTestCase):
1615 def setUp(self) -> None:
1616 super().setUp()
1618 self.post_dict[TabletParam.OPERATION] = Operations.UPLOAD_RECORD
1619 self.post_dict[TabletParam.PKNAME] = "id"
1621 def test_upload_inserts_record(self) -> None:
1622 now_utc_string = now("UTC").isoformat()
1623 patient = PatientFactory(_device=self.device)
1625 self.post_dict[TabletParam.TABLE] = "bmi"
1626 self.post_dict[TabletParam.PKVALUES] = "1"
1627 self.post_dict[TabletParam.FIELDS] = ",".join(
1628 [
1629 "id",
1630 "height_m",
1631 "mass_kg",
1632 "when_created",
1633 "when_last_modified",
1634 "_move_off_tablet",
1635 "patient_id",
1636 ]
1637 )
1638 self.post_dict[TabletParam.VALUES] = ",".join(
1639 [
1640 "1",
1641 "1.83",
1642 "67",
1643 now_utc_string,
1644 now_utc_string,
1645 "1",
1646 str(patient.id),
1647 ]
1648 )
1649 reply_dict = self.call_api()
1651 self.assertEqual(
1652 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1653 )
1654 self.assertEqual(
1655 reply_dict[TabletParam.RESULT],
1656 "UPLOAD-INSERT",
1657 msg=reply_dict,
1658 )
1659 bmi = self.dbsession.execute(select(Bmi)).scalar_one_or_none()
1660 self.assertIsNotNone(bmi)
1662 self.assertAlmostEqual(bmi.height_m, 1.83)
1663 self.assertAlmostEqual(bmi.mass_kg, 67)
1665 def test_upload_updates_record(self) -> None:
1666 now_utc_string = now("UTC").isoformat()
1667 patient = PatientFactory(_device=self.device)
1668 bmi = BmiFactory(
1669 patient=patient,
1670 _era=ERA_NOW,
1671 height_m=1.8,
1672 mass_kg=70,
1673 )
1675 self.post_dict[TabletParam.TABLE] = "bmi"
1676 self.post_dict[TabletParam.PKVALUES] = "1"
1677 self.post_dict[TabletParam.FIELDS] = ",".join(
1678 [
1679 "id",
1680 "height_m",
1681 "mass_kg",
1682 "when_created",
1683 "when_last_modified",
1684 "_move_off_tablet",
1685 "patient_id",
1686 ]
1687 )
1688 self.post_dict[TabletParam.VALUES] = ",".join(
1689 [
1690 str(bmi.id),
1691 "1.83",
1692 "67",
1693 now_utc_string,
1694 now_utc_string,
1695 "1",
1696 str(patient.id),
1697 ]
1698 )
1699 reply_dict = self.call_api()
1701 self.assertEqual(
1702 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1703 )
1704 self.assertEqual(
1705 reply_dict[TabletParam.RESULT],
1706 "UPLOAD-UPDATE",
1707 msg=reply_dict,
1708 )
1709 new_bmi = self.dbsession.execute(
1710 select(Bmi).where(Bmi._predecessor_pk == bmi._pk)
1711 ).scalar_one_or_none()
1712 self.assertIsNotNone(new_bmi)
1714 self.assertAlmostEqual(new_bmi.height_m, 1.83)
1715 self.assertAlmostEqual(new_bmi.mass_kg, 67)
1717 def test_fails_if_field_is_reserved(self) -> None:
1718 self.post_dict[TabletParam.TABLE] = "bmi"
1719 self.post_dict[TabletParam.PKVALUES] = "1"
1720 self.post_dict[TabletParam.FIELDS] = ",".join(
1721 [
1722 "id",
1723 "_current",
1724 ]
1725 )
1726 self.post_dict[TabletParam.VALUES] = ",".join(
1727 [
1728 "1",
1729 "0",
1730 ]
1731 )
1732 reply_dict = self.call_api()
1734 self.assertEqual(
1735 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1736 )
1737 self.assertIn(
1738 "Reserved field name for table",
1739 reply_dict[TabletParam.ERROR],
1740 msg=reply_dict,
1741 )
1743 def test_fails_if_field_does_not_exist(self) -> None:
1744 self.post_dict[TabletParam.TABLE] = "bmi"
1745 self.post_dict[TabletParam.PKVALUES] = "1"
1746 self.post_dict[TabletParam.FIELDS] = ",".join(
1747 [
1748 "id",
1749 "nonsense",
1750 ]
1751 )
1752 self.post_dict[TabletParam.VALUES] = ",".join(
1753 [
1754 "1",
1755 "0",
1756 ]
1757 )
1758 reply_dict = self.call_api()
1760 self.assertEqual(
1761 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1762 )
1763 self.assertIn(
1764 "Invalid field name for table",
1765 reply_dict[TabletParam.ERROR],
1766 msg=reply_dict,
1767 )
1769 def test_upload_inserts_patient_idnum_record(self) -> None:
1770 now_utc_string = now("UTC").isoformat()
1771 iddef = NHSIdNumDefinitionFactory()
1772 patient = PatientFactory(_device=self.device)
1774 self.post_dict[TabletParam.TABLE] = "patient_idnum"
1775 self.post_dict[TabletParam.PKVALUES] = "1"
1776 self.post_dict[TabletParam.FIELDS] = ",".join(
1777 [
1778 "id",
1779 "which_idnum",
1780 "idnum_value",
1781 "when_last_modified",
1782 "_move_off_tablet",
1783 "patient_id",
1784 ]
1785 )
1786 nhs_number = Fake.en_gb.nhs_number()
1787 self.post_dict[TabletParam.VALUES] = ",".join(
1788 [
1789 "1",
1790 str(iddef.which_idnum),
1791 str(nhs_number),
1792 now_utc_string,
1793 "1",
1794 str(patient.id),
1795 ]
1796 )
1797 reply_dict = self.call_api()
1799 self.assertEqual(
1800 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1801 )
1802 self.assertEqual(
1803 reply_dict[TabletParam.RESULT],
1804 "UPLOAD-INSERT",
1805 msg=reply_dict,
1806 )
1807 patient_idnum = self.dbsession.execute(
1808 select(PatientIdNum)
1809 ).scalar_one_or_none()
1810 self.assertIsNotNone(patient_idnum)
1812 self.assertEqual(patient_idnum.which_idnum, iddef.which_idnum)
1813 self.assertEqual(patient_idnum.idnum_value, nhs_number)
1815 def test_fails_if_patient_idnum_type_unknown(self) -> None:
1816 now_utc_string = now("UTC").isoformat()
1817 patient = PatientFactory(_device=self.device)
1819 self.post_dict[TabletParam.TABLE] = "patient_idnum"
1820 self.post_dict[TabletParam.PKVALUES] = "1"
1821 self.post_dict[TabletParam.FIELDS] = ",".join(
1822 [
1823 "id",
1824 "which_idnum",
1825 "idnum_value",
1826 "when_last_modified",
1827 "_move_off_tablet",
1828 "patient_id",
1829 ]
1830 )
1831 nhs_number = Fake.en_gb.nhs_number()
1832 self.post_dict[TabletParam.VALUES] = ",".join(
1833 [
1834 "1",
1835 "1",
1836 str(nhs_number),
1837 now_utc_string,
1838 "1",
1839 str(patient.id),
1840 ]
1841 )
1843 reply_dict = self.call_api()
1845 self.assertEqual(
1846 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1847 )
1848 self.assertIn(
1849 "No such ID number type: 1",
1850 reply_dict[TabletParam.ERROR],
1851 msg=reply_dict,
1852 )
1854 def test_fails_if_patient_idnum_invalid(self) -> None:
1855 now_utc_string = now("UTC").isoformat()
1856 iddef = NHSIdNumDefinitionFactory()
1857 patient = PatientFactory(_device=self.device)
1859 self.post_dict[TabletParam.TABLE] = "patient_idnum"
1860 self.post_dict[TabletParam.PKVALUES] = "1"
1861 self.post_dict[TabletParam.FIELDS] = ",".join(
1862 [
1863 "id",
1864 "which_idnum",
1865 "idnum_value",
1866 "when_last_modified",
1867 "_move_off_tablet",
1868 "patient_id",
1869 ]
1870 )
1871 invalid_nhs_number = 123
1872 self.post_dict[TabletParam.VALUES] = ",".join(
1873 [
1874 "1",
1875 str(iddef.which_idnum),
1876 str(invalid_nhs_number),
1877 now_utc_string,
1878 "1",
1879 str(patient.id),
1880 ]
1881 )
1883 reply_dict = self.call_api()
1885 self.assertEqual(
1886 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1887 )
1888 self.assertIn(
1889 f"For ID type {iddef.which_idnum}, ID number "
1890 f"{invalid_nhs_number} is invalid",
1891 reply_dict[TabletParam.ERROR],
1892 msg=reply_dict,
1893 )
1896class OpUploadTableTests(ClientApiTestCase):
1897 def setUp(self) -> None:
1898 super().setUp()
1899 self.post_dict[TabletParam.OPERATION] = Operations.UPLOAD_TABLE
1900 self.post_dict[TabletParam.TABLE] = "bmi"
1901 self.post_dict[TabletParam.PKNAME] = "id"
1902 self.post_dict[TabletParam.FIELDS] = ",".join(
1903 [
1904 "id",
1905 "height_m",
1906 "mass_kg",
1907 "when_created",
1908 "when_last_modified",
1909 "_move_off_tablet",
1910 "patient_id",
1911 ]
1912 )
1914 def test_table_uploaded(self) -> None:
1915 now_utc_string = now("UTC").isoformat()
1916 patient1 = PatientFactory(_device=self.device)
1917 patient2 = PatientFactory(_device=self.device)
1919 self.post_dict[TabletParam.NRECORDS] = "2"
1920 self.post_dict["record0"] = ",".join(
1921 [
1922 "1",
1923 "1.83",
1924 "67",
1925 now_utc_string,
1926 now_utc_string,
1927 "1",
1928 str(patient1.id),
1929 ]
1930 )
1931 self.post_dict["record1"] = ",".join(
1932 [
1933 "2",
1934 "1.6",
1935 "50",
1936 now_utc_string,
1937 now_utc_string,
1938 "1",
1939 str(patient2.id),
1940 ]
1941 )
1942 reply_dict = self.call_api()
1944 self.assertEqual(
1945 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
1946 )
1947 self.assertEqual(
1948 reply_dict[TabletParam.RESULT],
1949 "Table bmi upload successful",
1950 msg=reply_dict,
1951 )
1953 bmi_1 = self.dbsession.execute(
1954 select(Bmi).where(Bmi.id == 1)
1955 ).scalar_one()
1956 self.assertAlmostEqual(bmi_1.height_m, 1.83)
1957 self.assertAlmostEqual(bmi_1.mass_kg, 67)
1959 bmi_2 = self.dbsession.execute(
1960 select(Bmi).where(Bmi.id == 2)
1961 ).scalar_one()
1962 self.assertAlmostEqual(bmi_2.height_m, 1.6)
1963 self.assertAlmostEqual(bmi_2.mass_kg, 50)
1965 def test_fails_if_nrecords_less_than_zero(self) -> None:
1966 self.post_dict[TabletParam.NRECORDS] = "-1"
1967 reply_dict = self.call_api()
1968 self.assertEqual(
1969 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1970 )
1971 self.assertIn(
1972 reply_dict[TabletParam.ERROR],
1973 f"{TabletParam.NRECORDS}=-1: can't be less than 0",
1974 )
1976 def test_fails_if_fields_do_not_match_values(self) -> None:
1977 self.post_dict[TabletParam.NRECORDS] = "1"
1978 self.post_dict["record0"] = ",".join(["1"])
1979 reply_dict = self.call_api()
1980 self.assertEqual(
1981 reply_dict[TabletParam.SUCCESS], FAILURE_CODE, msg=reply_dict
1982 )
1983 self.assertIn(
1984 reply_dict[TabletParam.ERROR],
1985 "Number of fields in field list (7) "
1986 "doesn't match number of values in record 0 (1)",
1987 )
1989 def test_record_updated(self) -> None:
1990 now_utc_string = now("UTC").isoformat()
1991 patient = PatientFactory(_device=self.device)
1992 bmi = BmiFactory(
1993 patient=patient,
1994 _era=ERA_NOW,
1995 height_m=1.8,
1996 mass_kg=70,
1997 )
1999 self.post_dict[TabletParam.NRECORDS] = "1"
2000 self.post_dict["record0"] = ",".join(
2001 [
2002 str(bmi.id),
2003 "1.83",
2004 "67",
2005 now_utc_string,
2006 now_utc_string,
2007 "1",
2008 str(patient.id),
2009 ]
2010 )
2011 reply_dict = self.call_api()
2013 self.assertEqual(
2014 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2015 )
2016 self.assertEqual(
2017 reply_dict[TabletParam.RESULT],
2018 "Table bmi upload successful",
2019 msg=reply_dict,
2020 )
2022 bmi_1 = self.dbsession.execute(
2023 select(Bmi).where(Bmi._predecessor_pk == bmi._pk)
2024 ).scalar_one()
2025 self.assertAlmostEqual(bmi_1.height_m, 1.83)
2026 self.assertAlmostEqual(bmi_1.mass_kg, 67)
2028 def test_record_flagged_for_deletion(self) -> None:
2029 now_utc_string = now("UTC").isoformat()
2030 patient = PatientFactory(_device=self.device)
2031 bmi = BmiFactory(
2032 patient=patient,
2033 _era=ERA_NOW,
2034 height_m=1.8,
2035 mass_kg=70,
2036 )
2038 # No record for this on the tablet
2039 bmi_to_delete = BmiFactory(
2040 patient=patient,
2041 _era=ERA_NOW,
2042 )
2043 self.assertFalse(bmi_to_delete._removal_pending)
2045 self.post_dict[TabletParam.NRECORDS] = "1"
2046 self.post_dict["record0"] = ",".join(
2047 [
2048 str(bmi.id),
2049 "1.83",
2050 "67",
2051 now_utc_string,
2052 now_utc_string,
2053 "1",
2054 str(patient.id),
2055 ]
2056 )
2057 reply_dict = self.call_api()
2059 self.assertEqual(
2060 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2061 )
2062 self.dbsession.commit()
2064 self.assertTrue(bmi_to_delete._removal_pending)
2066 def test_no_records_marks_table_clean(self) -> None:
2067 self.device.currently_preserving = True
2068 self.device.ongoing_upload_batch_utc = now("UTC")
2069 self.dbsession.add(self.device)
2070 self.dbsession.commit()
2072 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2073 self.assertIsNotNone(
2074 self.dbsession.execute(
2075 select(DirtyTable).where(DirtyTable.tablename == "bmi")
2076 ).scalar_one_or_none()
2077 )
2078 self.post_dict[TabletParam.NRECORDS] = "0"
2079 reply_dict = self.call_api()
2081 self.assertEqual(
2082 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2083 )
2085 self.assertIsNone(
2086 self.dbsession.execute(
2087 select(DirtyTable).where(DirtyTable.tablename == "bmi")
2088 ).scalar_one_or_none()
2089 )
2092class OpEndUploadTests(ClientApiTestCase):
2093 def setUp(self) -> None:
2094 super().setUp()
2095 self.post_dict[TabletParam.OPERATION] = Operations.END_UPLOAD
2096 self.device.ongoing_upload_batch_utc = now("UTC")
2097 self.device.uploading_user_id = self.user.id
2098 self.dbsession.add(self.device)
2099 self.dbsession.commit()
2101 def test_updates_added_records(self) -> None:
2102 patient = PatientFactory(_device=self.device)
2103 bmi = BmiFactory(
2104 patient=patient,
2105 _era=ERA_NOW,
2106 _addition_pending=True,
2107 )
2109 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2111 reply_dict = self.call_api()
2113 self.assertEqual(
2114 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2115 )
2116 self.assertFalse(bmi._addition_pending)
2117 self.assertIsNotNone(bmi._when_added_exact)
2118 self.assertIsNotNone(bmi._when_added_batch_utc)
2120 def test_updates_removed_records(self) -> None:
2121 patient = PatientFactory(_device=self.device)
2122 bmi = BmiFactory(
2123 patient=patient,
2124 _era=ERA_NOW,
2125 _current=True,
2126 _removal_pending=True,
2127 )
2129 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2131 reply_dict = self.call_api()
2133 self.assertEqual(
2134 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2135 )
2136 self.assertFalse(bmi._current)
2137 self.assertFalse(bmi._removal_pending)
2138 self.assertEqual(bmi._removing_user_id, self.user.id)
2139 self.assertIsNotNone(bmi._when_removed_exact)
2140 self.assertIsNotNone(bmi._when_removed_batch_utc)
2142 def test_updates_preserved_records(self) -> None:
2143 patient = PatientFactory(_device=self.device)
2144 bmi = BmiFactory(
2145 patient=patient,
2146 _era=ERA_NOW,
2147 _move_off_tablet=True,
2148 )
2150 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2152 reply_dict = self.call_api()
2154 self.assertEqual(
2155 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2156 )
2157 self.assertFalse(bmi._move_off_tablet)
2158 self.assertEqual(bmi._preserving_user_id, self.user.id)
2159 self.assertNotEqual(bmi._era, ERA_NOW)
2162class OpStartUploadTests(ClientApiTestCase):
2163 def setUp(self) -> None:
2164 super().setUp()
2165 self.post_dict[TabletParam.OPERATION] = Operations.START_UPLOAD
2167 def test_updates_device_batch_utc_details(self) -> None:
2168 reply_dict = self.call_api()
2170 self.assertEqual(
2171 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2172 )
2173 self.dbsession.commit()
2174 self.assertIsNotNone(self.device.last_upload_batch_utc)
2175 self.assertIsNotNone(self.device.ongoing_upload_batch_utc)
2176 self.uploading_user_id = self.user.id
2178 def test_deletes_records_with_addition_pending(self) -> None:
2179 patient = PatientFactory(_device=self.device)
2180 BmiFactory(
2181 patient=patient,
2182 _era=ERA_NOW,
2183 _addition_pending=True,
2184 )
2185 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2187 reply_dict = self.call_api()
2189 self.assertEqual(
2190 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2191 )
2192 self.dbsession.commit()
2194 self.assertIsNone(
2195 self.dbsession.execute(select(Bmi)).scalar_one_or_none()
2196 )
2198 def test_updates_records_with_removal_pending(self) -> None:
2199 patient = PatientFactory(_device=self.device)
2200 BmiFactory(
2201 patient=patient,
2202 _era=ERA_NOW,
2203 _removal_pending=True,
2204 _when_added_exact=self.req.now,
2205 _when_removed_batch_utc=self.req.now_utc,
2206 _removing_user_id=self.user.id,
2207 _successor_pk=1234,
2208 )
2209 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2211 reply_dict = self.call_api()
2213 self.assertEqual(
2214 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2215 )
2216 self.dbsession.commit()
2218 bmi = self.dbsession.execute(select(Bmi)).scalar_one_or_none()
2219 self.assertFalse(bmi._removal_pending)
2220 self.assertIsNone(bmi._when_added_exact)
2221 self.assertIsNone(bmi._when_removed_batch_utc)
2222 self.assertIsNone(bmi._removing_user_id)
2223 self.assertIsNone(bmi._successor_pk)
2225 def test_sets_move_off_tablet_field_to_false(self) -> None:
2226 patient = PatientFactory(_device=self.device)
2227 BmiFactory(
2228 patient=patient,
2229 _era=ERA_NOW,
2230 _move_off_tablet=True,
2231 )
2232 DirtyTableFactory(tablename="bmi", device_id=self.device.id)
2234 reply_dict = self.call_api()
2236 self.assertEqual(
2237 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2238 )
2239 self.dbsession.commit()
2241 bmi = self.dbsession.execute(select(Bmi)).scalar_one_or_none()
2242 self.assertFalse(bmi._move_off_tablet)
2245class OpGetIdInfoTests(ClientApiTestCase):
2246 def setUp(self) -> None:
2247 super().setUp()
2248 self.post_dict[TabletParam.OPERATION] = Operations.GET_ID_INFO
2250 def test_returns_database_title(self) -> None:
2251 with mock.patch.object(
2252 self.req,
2253 "database_title",
2254 "test database",
2255 ):
2256 reply_dict = self.call_api()
2258 self.assertEqual(
2259 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2260 )
2261 self.assertEqual(
2262 reply_dict[TabletParam.DATABASE_TITLE], "test database"
2263 )
2265 def test_returns_upload_policy(self) -> None:
2266 self.group.upload_policy = "sex and anyidnum"
2267 self.dbsession.add(self.group)
2268 self.dbsession.commit()
2270 with mock.patch.object(
2271 self.req,
2272 "database_title",
2273 "test database",
2274 ):
2275 reply_dict = self.call_api()
2277 self.assertEqual(
2278 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2279 )
2280 self.assertEqual(
2281 reply_dict[TabletParam.ID_POLICY_UPLOAD], "sex and anyidnum"
2282 )
2284 def test_returns_finalize_policy(self) -> None:
2285 self.group.finalize_policy = "sex and anyidnum"
2286 self.dbsession.add(self.group)
2287 self.dbsession.commit()
2289 with mock.patch.object(
2290 self.req,
2291 "database_title",
2292 "test database",
2293 ):
2294 reply_dict = self.call_api()
2296 self.assertEqual(
2297 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2298 )
2299 self.assertEqual(
2300 reply_dict[TabletParam.ID_POLICY_FINALIZE], "sex and anyidnum"
2301 )
2303 def test_returns_server_version_string(self) -> None:
2304 with mock.patch.multiple(
2305 "camcops_server.cc_modules.client_api",
2306 CAMCOPS_SERVER_VERSION_STRING="test version",
2307 ):
2308 with mock.patch.object(
2309 self.req,
2310 "database_title",
2311 "test database",
2312 ):
2313 reply_dict = self.call_api()
2315 self.assertEqual(
2316 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2317 )
2318 self.assertEqual(
2319 reply_dict[TabletParam.SERVER_CAMCOPS_VERSION], "test version"
2320 )
2322 def test_returns_idnum_definition(self) -> None:
2323 mock_idnum_definition = mock.Mock(
2324 which_idnum=1,
2325 description="Mock NHS Number",
2326 short_description="Mock NHS#",
2327 validation_method="Mock validation method",
2328 )
2330 with mock.patch.multiple(
2331 self.req,
2332 database_title="test database",
2333 idnum_definitions=[mock_idnum_definition],
2334 ):
2335 reply_dict = self.call_api()
2337 self.assertEqual(
2338 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2339 )
2340 self.assertEqual(
2341 reply_dict["idDescription1"],
2342 "Mock NHS Number",
2343 )
2344 self.assertEqual(
2345 reply_dict["idShortDescription1"],
2346 "Mock NHS#",
2347 )
2348 self.assertEqual(
2349 reply_dict["idValidationMethod1"],
2350 "Mock validation method",
2351 )
2354class OpCheckUploadUserAndDeviceTests(ClientApiTestCase):
2355 def setUp(self) -> None:
2356 super().setUp()
2357 self.post_dict[TabletParam.OPERATION] = (
2358 Operations.CHECK_UPLOAD_USER_DEVICE
2359 )
2361 def test_succeeds(self) -> None:
2362 reply_dict = self.call_api()
2364 self.assertEqual(
2365 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2366 )
2369class OpGetAllowedTablesTests(ClientApiTestCase):
2370 def setUp(self) -> None:
2371 super().setUp()
2372 self.post_dict[TabletParam.OPERATION] = Operations.GET_ALLOWED_TABLES
2374 def test_returns_allowed_tables(self) -> None:
2375 mock_task_tables = mock.Mock(
2376 return_value={
2377 "table_1": Version("2.0.0"),
2378 "table_2": Version("2.0.1"),
2379 # These get overridden
2380 "blobs": Version("1.0.0"),
2381 "patient": Version("1.0.0"),
2382 "patient_idnum": Version("1.0.0"),
2383 }
2384 )
2386 with mock.patch.multiple(
2387 "camcops_server.cc_modules.client_api",
2388 all_task_tables_with_min_client_version=mock_task_tables,
2389 MINIMUM_TABLET_VERSION=Version("2.0.2"),
2390 ):
2391 reply_dict = self.call_api()
2393 self.assertEqual(
2394 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2395 )
2397 fields = ",".join(
2398 [
2399 AllowedTablesFieldNames.TABLENAME,
2400 AllowedTablesFieldNames.MIN_CLIENT_VERSION,
2401 ]
2402 )
2403 self.assertEqual(reply_dict[TabletParam.NFIELDS], "2")
2404 self.assertEqual(reply_dict[TabletParam.FIELDS], fields)
2405 self.assertEqual(reply_dict[TabletParam.NRECORDS], "5")
2406 self.assertEqual(reply_dict["record0"], "'table_1','2.0.0'")
2407 self.assertEqual(reply_dict["record1"], "'table_2','2.0.1'")
2408 self.assertEqual(reply_dict["record2"], "'blobs','2.0.2'")
2409 self.assertEqual(reply_dict["record3"], "'patient','2.0.2'")
2410 self.assertEqual(reply_dict["record4"], "'patient_idnum','2.0.2'")
2413class OpGetExtraStringsTests(ClientApiTestCase):
2414 def setUp(self) -> None:
2415 super().setUp()
2416 self.post_dict[TabletParam.OPERATION] = Operations.GET_EXTRA_STRINGS
2418 def test_returns_extra_strings(self) -> None:
2419 mock_extra_strings = mock.Mock(
2420 return_value=[
2421 ("task1", "name1", "language1", "value1"),
2422 ("task2", "name2", "language2", "value2"),
2423 ]
2424 )
2426 with mock.patch.object(
2427 self.req,
2428 "get_all_extra_strings",
2429 mock_extra_strings,
2430 ):
2431 reply_dict = self.call_api()
2433 self.assertEqual(
2434 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2435 )
2437 fields = ",".join(
2438 [
2439 ExtraStringFieldNames.TASK,
2440 ExtraStringFieldNames.NAME,
2441 ExtraStringFieldNames.LANGUAGE,
2442 ExtraStringFieldNames.VALUE,
2443 ]
2444 )
2445 self.assertEqual(reply_dict[TabletParam.NFIELDS], "4")
2446 self.assertEqual(reply_dict[TabletParam.FIELDS], fields)
2447 self.assertEqual(reply_dict[TabletParam.NRECORDS], "2")
2448 self.assertEqual(
2449 reply_dict["record0"], "'task1','name1','language1','value1'"
2450 )
2451 self.assertEqual(
2452 reply_dict["record1"], "'task2','name2','language2','value2'"
2453 )
2456class OpRegisterDeviceTests(ClientApiTestCase):
2457 def setUp(self) -> None:
2458 super().setUp()
2459 self.post_dict[TabletParam.OPERATION] = Operations.REGISTER
2461 def test_updates_existing_device(self) -> None:
2462 self.assertIsNone(self.device.when_registered_utc)
2463 self.assertIsNone(self.device.registered_by_user_id)
2464 self.post_dict[TabletParam.DEVICE_FRIENDLY_NAME] = "Test device name"
2466 reply_dict = self.call_api()
2468 self.assertEqual(
2469 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2470 )
2472 self.dbsession.commit()
2474 # Server ID info tested elsewhere
2475 self.assertIn(TabletParam.SERVER_CAMCOPS_VERSION, reply_dict)
2477 self.assertEqual(self.device.friendly_name, "Test device name")
2478 self.assertEqual(
2479 self.device.camcops_version,
2480 Version(self.req.tabletsession.tablet_version_str),
2481 )
2482 self.assertIsNotNone(self.device.when_registered_utc)
2483 self.assertEqual(self.device.registered_by_user_id, self.user.id)
2485 def test_registers_new_device(self) -> None:
2486 self.assertIsNone(self.device.when_registered_utc)
2487 self.assertIsNone(self.device.registered_by_user_id)
2488 self.post_dict[TabletParam.DEVICE] = "unregistered"
2489 self.post_dict[TabletParam.DEVICE_FRIENDLY_NAME] = "Test device name"
2491 reply_dict = self.call_api()
2493 self.assertEqual(
2494 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2495 )
2497 self.dbsession.commit()
2499 device = self.dbsession.execute(
2500 select(Device).where(Device.name == "unregistered")
2501 ).scalar_one()
2503 self.assertEqual(device.friendly_name, "Test device name")
2504 self.assertEqual(
2505 device.camcops_version,
2506 Version(self.req.tabletsession.tablet_version_str),
2507 )
2508 self.assertIsNotNone(device.when_registered_utc)
2509 self.assertEqual(device.registered_by_user_id, self.user.id)
2512class OpCheckDeviceRegisteredTests(ClientApiTestCase):
2513 def setUp(self) -> None:
2514 super().setUp()
2515 self.post_dict[TabletParam.OPERATION] = (
2516 Operations.CHECK_DEVICE_REGISTERED
2517 )
2519 def test_device_registered(self) -> None:
2520 reply_dict = self.call_api()
2522 self.assertEqual(
2523 reply_dict[TabletParam.SUCCESS], SUCCESS_CODE, msg=reply_dict
2524 )