Coverage for cc_modules/tests/cc_task_tests.py: 97%

1""" 

2camcops_server/cc_modules/tests/cc_task_tests.py 

3 

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

5 

6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry. 

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

8 

9 This file is part of CamCOPS. 

10 

11 CamCOPS is free software: you can redistribute it and/or modify 

12 it under the terms of the GNU General Public License as published by 

13 the Free Software Foundation, either version 3 of the License, or 

14 (at your option) any later version. 

15 

16 CamCOPS is distributed in the hope that it will be useful, 

17 but WITHOUT ANY WARRANTY; without even the implied warranty of 

18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19 GNU General Public License for more details. 

20 

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

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

23 

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

25 

26""" 

27 

28import logging 

29import os 

30from pathlib import Path 

31 

32from cardinal_pythonlib.logs import BraceStyleAdapter 

33from pendulum import Date, DateTime as Pendulum 

34 

35from camcops_server.cc_modules.cc_dummy_database import DummyDataInserter 

36from camcops_server.cc_modules.cc_task import Task 

37from camcops_server.cc_modules.cc_testfactories import UserFactory 

38from camcops_server.cc_modules.cc_unittest import DemoDatabaseTestCase 

39from camcops_server.cc_modules.cc_validators import validate_task_tablename 

40 

41log = BraceStyleAdapter(logging.getLogger(__name__)) 

42 

43 

44# ============================================================================= 

45# Unit testing 

46# ============================================================================= 

47 

48 

49class TaskTests(DemoDatabaseTestCase): 

50 

51 def test_query_phq9(self) -> None: 

52 self.announce("test_query_phq9") 

53 from camcops_server.tasks import Phq9 

54 

55 phq9_query = self.dbsession.query(Phq9) 

56 results = phq9_query.all() 

57 log.info("{}", results) 

58 

59 def test_all_tasks(self) -> None: 

60 self.announce("test_all_tasks") 

61 from datetime import date 

62 import hl7 

63 from sqlalchemy.sql.schema import Column 

64 from camcops_server.cc_modules.cc_ctvinfo import CtvInfo 

65 from camcops_server.cc_modules.cc_patient import Patient 

66 from camcops_server.cc_modules.cc_simpleobjects import IdNumReference 

67 from camcops_server.cc_modules.cc_snomed import ( 

68 SnomedExpression, 

69 ) 

70 from camcops_server.cc_modules.cc_string import APPSTRING_TASKNAME 

71 from camcops_server.cc_modules.cc_summaryelement import SummaryElement 

72 from camcops_server.cc_modules.cc_trackerhelpers import ( 

73 TrackerInfo, 

74 ) 

75 from camcops_server.cc_modules.cc_spreadsheet import ( 

76 SpreadsheetPage, 

77 ) 

78 from camcops_server.cc_modules.cc_xml import XmlElement 

79 

80 user = UserFactory() 

81 

82 subclasses = Task.all_subclasses_by_tablename() 

83 tables = [cls.tablename for cls in subclasses] 

84 log.info("Actual task table names: {!r} (n={})", tables, len(tables)) 

85 req = self.req 

86 recipdef = self.recipdef 

87 dummy_data_factory = DummyDataInserter() 

88 task_doc_root = os.path.join( 

89 Path(__file__).resolve().parents[4], "docs", "source", "tasks" 

90 ) 

91 for cls in subclasses: 

92 log.info("Testing {}", cls) 

93 assert cls.extrastring_taskname != APPSTRING_TASKNAME 

94 q = self.dbsession.query(cls) 

95 t = q.first() # type: Task 

96 

97 self.assertIsNotNone(t, "Missing task!") 

98 

99 # Name validity 

100 validate_task_tablename(t.tablename) 

101 

102 # Core functions 

103 self.assertIsInstance(t.is_complete(), bool) 

104 self.assertIsInstance(t.get_task_html(req), str) 

105 for trackerinfo in t.get_trackers(req): 

106 self.assertIsInstance(trackerinfo, TrackerInfo) 

107 ctvlist = t.get_clinical_text(req) 

108 if ctvlist is not None: 

109 for ctvinfo in ctvlist: 

110 self.assertIsInstance(ctvinfo, CtvInfo) 

111 for est in t.get_all_summary_tables(req): 

112 self.assertIsInstance( 

113 est.get_spreadsheet_page(), SpreadsheetPage 

114 ) 

115 self.assertIsInstance(est.get_xml_element(), XmlElement) 

116 

117 self.assertIsInstance(t.has_patient, bool) 

118 self.assertIsInstance(t.is_anonymous, bool) 

119 self.assertIsInstance(t.has_clinician, bool) 

120 self.assertIsInstance(t.has_respondent, bool) 

121 self.assertIsInstance(t.tablename, str) 

122 for fn in t.get_fieldnames(): 

123 self.assertIsInstance(fn, str) 

124 self.assertIsInstance(t.field_contents_valid(), bool) 

125 for msg in t.field_contents_invalid_because(): 

126 self.assertIsInstance(msg, str) 

127 for fn in t.get_blob_fields(): 

128 self.assertIsInstance(fn, str) 

129 

130 self.assertIsInstance(t.pk, int) # all our examples do have PKs 

131 self.assertIsInstance(t.is_preserved(), bool) 

132 self.assertIsInstance(t.was_forcibly_preserved(), bool) 

133 self.assertIsInstanceOrNone(t.get_creation_datetime(), Pendulum) 

134 self.assertIsInstanceOrNone( 

135 t.get_creation_datetime_utc(), Pendulum 

136 ) 

137 self.assertIsInstanceOrNone( 

138 t.get_seconds_from_creation_to_first_finish(), float 

139 ) 

140 

141 self.assertIsInstance(t.get_adding_user_id(), int) 

142 self.assertIsInstance(t.get_adding_user_username(), str) 

143 self.assertIsInstance(t.get_removing_user_username(), str) 

144 self.assertIsInstance(t.get_preserving_user_username(), str) 

145 self.assertIsInstance(t.get_manually_erasing_user_username(), str) 

146 

147 # Summaries 

148 for se in t.standard_task_summary_fields(): 

149 self.assertIsInstance(se, SummaryElement) 

150 

151 # SNOMED-CT 

152 if req.snomed_supported: 

153 for snomed_code in t.get_snomed_codes(req): 

154 self.assertIsInstance(snomed_code, SnomedExpression) 

155 

156 # Clinician 

157 self.assertIsInstance(t.get_clinician_name(), str) 

158 

159 # Respondent 

160 self.assertIsInstance(t.is_respondent_complete(), bool) 

161 

162 # Patient 

163 self.assertIsInstanceOrNone(t.patient, Patient) 

164 self.assertIsInstance(t.is_female(), bool) 

165 self.assertIsInstance(t.is_male(), bool) 

166 self.assertIsInstanceOrNone(t.get_patient_server_pk(), int) 

167 self.assertIsInstance(t.get_patient_forename(), str) 

168 self.assertIsInstance(t.get_patient_surname(), str) 

169 dob = t.get_patient_dob() 

170 assert ( 

171 dob is None or isinstance(dob, date) or isinstance(dob, Date) 

172 ) 

173 self.assertIsInstanceOrNone(t.get_patient_dob_first11chars(), str) 

174 self.assertIsInstance(t.get_patient_sex(), str) 

175 self.assertIsInstance(t.get_patient_address(), str) 

176 for idnum in t.get_patient_idnum_objects(): 

177 self.assertIsInstance( 

178 idnum.get_idnum_reference(), IdNumReference 

179 ) 

180 self.assertIsInstance(idnum.is_superficially_valid(), bool) 

181 self.assertIsInstance(idnum.description(req), str) 

182 self.assertIsInstance(idnum.short_description(req), str) 

183 self.assertIsInstance(idnum.get_filename_component(req), str) 

184 

185 # HL7 v2 

186 pidseg = t.get_patient_hl7_pid_segment(req, recipdef) 

187 assert isinstance(pidseg, str) or isinstance(pidseg, hl7.Segment) 

188 for dataseg in t.get_hl7_data_segments(req, recipdef): 

189 self.assertIsInstance(dataseg, hl7.Segment) 

190 for dataseg in t.get_hl7_extra_data_segments(recipdef): 

191 self.assertIsInstance(dataseg, hl7.Segment) 

192 

193 # FHIR 

194 self.assertIsInstance( 

195 t.get_fhir_bundle(req, recipdef).as_json(), dict 

196 ) # the main test is not crashing! 

197 

198 # Other properties 

199 self.assertIsInstance(t.is_erased(), bool) 

200 self.assertIsInstance(t.is_live_on_tablet(), bool) 

201 for attrname, col in t.gen_text_filter_columns(): 

202 self.assertIsInstance(attrname, str) 

203 self.assertIsInstance(col, Column) 

204 

205 # Views 

206 for page in t.get_spreadsheet_pages(req): 

207 self.assertIsInstance(page.get_tsv(), str) 

208 self.assertIsInstance(t.get_xml(req), str) 

209 self.assertIsInstance(t.get_html(req), str) 

210 self.assertIsInstance(t.get_pdf(req), bytes) 

211 self.assertIsInstance(t.get_pdf_html(req), str) 

212 self.assertIsInstance(t.suggested_pdf_filename(req), str) 

213 self.assertIsInstance( 

214 t.get_rio_metadata( 

215 req, 

216 which_idnum=1, 

217 uploading_user_id=user.id, 

218 document_type="some_doc_type", 

219 ), 

220 str, 

221 ) 

222 

223 # Help 

224 help_file = f"{t.help_url_basename()}.rst" 

225 task_help_file = os.path.join(task_doc_root, help_file) 

226 self.assertTrue( 

227 os.path.exists(task_help_file), 

228 msg=f"Task help not found at {task_help_file}", 

229 ) 

230 

231 # Special operations 

232 t.apply_special_note( 

233 req, "Debug: Special note! (1)", from_console=True 

234 ) 

235 t.apply_special_note( 

236 req, "Debug: Special note! (2)", from_console=False 

237 ) 

238 self.assertIsInstance(t.special_notes, list) 

239 t.cancel_from_export_log(req, from_console=True) 

240 t.cancel_from_export_log(req, from_console=False) 

241 

242 # Insert random data and check it doesn't crash. 

243 dummy_data_factory.fill_in_task_fields(t) 

244 self.assertIsInstance(t.get_html(req), str) 

245 

246 # Destructive special operations 

247 self.assertFalse(t.is_erased()) 

248 t.manually_erase(req) 

249 self.assertTrue(t.is_erased()) 

250 t.delete_entirely(req)