Coverage for cc_modules/cc_testfactories.py: 91%

1""" 

2camcops_server/cc_modules/cc_testfactories.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**Factory Boy SQL Alchemy test factories.** 

27 

28""" 

29 

30from typing import Any, cast, Optional, TYPE_CHECKING 

31 

32from cardinal_pythonlib.datetimefunc import ( 

33 convert_datetime_to_utc, 

34 format_datetime, 

35) 

36import factory 

37from faker import Faker 

38import pendulum 

39 

40from camcops_server.cc_modules.cc_blob import Blob 

41from camcops_server.cc_modules.cc_constants import DateFormat, ERA_NOW 

42from camcops_server.cc_modules.cc_device import Device 

43from camcops_server.cc_modules.cc_dirtytables import DirtyTable 

44from camcops_server.cc_modules.cc_email import Email 

45from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient 

46from camcops_server.cc_modules.cc_group import Group 

47from camcops_server.cc_modules.cc_idnumdef import IdNumDefinition 

48from camcops_server.cc_modules.cc_ipuse import IpUse 

49from camcops_server.cc_modules.cc_membership import UserGroupMembership 

50from camcops_server.cc_modules.cc_patient import Patient 

51from camcops_server.cc_modules.cc_patientidnum import PatientIdNum 

52from camcops_server.cc_modules.cc_specialnote import SpecialNote 

53from camcops_server.cc_modules.cc_testproviders import register_all_providers 

54from camcops_server.cc_modules.cc_taskschedule import ( 

55 PatientTaskSchedule, 

56 PatientTaskScheduleEmail, 

57 TaskSchedule, 

58 TaskScheduleItem, 

59) 

60from camcops_server.cc_modules.cc_user import User 

61 

62if TYPE_CHECKING: 

63 from factory.builder import Resolver 

64 from camcops_server.cc_modules.cc_request import CamcopsRequest 

65 

66 

67# Avoid any ID clashes with objects not created with factories 

68ID_OFFSET = 1000 

69RIO_ID_OFFSET = 10000 

70STUDY_ID_OFFSET = 5000 

71 

72 

73class Fake: 

74 # Factory Boy has its own interface to Faker (factory.Faker()). This 

75 # takes a function to be called at object generation time and as far as I 

76 # can tell this doesn't support being able to create fake data based on 

77 # other fake attributes such as notes for a patient. You can work 

78 # around this by adding a lot of logic to the factories. To me it makes 

79 # sense to keep the factories simple and do as much as possible of the 

80 # content generation in the providers. So we call Faker directly instead. 

81 en_gb = Faker("en_GB") # For UK postcodes, phone numbers etc 

82 en_us = Faker("en_US") # en_GB gives Lorem ipsum for pad words. 

83 

84 

85register_all_providers(Fake.en_gb) 

86 

87 

88# sqlalchemy_session gets poked in by DemoRequestCase.setUp() 

89class BaseFactory(factory.alchemy.SQLAlchemyModelFactory): 

90 class Meta: 

91 sqlalchemy_session_persistence = "commit" 

92 

93 

94class DeviceFactory(BaseFactory): 

95 class Meta: 

96 model = Device 

97 

98 id = factory.Sequence(lambda n: n + ID_OFFSET) 

99 name = factory.Sequence(lambda n: f"test-device-{n + ID_OFFSET}") 

100 

101 

102class IpUseFactory(BaseFactory): 

103 class Meta: 

104 model = IpUse 

105 

106 clinical = factory.LazyFunction(Fake.en_gb.pybool) 

107 commercial = factory.LazyFunction(Fake.en_gb.pybool) 

108 educational = factory.LazyFunction(Fake.en_gb.pybool) 

109 research = factory.LazyFunction(Fake.en_gb.pybool) 

110 

111 

112class GroupFactory(BaseFactory): 

113 class Meta: 

114 model = Group 

115 

116 id = factory.Sequence(lambda n: n + ID_OFFSET) 

117 name = factory.Sequence(lambda n: f"Group {n + ID_OFFSET}") 

118 ip_use = factory.SubFactory(IpUseFactory) 

119 

120 

121class AnyIdNumGroupFactory(GroupFactory): 

122 upload_policy = "sex and anyidnum" 

123 finalize_policy = "sex and anyidnum" 

124 

125 

126class UserFactory(BaseFactory): 

127 class Meta: 

128 model = User 

129 

130 username = factory.Sequence(lambda n: f"user{n}") 

131 hashedpw = "" 

132 

133 @factory.post_generation 

134 def password( 

135 obj: User, 

136 create: bool, 

137 password: Optional[str], 

138 request: "CamcopsRequest" = None, 

139 **kwargs: Any, 

140 ) -> None: 

141 if not create: 

142 return 

143 

144 if password is None: 

145 return 

146 

147 assert request is not None 

148 

149 obj.set_password(request, password) 

150 

151 

152class GenericTabletRecordFactory(BaseFactory): 

153 class Meta: 

154 exclude = ("default_iso_datetime",) 

155 abstract = True 

156 

157 default_iso_datetime = "1970-01-01T12:00" 

158 

159 _pk = factory.Sequence(lambda n: n + ID_OFFSET) 

160 _device = factory.SubFactory(DeviceFactory) 

161 _group = factory.SubFactory(AnyIdNumGroupFactory) 

162 _adding_user = factory.SubFactory(UserFactory) 

163 

164 @factory.lazy_attribute 

165 def _when_added_exact(obj: "Resolver") -> pendulum.DateTime: 

166 datetime = cast( 

167 pendulum.DateTime, pendulum.parse(obj.default_iso_datetime) 

168 ) 

169 

170 return datetime 

171 

172 @factory.lazy_attribute 

173 def _when_added_batch_utc(obj: "Resolver") -> pendulum.DateTime: 

174 era_time = pendulum.parse(obj.default_iso_datetime) 

175 return convert_datetime_to_utc(era_time) # type: ignore[arg-type] 

176 

177 @factory.lazy_attribute 

178 def _era(obj: "Resolver") -> str: 

179 era_time = pendulum.parse(obj.default_iso_datetime) 

180 return format_datetime(era_time, DateFormat.ISO8601) # type: ignore[arg-type] # noqa: E501 

181 

182 @factory.lazy_attribute 

183 def _current(obj: "Resolver") -> bool: 

184 # _current = True gets ignored for some reason 

185 return True 

186 

187 @factory.lazy_attribute 

188 def when_last_modified(obj: "Resolver") -> str: 

189 era_time = pendulum.parse(obj.default_iso_datetime) 

190 return format_datetime(era_time, DateFormat.ISO8601) # type: ignore[arg-type] # noqa: E501 

191 

192 

193class PatientFactory(GenericTabletRecordFactory): 

194 class Meta: 

195 model = Patient 

196 

197 id = factory.Sequence(lambda n: n + ID_OFFSET) 

198 sex = factory.LazyFunction(Fake.en_gb.sex) 

199 dob = factory.LazyFunction(Fake.en_gb.consistent_date_of_birth) 

200 address = factory.LazyFunction(Fake.en_gb.address) 

201 gp = factory.LazyFunction(Fake.en_gb.name) 

202 other = factory.LazyFunction(Fake.en_us.paragraph) 

203 email = factory.LazyFunction(Fake.en_gb.email) 

204 

205 @factory.lazy_attribute 

206 def forename(obj: "Resolver") -> str: 

207 return Fake.en_gb.forename(obj.sex) 

208 

209 surname = factory.LazyFunction(Fake.en_gb.last_name) 

210 

211 

212class ServerCreatedPatientFactory(PatientFactory): 

213 @factory.lazy_attribute 

214 def _device(obj: "Resolver") -> Device: 

215 # May have been created in BasicDatabaseTestCase.setUp 

216 return Device.get_server_device( 

217 ServerCreatedPatientFactory._meta.sqlalchemy_session 

218 ) 

219 

220 @factory.lazy_attribute 

221 def _era(obj: "Resolver") -> str: 

222 return ERA_NOW 

223 

224 

225class IdNumDefinitionFactory(BaseFactory): 

226 class Meta: 

227 model = IdNumDefinition 

228 

229 which_idnum = factory.Sequence(lambda n: n + ID_OFFSET) 

230 

231 

232class NHSIdNumDefinitionFactory(IdNumDefinitionFactory): 

233 description = "NHS number" 

234 short_description = "NHS#" 

235 hl7_assigning_authority = "NHS" 

236 hl7_id_type = "NHSN" 

237 validation_method = "uk_nhs_number" 

238 

239 

240class StudyIdNumDefinitionFactory(IdNumDefinitionFactory): 

241 description = "Study number" 

242 short_description = "Study" 

243 

244 

245class RioIdNumDefinitionFactory(IdNumDefinitionFactory): 

246 description = "RiO number" 

247 short_description = "RiO" 

248 hl7_assigning_authority = "CPFT" 

249 hl7_id_type = "CPRiO" 

250 

251 

252class PatientIdNumFactory(GenericTabletRecordFactory): 

253 class Meta: 

254 model = PatientIdNum 

255 

256 id = factory.Sequence(lambda n: n + ID_OFFSET) 

257 patient = factory.SubFactory(PatientFactory) 

258 patient_id = factory.SelfAttribute("patient.id") 

259 _group = factory.SelfAttribute("patient._group") 

260 _device = factory.SelfAttribute("patient._device") 

261 

262 

263class NHSPatientIdNumFactory(PatientIdNumFactory): 

264 class Meta: 

265 exclude = PatientIdNumFactory._meta.exclude + ("iddef",) 

266 

267 iddef = factory.SubFactory(NHSIdNumDefinitionFactory) 

268 

269 which_idnum = factory.SelfAttribute("iddef.which_idnum") 

270 idnum_value = factory.LazyFunction(Fake.en_gb.nhs_number) 

271 

272 

273class RioPatientIdNumFactory(PatientIdNumFactory): 

274 class Meta: 

275 exclude = PatientIdNumFactory._meta.exclude + ("iddef",) 

276 

277 iddef = factory.SubFactory(RioIdNumDefinitionFactory) 

278 

279 which_idnum = factory.SelfAttribute("iddef.which_idnum") 

280 idnum_value = factory.Sequence(lambda n: n + RIO_ID_OFFSET) 

281 

282 

283class StudyPatientIdNumFactory(PatientIdNumFactory): 

284 class Meta: 

285 exclude = PatientIdNumFactory._meta.exclude + ("iddef",) 

286 

287 iddef = factory.SubFactory(StudyIdNumDefinitionFactory) 

288 

289 which_idnum = factory.SelfAttribute("iddef.which_idnum") 

290 idnum_value = factory.Sequence(lambda n: n + STUDY_ID_OFFSET) 

291 

292 

293class ServerCreatedPatientIdNumFactory(PatientIdNumFactory): 

294 patient = factory.SubFactory(ServerCreatedPatientFactory) 

295 

296 @factory.lazy_attribute 

297 def _device(obj: "Resolver") -> Device: 

298 # Should have been created in BasicDatabaseTestCase.setUp 

299 return Device.get_server_device( 

300 ServerCreatedPatientIdNumFactory._meta.sqlalchemy_session 

301 ) 

302 

303 @factory.lazy_attribute 

304 def _era(obj: "Resolver") -> str: 

305 return ERA_NOW 

306 

307 

308class ServerCreatedNHSPatientIdNumFactory( 

309 ServerCreatedPatientIdNumFactory, NHSPatientIdNumFactory 

310): 

311 class Meta: 

312 exclude = ( 

313 ServerCreatedPatientIdNumFactory._meta.exclude 

314 + NHSPatientIdNumFactory._meta.exclude 

315 ) 

316 

317 

318class ServerCreatedRioPatientIdNumFactory( 

319 ServerCreatedPatientIdNumFactory, RioPatientIdNumFactory 

320): 

321 class Meta: 

322 exclude = ( 

323 ServerCreatedPatientIdNumFactory._meta.exclude 

324 + RioPatientIdNumFactory._meta.exclude 

325 ) 

326 

327 

328class ServerCreatedStudyPatientIdNumFactory( 

329 ServerCreatedPatientIdNumFactory, StudyPatientIdNumFactory 

330): 

331 class Meta: 

332 exclude = ( 

333 ServerCreatedPatientIdNumFactory._meta.exclude 

334 + StudyPatientIdNumFactory._meta.exclude 

335 ) 

336 

337 

338class TaskScheduleFactory(BaseFactory): 

339 class Meta: 

340 model = TaskSchedule 

341 

342 group = factory.SubFactory(GroupFactory) 

343 name = factory.Sequence(lambda n: f"Schedule {n + ID_OFFSET}") 

344 

345 

346class TaskScheduleItemFactory(BaseFactory): 

347 class Meta: 

348 model = TaskScheduleItem 

349 

350 task_schedule = factory.SubFactory(TaskScheduleFactory) 

351 

352 

353class PatientTaskScheduleFactory(BaseFactory): 

354 class Meta: 

355 model = PatientTaskSchedule 

356 

357 task_schedule = factory.SubFactory(TaskScheduleFactory) 

358 # If patient has not been set explicitly, 

359 # ensure Patient and TaskSchedule end up in the same group 

360 start_datetime = None 

361 patient = factory.SubFactory( 

362 ServerCreatedPatientFactory, 

363 _group=factory.SelfAttribute("..task_schedule.group"), 

364 ) 

365 

366 

367class EmailFactory(BaseFactory): 

368 class Meta: 

369 model = Email 

370 

371 # Although sent and sent_at_utc are columns, they are not keyword 

372 # arguments to Email's constructor so they are populated after the object 

373 # has been created. For some reason 'sent' needs to be set explicitly 

374 # when creating the factory even though the default should be False. Might 

375 # be a SQLite thing. 

376 @factory.post_generation 

377 def sent_at_utc( 

378 obj: Email, create: bool, sent_at_utc: pendulum.DateTime, **kwargs: Any 

379 ) -> None: 

380 if not create: 

381 return 

382 

383 obj.sent_at_utc = sent_at_utc 

384 

385 @factory.post_generation 

386 def sent(obj: Email, create: bool, sent: bool, **kwargs: Any) -> None: 

387 if not create: 

388 return 

389 

390 obj.sent = sent 

391 

392 

393class PatientTaskScheduleEmailFactory(BaseFactory): 

394 class Meta: 

395 model = PatientTaskScheduleEmail 

396 

397 patient_task_schedule = factory.SubFactory( 

398 PatientTaskScheduleFactory, 

399 ) 

400 email = factory.SubFactory(EmailFactory, sent=True) 

401 

402 

403class UserGroupMembershipFactory(BaseFactory): 

404 class Meta: 

405 model = UserGroupMembership 

406 

407 

408class BlobFactory(GenericTabletRecordFactory): 

409 class Meta: 

410 model = Blob 

411 

412 id = factory.Sequence(lambda n: n + ID_OFFSET) 

413 

414 

415class DirtyTableFactory(BaseFactory): 

416 class Meta: 

417 model = DirtyTable 

418 

419 

420class SpecialNoteFactory(BaseFactory): 

421 class Meta: 

422 model = SpecialNote 

423 

424 @classmethod 

425 def create(cls, *args: Any, **kwargs: Any) -> SpecialNote: 

426 task = kwargs.pop("task", None) 

427 if task is not None: 

428 if "task_id" in kwargs: 

429 raise TypeError( 

430 "Both 'task' and 'task_id' keyword arguments " 

431 f"unexpectedly passed to {cls.__name__}. Use one or the " 

432 "other." 

433 ) 

434 kwargs["task_id"] = task.id 

435 

436 if "basetable" not in kwargs: 

437 kwargs["basetable"] = task.__tablename__ 

438 if "device_id" not in kwargs: 

439 kwargs["device_id"] = task._device.id 

440 if "era" not in kwargs: 

441 kwargs["era"] = task._era 

442 

443 return super().create(*args, **kwargs) 

444 

445 

446class ExportRecipientFactory(BaseFactory): 

447 class Meta: 

448 exclude = ("iddef",) 

449 model = ExportRecipient 

450 

451 id = factory.Sequence(lambda n: n + ID_OFFSET) 

452 

453 iddef = factory.SubFactory(IdNumDefinitionFactory) 

454 primary_idnum = factory.SelfAttribute("iddef.which_idnum")