Coverage for website/mailinglists/tests/test_gsuite.py: 99.31%

1from unittest import mock 

2from unittest.mock import MagicMock 

3 

4from django.conf import settings 

5from django.test import TestCase, override_settings 

6 

7from googleapiclient.errors import HttpError 

8from httplib2 import Response 

9 

10from mailinglists.gsuite import GSuiteSyncService 

11from mailinglists.models import ListAlias, MailingList, VerbatimAddress 

12 

13 

14def assert_not_called_with(self, *args, **kwargs): 

15 try: 

16 self.assert_any_call(*args, **kwargs) 

17 except AssertionError: 

18 return 

19 raise AssertionError( 

20 f"Expected {self._format_mock_call_signature(args, kwargs)} to not have been called." 

21 ) 

22 

23 

24MagicMock.assert_not_called_with = assert_not_called_with 

25 

26 

27@override_settings(SUSPEND_SIGNALS=True) 

28class GSuiteSyncTestCase(TestCase): 

29 @classmethod 

30 def setUpTestData(cls): 

31 cls.settings_api = MagicMock() 

32 cls.directory_api = MagicMock() 

33 

34 cls.sync_service = GSuiteSyncService(cls.settings_api, cls.directory_api) 

35 cls.mailinglist = MailingList.objects.create( 

36 name="new_group", description="some description", moderated=False 

37 ) 

38 ListAlias.objects.create(mailinglist=cls.mailinglist, alias="alias2") 

39 VerbatimAddress.objects.create( 

40 mailinglist=cls.mailinglist, address=f"test2@{settings.GSUITE_DOMAIN}" 

41 ) 

42 

43 def setUp(self): 

44 self.settings_api.reset_mock() 

45 self.directory_api.reset_mock() 

46 

47 self.assertEqual(len(self.sync_service._get_default_lists()), 15) 

48 

49 def test_automatic_to_group(self): 

50 group = GSuiteSyncService._automatic_to_group( 

51 { 

52 "moderated": False, 

53 "name": "new_group", 

54 "description": "some description", 

55 "aliases": ["alias1"], 

56 "addresses": [f"test1@{settings.GSUITE_DOMAIN}"], 

57 } 

58 ) 

59 self.assertEqual( 

60 group, 

61 GSuiteSyncService.GroupData( 

62 "new_group", 

63 "some description", 

64 False, 

65 ["alias1"], 

66 [f"test1@{settings.GSUITE_DOMAIN}"], 

67 ), 

68 ) 

69 

70 def test_mailing_list_to_group(self): 

71 group = GSuiteSyncService.mailing_list_to_group(self.mailinglist) 

72 self.assertEqual( 

73 group, 

74 GSuiteSyncService.GroupData( 

75 "new_group", 

76 "some description", 

77 False, 

78 ["alias2"], 

79 [f"test2@{settings.GSUITE_DOMAIN}"], 

80 "new_group", 

81 ), 

82 ) 

83 

84 def test_group_settings(self): 

85 self.assertEqual( 

86 self.sync_service._group_settings(False), 

87 { 

88 "allowExternalMembers": "true", 

89 "allowWebPosting": "true", 

90 "archiveOnly": "false", 

91 "enableCollaborativeInbox": "true", 

92 "isArchived": "true", 

93 "membersCanPostAsTheGroup": "true", 

94 "messageModerationLevel": "MODERATE_NONE", 

95 "replyTo": "REPLY_TO_SENDER", 

96 "whoCanAssistContent": "ALL_MEMBERS", 

97 "whoCanContactOwner": "ALL_MANAGERS_CAN_CONTACT", 

98 "whoCanDiscoverGroup": "ALL_MEMBERS_CAN_DISCOVER", 

99 "whoCanJoin": "INVITED_CAN_JOIN", 

100 "whoCanLeaveGroup": "NONE_CAN_LEAVE", 

101 "whoCanModerateContent": "OWNERS_AND_MANAGERS", 

102 "whoCanModerateMembers": "NONE", 

103 "whoCanPostMessage": "ANYONE_CAN_POST", 

104 "whoCanViewGroup": "ALL_MEMBERS_CAN_VIEW", 

105 "whoCanViewMembership": "ALL_MANAGERS_CAN_VIEW", 

106 }, 

107 ) 

108 self.assertEqual( 

109 self.sync_service._group_settings(True), 

110 { 

111 "allowExternalMembers": "true", 

112 "allowWebPosting": "true", 

113 "archiveOnly": "false", 

114 "enableCollaborativeInbox": "true", 

115 "isArchived": "true", 

116 "membersCanPostAsTheGroup": "true", 

117 "messageModerationLevel": "MODERATE_ALL_MESSAGES", 

118 "replyTo": "REPLY_TO_SENDER", 

119 "whoCanAssistContent": "ALL_MEMBERS", 

120 "whoCanContactOwner": "ALL_MANAGERS_CAN_CONTACT", 

121 "whoCanDiscoverGroup": "ALL_MEMBERS_CAN_DISCOVER", 

122 "whoCanJoin": "INVITED_CAN_JOIN", 

123 "whoCanLeaveGroup": "NONE_CAN_LEAVE", 

124 "whoCanModerateContent": "OWNERS_AND_MANAGERS", 

125 "whoCanModerateMembers": "NONE", 

126 "whoCanPostMessage": "ANYONE_CAN_POST", 

127 "whoCanViewGroup": "ALL_MEMBERS_CAN_VIEW", 

128 "whoCanViewMembership": "ALL_MANAGERS_CAN_VIEW", 

129 }, 

130 ) 

131 

132 @mock.patch("mailinglists.gsuite.logger") 

133 def test_create_group(self, logger_mock): 

134 with self.subTest("Successful"): 

135 self.sync_service.create_group( 

136 GSuiteSyncService.GroupData( 

137 "new_group", 

138 "some description", 

139 False, 

140 ["alias2"], 

141 [f"test2@{settings.GSUITE_DOMAIN}"], 

142 ) 

143 ) 

144 

145 self.directory_api.groups().insert.assert_called_once_with( 

146 body={ 

147 "email": f"new_group@{settings.GSUITE_DOMAIN}", 

148 "name": "new_group", 

149 "description": "some description", 

150 } 

151 ) 

152 

153 self.settings_api.groups().update.assert_called_once_with( 

154 groupUniqueId=f"new_group@{settings.GSUITE_DOMAIN}", 

155 body=self.sync_service._group_settings(False), 

156 ) 

157 

158 self.directory_api.members().list.assert_called() 

159 self.directory_api.groups().aliases().list.assert_called() 

160 

161 self.settings_api.reset_mock() 

162 self.directory_api.reset_mock() 

163 

164 with self.subTest("Failure"): 

165 self.directory_api.groups().insert().execute.side_effect = HttpError( 

166 Response({"status": 500}), b"" 

167 ) 

168 

169 self.sync_service.create_group( 

170 GSuiteSyncService.GroupData( 

171 "new_group", 

172 "some description", 

173 False, 

174 ["alias2"], 

175 [f"test2@{settings.GSUITE_DOMAIN}"], 

176 ) 

177 ) 

178 

179 self.directory_api.members().list.assert_not_called() 

180 self.directory_api.groups().aliases().list.assert_not_called() 

181 

182 logger_mock.exception.assert_called_once_with( 

183 "Could not successfully finish creating the list new_group:", 

184 ) 

185 

186 @mock.patch("mailinglists.gsuite.logger") 

187 def test_update_group(self, logger_mock): 

188 with self.subTest("Successful"): 

189 self.sync_service.update_group( 

190 "new_group", 

191 GSuiteSyncService.GroupData( 

192 "new_group", 

193 "some description", 

194 False, 

195 ["alias2"], 

196 [f"test2@{settings.GSUITE_DOMAIN}"], 

197 ), 

198 ) 

199 

200 self.directory_api.groups().update.assert_called_once_with( 

201 body={ 

202 "email": f"new_group@{settings.GSUITE_DOMAIN}", 

203 "name": "new_group", 

204 "description": "some description", 

205 }, 

206 groupKey=f"new_group@{settings.GSUITE_DOMAIN}", 

207 ) 

208 

209 self.settings_api.groups().update.assert_called_once_with( 

210 groupUniqueId=f"new_group@{settings.GSUITE_DOMAIN}", 

211 body=self.sync_service._group_settings(False), 

212 ) 

213 

214 self.directory_api.members().list.assert_called() 

215 self.directory_api.groups().aliases().list.assert_called() 

216 

217 self.settings_api.reset_mock() 

218 self.directory_api.reset_mock() 

219 

220 with self.subTest("Failure"): 

221 self.directory_api.groups().update().execute.side_effect = HttpError( 

222 Response({"status": 500}), b"" 

223 ) 

224 

225 self.sync_service.update_group( 

226 "new_group", 

227 GSuiteSyncService.GroupData( 

228 "new_group", 

229 "some description", 

230 False, 

231 ["alias2"], 

232 [f"test2@{settings.GSUITE_DOMAIN}"], 

233 ), 

234 ) 

235 

236 self.directory_api.members().list.assert_not_called() 

237 self.directory_api.groups().aliases().list.assert_not_called() 

238 

239 logger_mock.exception.assert_called_once_with( 

240 "Could not update list new_group" 

241 ) 

242 

243 @mock.patch("mailinglists.gsuite.logger") 

244 def test_archive_group(self, logger_mock): 

245 with self.subTest("Successful"): 

246 self.sync_service.archive_group("new_group") 

247 

248 self.settings_api.groups().patch.assert_called_once_with( 

249 body={"archiveOnly": "true", "whoCanPostMessage": "NONE_CAN_POST"}, 

250 groupUniqueId=f"new_group@{settings.GSUITE_DOMAIN}", 

251 ) 

252 

253 self.directory_api.members().list.assert_called() 

254 self.directory_api.groups().aliases().list.assert_called() 

255 

256 self.settings_api.reset_mock() 

257 self.directory_api.reset_mock() 

258 

259 with self.subTest("Failure"): 

260 self.settings_api.groups().patch().execute.side_effect = HttpError( 

261 Response({"status": 500}), b"" 

262 ) 

263 

264 self.sync_service.archive_group("new_group") 

265 

266 self.directory_api.members().list.assert_not_called() 

267 self.directory_api.groups().aliases().list.assert_not_called() 

268 

269 logger_mock.exception.assert_called_once_with( 

270 "Could not archive list new_group" 

271 ) 

272 

273 @mock.patch("mailinglists.gsuite.logger") 

274 def test_update_group_aliases(self, logger_mock): 

275 with self.subTest("Error getting existing list"): 

276 self.directory_api.groups().aliases().list().execute.side_effect = ( 

277 HttpError(Response({"status": 500}), b"") 

278 ) 

279 self.sync_service._update_group_aliases( 

280 GSuiteSyncService.GroupData(name="update_group") 

281 ) 

282 

283 logger_mock.exception.assert_called_once_with( 

284 "Could not obtain existing aliases for list update_group:", 

285 ) 

286 

287 self.directory_api.reset_mock() 

288 

289 with self.subTest("Successful with some errors"): 

290 group_data = GSuiteSyncService.GroupData( 

291 name="update_group", 

292 aliases=["not_synced", "not_synced_error", "already_synced"], 

293 ) 

294 

295 existing_aliases = [ 

296 {"alias": f"deleteme@{settings.GSUITE_DOMAIN}"}, 

297 {"alias": f"deleteme_error@{settings.GSUITE_DOMAIN}"}, 

298 {"alias": f"already_synced@{settings.GSUITE_DOMAIN}"}, 

299 ] 

300 

301 self.directory_api.groups().aliases().list().execute.side_effect = [ 

302 {"aliases": existing_aliases} 

303 ] 

304 

305 self.directory_api.groups().aliases().insert().execute.side_effect = [ 

306 "success", 

307 HttpError(Response({"status": 500}), b""), 

308 ] 

309 

310 self.directory_api.groups().aliases().delete().execute.side_effect = [ 

311 "success", 

312 HttpError(Response({"status": 500}), b""), 

313 ] 

314 

315 self.sync_service._update_group_aliases(group_data) 

316 

317 self.directory_api.groups().aliases().insert.assert_any_call( 

318 groupKey=f"update_group@{settings.GSUITE_DOMAIN}", 

319 body={"alias": f"not_synced@{settings.GSUITE_DOMAIN}"}, 

320 ) 

321 

322 self.directory_api.groups().aliases().delete.assert_any_call( 

323 groupKey=f"update_group@{settings.GSUITE_DOMAIN}", 

324 alias=f"deleteme@{settings.GSUITE_DOMAIN}", 

325 ) 

326 

327 @mock.patch("mailinglists.gsuite.logger") 

328 def test_update_group_members(self, logger_mock): 

329 with self.subTest("Error getting existing list"): 

330 self.directory_api.members().list().execute.side_effect = HttpError( 

331 Response({"status": 500}), b"" 

332 ) 

333 self.sync_service._update_group_members( 

334 GSuiteSyncService.GroupData(name="update_group") 

335 ) 

336 

337 logger_mock.exception.assert_called_once_with( 

338 "Could not obtain list member data for update_group" 

339 ) 

340 

341 self.directory_api.reset_mock() 

342 

343 with self.subTest("Successful with some errors"): 

344 group_data = GSuiteSyncService.GroupData( 

345 name="update_group", 

346 addresses=[ 

347 "not_synced@example.com", 

348 "not_synced_error@example.com", 

349 "already_synced@example.com", 

350 ], 

351 ) 

352 

353 existing_aliases = [ 

354 {"email": "deleteme@example.com", "role": "MEMBER"}, 

355 {"email": "deleteme_error@example.com", "role": "MEMBER"}, 

356 {"email": "already_synced@example.com", "role": "MEMBER"}, 

357 {"email": "donotdelete@example.com", "role": "MANAGER"}, 

358 ] 

359 

360 self.directory_api.members().list().execute.side_effect = [ 

361 {"members": existing_aliases[:1], "nextPageToken": "some_token"}, 

362 {"members": existing_aliases[1:]}, 

363 ] 

364 

365 self.directory_api.members().insert().execute.side_effect = [ 

366 "success", 

367 HttpError(Response({"status": 500}), b""), 

368 ] 

369 

370 self.directory_api.members().delete().execute.side_effect = [ 

371 "success", 

372 HttpError(Response({"status": 500}), b""), 

373 ] 

374 

375 self.sync_service._update_group_members(group_data) 

376 

377 self.directory_api.members().insert.assert_any_call( 

378 groupKey=f"update_group@{settings.GSUITE_DOMAIN}", 

379 body={"email": "not_synced@example.com", "role": "MEMBER"}, 

380 ) 

381 

382 self.directory_api.members().delete.assert_any_call( 

383 groupKey=f"update_group@{settings.GSUITE_DOMAIN}", 

384 memberKey="deleteme@example.com", 

385 ) 

386 

387 self.directory_api.members().delete.assert_not_called_with( 

388 groupKey=f"update_group@{settings.GSUITE_DOMAIN}", 

389 memberKey="donotdelete@example.com", 

390 ) 

391 

392 @mock.patch("mailinglists.gsuite.logger") 

393 def test_sync_mailing_lists(self, logger_mock): 

394 original_create = self.sync_service.create_group 

395 original_update = self.sync_service.update_group 

396 original_archive = self.sync_service.archive_group 

397 

398 self.sync_service.create_group = MagicMock() 

399 self.sync_service.update_group = MagicMock() 

400 self.sync_service.archive_group = MagicMock() 

401 

402 with self.subTest("Error getting existing list"): 

403 self.directory_api.groups().list().execute.side_effect = HttpError( 

404 Response({"status": 500}), b"" 

405 ) 

406 self.sync_service.sync_mailing_lists() 

407 

408 logger_mock.exception.assert_called_with( 

409 "Could not get the existing groups" 

410 ) 

411 

412 self.directory_api.reset_mock() 

413 

414 with self.subTest("Successful"): 

415 existing_groups = [ 

416 {"name": "deleteme", "directMembersCount": "3"}, 

417 {"name": "already_synced", "directMembersCount": "2"}, 

418 {"name": "ignore", "directMembersCount": "0"}, 

419 ] 

420 

421 self.directory_api.groups().list().execute.side_effect = [ 

422 {"groups": existing_groups[:1], "nextPageToken": "some_token"}, 

423 {"groups": existing_groups[1:]}, 

424 ] 

425 

426 self.sync_service.sync_mailing_lists( 

427 [ 

428 GSuiteSyncService.GroupData(name="syncme", addresses=["someone"]), 

429 GSuiteSyncService.GroupData( 

430 name="already_synced", addresses=["someone"] 

431 ), 

432 GSuiteSyncService.GroupData(name="ignore2", addresses=[]), 

433 ] 

434 ) 

435 

436 self.sync_service.create_group.assert_called_with( 

437 GSuiteSyncService.GroupData(name="syncme", addresses=["someone"]) 

438 ) 

439 

440 self.sync_service.update_group.assert_called_with( 

441 "already_synced", 

442 GSuiteSyncService.GroupData( 

443 name="already_synced", addresses=["someone"] 

444 ), 

445 ) 

446 

447 self.sync_service.archive_group.assert_called_with("deleteme") 

448 

449 self.sync_service.create_group.assert_not_called_with( 

450 GSuiteSyncService.GroupData(name="ignore2", addresses=[]) 

451 ) 

452 self.sync_service.update_group.assert_not_called_with( 

453 "ignore2", GSuiteSyncService.GroupData(name="ignore2", addresses=[]) 

454 ) 

455 self.sync_service.archive_group.assert_not_called_with("ignore2") 

456 

457 self.sync_service.create_group = original_create 

458 self.sync_service.update_group = original_update 

459 self.sync_service.archive_group = original_archive