Coverage for app/backend/src/tests/test_jail.py: 100%

1import grpc 

2import pytest 

3from google.protobuf import empty_pb2 

4 

5from couchers.constants import TOS_VERSION 

6from couchers.models import users 

7from couchers.proto import admin_pb2, api_pb2, jail_pb2 

8from couchers.servicers import jail as servicers_jail 

9from couchers.utils import create_coordinate, to_aware_datetime 

10from tests.fixtures.db import generate_user 

11from tests.fixtures.misc import EmailCollector, PushCollector 

12from tests.fixtures.sessions import real_account_session, real_admin_session, real_api_session, real_jail_session 

13 

14 

15def test_jail_basic(db): 

16 user1, token1 = generate_user() 

17 

18 with real_api_session(token1) as api: 

19 api.Ping(api_pb2.PingReq()) 

20 

21 with real_jail_session(token1) as jail: 

22 res = jail.JailInfo(empty_pb2.Empty()) 

23 # check every field is false 

24 for field in res.DESCRIPTOR.fields: 

25 assert not getattr(res, field.name) 

26 

27 assert not res.jailed 

28 

29 # make the user jailed 

30 user2, token2 = generate_user(accepted_tos=0) 

31 

32 with real_api_session(token2) as api, pytest.raises(grpc.RpcError) as e: 

33 api.Ping(api_pb2.PingReq()) 

34 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

35 

36 with real_jail_session(token2) as jail: 

37 res = jail.JailInfo(empty_pb2.Empty()) 

38 

39 assert res.jailed 

40 

41 reason_count = 0 

42 

43 # check at least one field is true 

44 for field in res.DESCRIPTOR.fields: 

45 reason_count += getattr(res, field.name) == True 

46 

47 assert reason_count > 0 

48 

49 

50def test_JailInfo(db): 

51 user1, token1 = generate_user(accepted_tos=0) 

52 

53 with real_jail_session(token1) as jail: 

54 res = jail.JailInfo(empty_pb2.Empty()) 

55 assert res.jailed 

56 assert res.has_not_accepted_tos 

57 

58 with real_api_session(token1) as api, pytest.raises(grpc.RpcError) as e: 

59 res = api.Ping(api_pb2.PingReq()) 

60 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

61 

62 # make the user not jailed 

63 user2, token2 = generate_user() 

64 

65 with real_jail_session(token2) as jail: 

66 res = jail.JailInfo(empty_pb2.Empty()) 

67 assert not res.jailed 

68 assert not res.has_not_accepted_tos 

69 

70 with real_api_session(token2) as api: 

71 res = api.Ping(api_pb2.PingReq()) 

72 

73 

74def test_AcceptTOS(db): 

75 # make them have not accepted TOS 

76 user1, token1 = generate_user(accepted_tos=0) 

77 

78 with real_jail_session(token1) as jail: 

79 res = jail.JailInfo(empty_pb2.Empty()) 

80 assert res.jailed 

81 assert res.has_not_accepted_tos 

82 

83 # make sure we can't unaccept 

84 with pytest.raises(grpc.RpcError) as e: 

85 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

86 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

87 assert e.value.details() == "You cannot revoke acceptance of the Terms of Service." 

88 

89 res = jail.JailInfo(empty_pb2.Empty()) 

90 assert res.jailed 

91 assert res.has_not_accepted_tos 

92 

93 # now accept 

94 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

95 

96 res = jail.JailInfo(empty_pb2.Empty()) 

97 assert not res.jailed 

98 assert not res.has_not_accepted_tos 

99 

100 # make sure we can't unaccept 

101 with pytest.raises(grpc.RpcError) as e: 

102 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

103 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

104 assert e.value.details() == "You cannot revoke acceptance of the Terms of Service." 

105 

106 # make them have accepted TOS 

107 user2, token2 = generate_user() 

108 

109 with real_jail_session(token2) as jail: 

110 res = jail.JailInfo(empty_pb2.Empty()) 

111 assert not res.jailed 

112 assert not res.has_not_accepted_tos 

113 

114 # make sure we can't unaccept 

115 with pytest.raises(grpc.RpcError) as e: 

116 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

117 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

118 assert e.value.details() == "You cannot revoke acceptance of the Terms of Service." 

119 

120 # accepting again doesn't do anything 

121 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

122 

123 res = jail.JailInfo(empty_pb2.Empty()) 

124 assert not res.jailed 

125 assert not res.has_not_accepted_tos 

126 

127 

128def test_TOS_increase(db, monkeypatch): 

129 # test if the TOS version is updated 

130 

131 # not jailed yet 

132 user, token = generate_user() 

133 

134 with real_jail_session(token) as jail: 

135 res = jail.JailInfo(empty_pb2.Empty()) 

136 assert not res.jailed 

137 assert not res.has_not_accepted_tos 

138 

139 with real_api_session(token) as api: 

140 res = api.Ping(api_pb2.PingReq()) 

141 

142 # now we pretend to update the TOS version 

143 new_TOS_VERSION = TOS_VERSION + 1 

144 

145 monkeypatch.setattr(users, "TOS_VERSION", new_TOS_VERSION) 

146 monkeypatch.setattr(servicers_jail, "TOS_VERSION", new_TOS_VERSION) 

147 

148 # make sure we're jailed 

149 with real_api_session(token) as api, pytest.raises(grpc.RpcError) as e: 

150 api.Ping(api_pb2.PingReq()) 

151 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

152 

153 with real_jail_session(token) as jail: 

154 res = jail.JailInfo(empty_pb2.Empty()) 

155 assert res.jailed 

156 assert res.has_not_accepted_tos 

157 

158 # now accept 

159 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

160 

161 res = jail.JailInfo(empty_pb2.Empty()) 

162 assert not res.jailed 

163 assert not res.has_not_accepted_tos 

164 

165 

166def test_SetLocation(db): 

167 # make them need to update location 

168 user1, token1 = generate_user(geom=create_coordinate(0, 0), geom_radius=0, needs_to_update_location=True) 

169 

170 

171def test_MarkUserNeedsLocationUpdate(db): 

172 user, token = generate_user() 

173 super_user, super_token = generate_user(is_superuser=True) 

174 

175 with real_jail_session(token) as jail: 

176 res = jail.JailInfo(empty_pb2.Empty()) 

177 assert not res.jailed 

178 assert not res.needs_to_update_location 

179 assert len(res.pending_mod_notes) == 0 

180 

181 with real_admin_session(super_token) as admin: 

182 admin.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=user.username)) 

183 

184 with real_jail_session(token) as jail: 

185 res = jail.JailInfo(empty_pb2.Empty()) 

186 assert res.jailed 

187 assert res.needs_to_update_location 

188 

189 res = jail.SetLocation( 

190 jail_pb2.SetLocationReq( 

191 city="New York City", 

192 lat=40.7812, 

193 lng=-73.9647, 

194 radius=250, 

195 ) 

196 ) 

197 

198 assert not res.jailed 

199 assert not res.needs_to_update_location 

200 

201 res = jail.JailInfo(empty_pb2.Empty()) 

202 assert not res.jailed 

203 assert not res.needs_to_update_location 

204 

205 

206def test_AcceptCommunityGuidelines(db): 

207 # make them have not accepted GC 

208 user1, token1 = generate_user(accepted_community_guidelines=0) 

209 

210 with real_jail_session(token1) as jail: 

211 res = jail.JailInfo(empty_pb2.Empty()) 

212 assert res.jailed 

213 assert res.has_not_accepted_community_guidelines 

214 

215 # make sure we can't unaccept 

216 with pytest.raises(grpc.RpcError) as e: 

217 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

218 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

219 assert e.value.details() == "You cannot revoke acceptance of the Community Guidelines." 

220 

221 res = jail.JailInfo(empty_pb2.Empty()) 

222 assert res.jailed 

223 assert res.has_not_accepted_community_guidelines 

224 

225 # now accept 

226 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=True)) 

227 

228 res = jail.JailInfo(empty_pb2.Empty()) 

229 assert not res.jailed 

230 assert not res.has_not_accepted_community_guidelines 

231 

232 # make sure we can't unaccept 

233 with pytest.raises(grpc.RpcError) as e: 

234 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

235 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

236 assert e.value.details() == "You cannot revoke acceptance of the Community Guidelines." 

237 

238 # make them have accepted GC 

239 user2, token2 = generate_user() 

240 

241 with real_jail_session(token2) as jail: 

242 res = jail.JailInfo(empty_pb2.Empty()) 

243 assert not res.jailed 

244 assert not res.has_not_accepted_community_guidelines 

245 

246 # make sure we can't unaccept 

247 with pytest.raises(grpc.RpcError) as e: 

248 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

249 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

250 assert e.value.details() == "You cannot revoke acceptance of the Community Guidelines." 

251 

252 # accepting again doesn't do anything 

253 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=True)) 

254 

255 res = jail.JailInfo(empty_pb2.Empty()) 

256 assert not res.jailed 

257 assert not res.has_not_accepted_community_guidelines 

258 

259 

260def test_modnotes(db, email_collector: EmailCollector, push_collector: PushCollector): 

261 user, token = generate_user() 

262 super_user, super_token = generate_user(is_superuser=True) 

263 

264 with real_jail_session(token) as jail: 

265 res = jail.JailInfo(empty_pb2.Empty()) 

266 assert not res.jailed 

267 assert not res.has_pending_mod_notes 

268 assert len(res.pending_mod_notes) == 0 

269 

270 with real_account_session(token) as account: 

271 res = account.ListModNotes(empty_pb2.Empty()) 

272 assert len(res.mod_notes) == 0 

273 

274 with real_admin_session(super_token) as admin: 

275 admin.SendModNote( 

276 admin_pb2.SendModNoteReq( 

277 user=user.username, 

278 content="# Important note\nThis is a sample mod note.", 

279 internal_id="sample_note", 

280 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY, 

281 ) 

282 ) 

283 

284 email = email_collector.pop_for_recipient(user.email, last=True) 

285 assert email.subject == "[TEST] You have received a moderator note" 

286 assert email.sender_name == "Couchers.org Moderation" 

287 assert email.sender_email == "moderation@couchers.org.invalid" 

288 assert "This is a sample mod note." not in email.plain 

289 

290 push = push_collector.pop_for_user(user.id, last=True) 

291 assert push.content.title == "New moderator note" 

292 assert push.content.body == "You received a moderator note. Read and acknowledge it to continue using the platform." 

293 

294 with real_jail_session(token) as jail: 

295 res = jail.JailInfo(empty_pb2.Empty()) 

296 assert res.jailed 

297 assert res.has_pending_mod_notes 

298 assert len(res.pending_mod_notes) == 1 

299 note = res.pending_mod_notes[0] 

300 assert note.note_content == "# Important note\nThis is a sample mod note." 

301 

302 note_id = note.note_id 

303 

304 with pytest.raises(grpc.RpcError) as e: 

305 jail.AcknowledgePendingModNote( 

306 jail_pb2.AcknowledgePendingModNoteReq( 

307 note_id=note_id, 

308 acknowledge=False, 

309 ) 

310 ) 

311 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

312 assert e.value.details() == "You need to read and acknowledge the moderator note." 

313 

314 assert res.jailed 

315 assert res.has_pending_mod_notes 

316 assert len(res.pending_mod_notes) == 1 

317 note = res.pending_mod_notes[0] 

318 assert note.note_content == "# Important note\nThis is a sample mod note." 

319 

320 res = jail.AcknowledgePendingModNote( 

321 jail_pb2.AcknowledgePendingModNoteReq( 

322 note_id=note_id, 

323 acknowledge=True, 

324 ) 

325 ) 

326 assert not res.jailed 

327 assert not res.has_pending_mod_notes 

328 assert len(res.pending_mod_notes) == 0 

329 

330 with pytest.raises(grpc.RpcError) as e: 

331 jail.AcknowledgePendingModNote( 

332 jail_pb2.AcknowledgePendingModNoteReq( 

333 note_id=note_id, 

334 acknowledge=False, 

335 ) 

336 ) 

337 assert e.value.code() == grpc.StatusCode.NOT_FOUND 

338 assert e.value.details() == "Moderator note not found." 

339 

340 with real_account_session(token) as account: 

341 res = account.ListModNotes(empty_pb2.Empty()) 

342 assert len(res.mod_notes) == 1 

343 note = res.mod_notes[0] 

344 assert note.note_id == note_id 

345 assert note.note_content == "# Important note\nThis is a sample mod note." 

346 

347 assert to_aware_datetime(note.acknowledged) > to_aware_datetime(note.created) 

348 

349 

350def test_modnotes_notify_with_content(db, email_collector: EmailCollector, push_collector: PushCollector): 

351 user, token = generate_user() 

352 super_user, super_token = generate_user(is_superuser=True) 

353 

354 with real_admin_session(super_token) as admin: 

355 admin.SendModNote( 

356 admin_pb2.SendModNoteReq( 

357 user=user.username, 

358 content="# Important note\nPlease remove your phone number.", 

359 internal_id="sample_note", 

360 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT, 

361 ) 

362 ) 

363 

364 email = email_collector.pop_for_recipient(user.email, last=True) 

365 assert email.subject == "[TEST] You have received a moderator note" 

366 # sent from the moderation mailbox so the user can reply to it 

367 assert email.sender_name == "Couchers.org Moderation" 

368 assert email.sender_email == "moderation@couchers.org.invalid" 

369 assert "Please remove your phone number." in email.plain 

370 assert "Please remove your phone number." in email.html 

371 assert "replying to this email" in email.plain 

372 

373 # the note itself is never included in the push notification 

374 push = push_collector.pop_for_user(user.id, last=True) 

375 assert "phone number" not in push.content.body 

376 

377 with real_jail_session(token) as jail: 

378 res = jail.JailInfo(empty_pb2.Empty()) 

379 assert res.jailed 

380 assert res.has_pending_mod_notes 

381 

382 

383def test_modnotes_notification_must_be_specified(db): 

384 user, token = generate_user() 

385 super_user, super_token = generate_user(is_superuser=True) 

386 

387 with real_admin_session(super_token) as admin: 

388 with pytest.raises(grpc.RpcError) as e: 

389 admin.SendModNote( 

390 admin_pb2.SendModNoteReq( 

391 user=user.username, 

392 content="A note", 

393 internal_id="sample_note", 

394 ) 

395 ) 

396 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

397 assert e.value.details() == "Whether and how to notify the user about the mod note must be specified." 

398 

399 with real_jail_session(token) as jail: 

400 res = jail.JailInfo(empty_pb2.Empty()) 

401 assert not res.jailed 

402 assert not res.has_pending_mod_notes 

403 

404 

405def test_modnotes_no_notify(db, email_collector: EmailCollector, push_collector: PushCollector): 

406 user, token = generate_user() 

407 super_user, super_token = generate_user(is_superuser=True) 

408 

409 with real_jail_session(token) as jail: 

410 res = jail.JailInfo(empty_pb2.Empty()) 

411 assert not res.jailed 

412 assert not res.has_pending_mod_notes 

413 assert len(res.pending_mod_notes) == 0 

414 

415 with real_account_session(token) as account: 

416 res = account.ListModNotes(empty_pb2.Empty()) 

417 assert len(res.mod_notes) == 0 

418 

419 with real_admin_session(super_token) as admin: 

420 admin.SendModNote( 

421 admin_pb2.SendModNoteReq( 

422 user=user.username, 

423 content="# Important note\nThis is a sample mod note.", 

424 internal_id="sample_note", 

425 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NONE, 

426 ) 

427 ) 

428 

429 assert email_collector.count_for_recipient(user.email) == 0 

430 

431 with real_jail_session(token) as jail: 

432 res = jail.JailInfo(empty_pb2.Empty()) 

433 assert res.jailed 

434 assert res.has_pending_mod_notes 

435 assert len(res.pending_mod_notes) == 1 

436 note = res.pending_mod_notes[0] 

437 assert note.note_content == "# Important note\nThis is a sample mod note."