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

362 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-19 15:47 +0000

1from datetime import timedelta 

2from pathlib import Path 

3from unittest.mock import patch 

4 

5import grpc 

6import pytest 

7from google.protobuf import empty_pb2 

8from sqlalchemy import select 

9 

10from couchers.config import config 

11from couchers.constants import ( 

12 POSTAL_VERIFICATION_CODE_LIFETIME, 

13 POSTAL_VERIFICATION_MAX_ATTEMPTS, 

14 POSTAL_VERIFICATION_RATE_LIMIT, 

15) 

16from couchers.db import session_scope 

17from couchers.helpers.postal_verification import generate_postal_verification_code, has_postal_verification 

18from couchers.jobs.handlers import check_mypostcard_jobs 

19from couchers.jobs.worker import process_job 

20from couchers.models import User 

21from couchers.models.postal_verification import PostalVerificationAttempt, PostalVerificationStatus 

22from couchers.postal.my_postcard import _generate_back_left_side_png 

23from couchers.proto import postal_verification_pb2 

24from couchers.resources import get_postcard_front_image 

25from couchers.utils import now 

26from tests.fixtures.db import generate_user 

27from tests.fixtures.sessions import postal_verification_session 

28 

29 

30def test_generate_postal_verification_code(): 

31 """Test that generated codes meet requirements.""" 

32 allowed = set("ABCDEFGHJKLMNPQRSTUVWXYZ23456789") 

33 for _ in range(100): 

34 code = generate_postal_verification_code() 

35 assert len(code) == 6 

36 assert all(c in allowed for c in code) 

37 # Should not contain confusing characters 

38 for char in "IO01": 

39 assert char not in code 

40 

41 

42def test_postal_verification_disabled(db, feature_flags): 

43 """Test that postal verification is disabled.""" 

44 feature_flags.set("postal_verification_enabled", False) 

45 user, token = generate_user() 

46 

47 with postal_verification_session(token) as pv: 

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

49 pv.InitiatePostalVerification( 

50 postal_verification_pb2.InitiatePostalVerificationReq( 

51 address=postal_verification_pb2.PostalAddress( 

52 address_line_1="123 Main St", 

53 city="Test City", 

54 country_code="US", 

55 ) 

56 ) 

57 ) 

58 assert e.value.code() == grpc.StatusCode.UNAVAILABLE 

59 

60 

61def test_postal_verification_confirm_disabled(db, feature_flags): 

62 """Confirming (which queues the paid postcard) must respect the flag, not just initiation.""" 

63 feature_flags.set("postal_verification_enabled", False) 

64 user, token = generate_user() 

65 

66 # Seed a pending attempt directly, since initiation is gated by the same flag. 

67 with session_scope() as session: 

68 attempt = PostalVerificationAttempt( 

69 user_id=user.id, 

70 status=PostalVerificationStatus.pending_address_confirmation, 

71 address_line_1="123 Main St", 

72 city="Test City", 

73 country_code="US", 

74 ) 

75 session.add(attempt) 

76 session.flush() 

77 attempt_id = attempt.id 

78 

79 with postal_verification_session(token) as pv: 

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

81 pv.ConfirmPostalAddress( 

82 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

83 ) 

84 assert e.value.code() == grpc.StatusCode.UNAVAILABLE 

85 

86 

87def test_postal_verification_happy_path(db): 

88 """Test the complete happy path for postal verification.""" 

89 user, token = generate_user() 

90 

91 # Check initial status 

92 with postal_verification_session(token) as pv: 

93 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

94 assert not status.has_postal_verification 

95 assert status.can_initiate_new_attempt 

96 assert not status.has_active_attempt 

97 

98 # Step 1: Initiate postal verification 

99 with postal_verification_session(token) as pv: 

100 res = pv.InitiatePostalVerification( 

101 postal_verification_pb2.InitiatePostalVerificationReq( 

102 address=postal_verification_pb2.PostalAddress( 

103 address_line_1="123 Main St", 

104 address_line_2="Apt 4", 

105 city="Test City", 

106 state="CA", 

107 postal_code="12345", 

108 country_code="US", 

109 ) 

110 ) 

111 ) 

112 attempt_id = res.postal_verification_attempt_id 

113 assert attempt_id > 0 

114 

115 # Check status after initiation 

116 with postal_verification_session(token) as pv: 

117 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

118 assert not status.has_postal_verification 

119 assert not status.can_initiate_new_attempt 

120 assert status.has_active_attempt 

121 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_PENDING_ADDRESS_CONFIRMATION 

122 

123 # Step 2: Confirm address 

124 with postal_verification_session(token) as pv: 

125 pv.ConfirmPostalAddress( 

126 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

127 ) 

128 

129 # Check status after confirmation 

130 with postal_verification_session(token) as pv: 

131 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

132 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_IN_PROGRESS 

133 

134 # Process background job to send postcard 

135 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

136 mock_send.return_value = 12345 

137 while process_job(): 

138 pass 

139 

140 # Check status after postcard sent 

141 with postal_verification_session(token) as pv: 

142 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

143 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_AWAITING_VERIFICATION 

144 assert status.postcard_sent_at.seconds > 0 

145 

146 # Get the verification code from the database 

147 with session_scope() as session: 

148 attempt = session.execute( 

149 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

150 ).scalar_one() 

151 verification_code = attempt.verification_code 

152 

153 # Step 3: Verify the code 

154 with postal_verification_session(token) as pv: 

155 res = pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code=verification_code)) 

156 assert res.success 

157 

158 # Check final status 

159 with postal_verification_session(token) as pv: 

160 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

161 assert status.has_postal_verification 

162 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_SUCCEEDED 

163 

164 # Verify with helper function 

165 with session_scope() as session: 

166 db_user = session.execute(select(User).where(User.id == user.id)).scalar_one() 

167 assert has_postal_verification(session, db_user) 

168 

169 

170def test_postal_verification_wrong_code(db): 

171 """Test entering wrong verification codes.""" 

172 user, token = generate_user() 

173 

174 # Initiate and confirm 

175 with postal_verification_session(token) as pv: 

176 res = pv.InitiatePostalVerification( 

177 postal_verification_pb2.InitiatePostalVerificationReq( 

178 address=postal_verification_pb2.PostalAddress( 

179 address_line_1="123 Main St", 

180 city="Test City", 

181 country_code="US", 

182 ) 

183 ) 

184 ) 

185 attempt_id = res.postal_verification_attempt_id 

186 

187 with postal_verification_session(token) as pv: 

188 pv.ConfirmPostalAddress( 

189 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

190 ) 

191 

192 # Process background job 

193 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

194 mock_send.return_value = 12345 

195 while process_job(): 

196 pass 

197 

198 # Try wrong codes 

199 with postal_verification_session(token) as pv: 

200 for i in range(POSTAL_VERIFICATION_MAX_ATTEMPTS - 1): 

201 res = pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code="WRONGX")) 

202 assert not res.success 

203 assert res.remaining_attempts == POSTAL_VERIFICATION_MAX_ATTEMPTS - 1 - i 

204 

205 # Last attempt should fail and lock the attempt 

206 res = pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code="WRONGX")) 

207 assert not res.success 

208 assert res.remaining_attempts == 0 

209 

210 # Check status is failed 

211 with postal_verification_session(token) as pv: 

212 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

213 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_FAILED 

214 

215 

216def test_postal_verification_code_expiry(db): 

217 """Test that codes expire after the configured lifetime.""" 

218 user, token = generate_user() 

219 

220 # Initiate and confirm 

221 with postal_verification_session(token) as pv: 

222 res = pv.InitiatePostalVerification( 

223 postal_verification_pb2.InitiatePostalVerificationReq( 

224 address=postal_verification_pb2.PostalAddress( 

225 address_line_1="123 Main St", 

226 city="Test City", 

227 country_code="US", 

228 ) 

229 ) 

230 ) 

231 attempt_id = res.postal_verification_attempt_id 

232 

233 with postal_verification_session(token) as pv: 

234 pv.ConfirmPostalAddress( 

235 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

236 ) 

237 

238 # Process background job 

239 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

240 mock_send.return_value = 12345 

241 while process_job(): 

242 pass 

243 

244 # Get the code 

245 with session_scope() as session: 

246 attempt = session.execute( 

247 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

248 ).scalar_one() 

249 verification_code = attempt.verification_code 

250 # Set postcard_sent_at to be past expiry 

251 attempt.postcard_sent_at = now() - POSTAL_VERIFICATION_CODE_LIFETIME - timedelta(days=1) 

252 

253 # Try to verify - should fail due to expiry 

254 with postal_verification_session(token) as pv: 

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

256 pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code=verification_code)) 

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

258 

259 

260def test_postal_verification_rate_limit(db): 

261 """Test rate limiting on postal verification attempts.""" 

262 user, token = generate_user() 

263 

264 # First attempt 

265 with postal_verification_session(token) as pv: 

266 res = pv.InitiatePostalVerification( 

267 postal_verification_pb2.InitiatePostalVerificationReq( 

268 address=postal_verification_pb2.PostalAddress( 

269 address_line_1="123 Main St", 

270 city="Test City", 

271 country_code="US", 

272 ) 

273 ) 

274 ) 

275 attempt_id = res.postal_verification_attempt_id 

276 

277 # Cancel the first attempt 

278 with postal_verification_session(token) as pv: 

279 pv.CancelPostalVerification( 

280 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=attempt_id) 

281 ) 

282 

283 # Try to initiate again immediately - should be rate limited 

284 with postal_verification_session(token) as pv: 

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

286 pv.InitiatePostalVerification( 

287 postal_verification_pb2.InitiatePostalVerificationReq( 

288 address=postal_verification_pb2.PostalAddress( 

289 address_line_1="456 Other St", 

290 city="Test City", 

291 country_code="US", 

292 ) 

293 ) 

294 ) 

295 assert e.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED 

296 

297 # Check status shows rate limit info 

298 with postal_verification_session(token) as pv: 

299 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

300 assert not status.can_initiate_new_attempt 

301 assert status.next_attempt_allowed_at.seconds > 0 

302 

303 

304def test_postal_verification_already_in_progress(db): 

305 """Test that you can't start a new attempt while one is in progress.""" 

306 user, token = generate_user() 

307 

308 # First attempt 

309 with postal_verification_session(token) as pv: 

310 pv.InitiatePostalVerification( 

311 postal_verification_pb2.InitiatePostalVerificationReq( 

312 address=postal_verification_pb2.PostalAddress( 

313 address_line_1="123 Main St", 

314 city="Test City", 

315 country_code="US", 

316 ) 

317 ) 

318 ) 

319 

320 # Try to initiate another - should fail 

321 with postal_verification_session(token) as pv: 

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

323 pv.InitiatePostalVerification( 

324 postal_verification_pb2.InitiatePostalVerificationReq( 

325 address=postal_verification_pb2.PostalAddress( 

326 address_line_1="456 Other St", 

327 city="Test City", 

328 country_code="US", 

329 ) 

330 ) 

331 ) 

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

333 

334 

335def test_postal_verification_cancel(db): 

336 """Test cancelling a postal verification attempt.""" 

337 user, token = generate_user() 

338 

339 # Initiate 

340 with postal_verification_session(token) as pv: 

341 res = pv.InitiatePostalVerification( 

342 postal_verification_pb2.InitiatePostalVerificationReq( 

343 address=postal_verification_pb2.PostalAddress( 

344 address_line_1="123 Main St", 

345 city="Test City", 

346 country_code="US", 

347 ) 

348 ) 

349 ) 

350 attempt_id = res.postal_verification_attempt_id 

351 

352 # Cancel 

353 with postal_verification_session(token) as pv: 

354 pv.CancelPostalVerification( 

355 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=attempt_id) 

356 ) 

357 

358 # Check status 

359 with postal_verification_session(token) as pv: 

360 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

361 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_CANCELLED 

362 assert not status.has_active_attempt 

363 

364 

365def test_postal_verification_can_cancel_after_postcard_sent(db): 

366 """Test that you CAN cancel after the postcard is sent (e.g., if postcard is lost).""" 

367 user, token = generate_user() 

368 

369 # Initiate and confirm 

370 with postal_verification_session(token) as pv: 

371 res = pv.InitiatePostalVerification( 

372 postal_verification_pb2.InitiatePostalVerificationReq( 

373 address=postal_verification_pb2.PostalAddress( 

374 address_line_1="123 Main St", 

375 city="Test City", 

376 country_code="US", 

377 ) 

378 ) 

379 ) 

380 attempt_id = res.postal_verification_attempt_id 

381 

382 with postal_verification_session(token) as pv: 

383 pv.ConfirmPostalAddress( 

384 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

385 ) 

386 

387 # Process background job 

388 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

389 mock_send.return_value = 12345 

390 while process_job(): 

391 pass 

392 

393 # Verify status is awaiting_verification 

394 with postal_verification_session(token) as pv: 

395 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

396 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_AWAITING_VERIFICATION 

397 

398 # Cancel - should succeed (user can cancel if postcard is lost) 

399 with postal_verification_session(token) as pv: 

400 pv.CancelPostalVerification( 

401 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=attempt_id) 

402 ) 

403 

404 # Verify status is cancelled 

405 with postal_verification_session(token) as pv: 

406 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

407 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_CANCELLED 

408 assert not status.has_active_attempt 

409 

410 

411def test_postal_verification_list_attempts(db): 

412 """Test listing postal verification attempts.""" 

413 user, token = generate_user() 

414 

415 # Create first attempt and cancel it 

416 with postal_verification_session(token) as pv: 

417 res = pv.InitiatePostalVerification( 

418 postal_verification_pb2.InitiatePostalVerificationReq( 

419 address=postal_verification_pb2.PostalAddress( 

420 address_line_1="123 Main St", 

421 city="Test City", 

422 country_code="US", 

423 ) 

424 ) 

425 ) 

426 attempt_id_1 = res.postal_verification_attempt_id 

427 

428 with postal_verification_session(token) as pv: 

429 pv.CancelPostalVerification( 

430 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=attempt_id_1) 

431 ) 

432 

433 # Move created time back to bypass rate limit 

434 with session_scope() as session: 

435 attempt = session.execute( 

436 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id_1) 

437 ).scalar_one() 

438 attempt.created = now() - POSTAL_VERIFICATION_RATE_LIMIT - timedelta(days=1) 

439 

440 # Create second attempt 

441 with postal_verification_session(token) as pv: 

442 res = pv.InitiatePostalVerification( 

443 postal_verification_pb2.InitiatePostalVerificationReq( 

444 address=postal_verification_pb2.PostalAddress( 

445 address_line_1="456 Other St", 

446 city="Other City", 

447 country_code="CA", 

448 ) 

449 ) 

450 ) 

451 attempt_id_2 = res.postal_verification_attempt_id 

452 

453 # List attempts 

454 with postal_verification_session(token) as pv: 

455 res = pv.ListPostalVerificationAttempts(postal_verification_pb2.ListPostalVerificationAttemptsReq()) 

456 assert len(res.attempts) == 2 

457 # Most recent first 

458 assert res.attempts[0].postal_verification_attempt_id == attempt_id_2 

459 assert res.attempts[1].postal_verification_attempt_id == attempt_id_1 

460 

461 

462def test_postal_verification_address_validation(db): 

463 """Test address validation errors.""" 

464 user, token = generate_user() 

465 

466 # Missing required fields 

467 with postal_verification_session(token) as pv: 

468 # Missing address_line_1 

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

470 pv.InitiatePostalVerification( 

471 postal_verification_pb2.InitiatePostalVerificationReq( 

472 address=postal_verification_pb2.PostalAddress( 

473 city="Test City", 

474 country_code="US", 

475 ) 

476 ) 

477 ) 

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

479 

480 # Missing city 

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

482 pv.InitiatePostalVerification( 

483 postal_verification_pb2.InitiatePostalVerificationReq( 

484 address=postal_verification_pb2.PostalAddress( 

485 address_line_1="123 Main St", 

486 country_code="US", 

487 ) 

488 ) 

489 ) 

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

491 

492 # Missing country 

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

494 pv.InitiatePostalVerification( 

495 postal_verification_pb2.InitiatePostalVerificationReq( 

496 address=postal_verification_pb2.PostalAddress( 

497 address_line_1="123 Main St", 

498 city="Test City", 

499 ) 

500 ) 

501 ) 

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

503 

504 

505def test_postal_verification_postcard_send_failure(db): 

506 """Test handling of postcard send failure.""" 

507 user, token = generate_user() 

508 

509 # Initiate and confirm 

510 with postal_verification_session(token) as pv: 

511 res = pv.InitiatePostalVerification( 

512 postal_verification_pb2.InitiatePostalVerificationReq( 

513 address=postal_verification_pb2.PostalAddress( 

514 address_line_1="123 Main St", 

515 city="Test City", 

516 country_code="US", 

517 ) 

518 ) 

519 ) 

520 attempt_id = res.postal_verification_attempt_id 

521 

522 with postal_verification_session(token) as pv: 

523 pv.ConfirmPostalAddress( 

524 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

525 ) 

526 

527 # Simulate postcard send failure 

528 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

529 mock_send.side_effect = Exception("API error") 

530 with pytest.raises(Exception, match="API error"): 

531 process_job() 

532 

533 # Attempt should still be in_progress (job failed, not the attempt) 

534 with postal_verification_session(token) as pv: 

535 status = pv.GetPostalVerificationStatus(postal_verification_pb2.GetPostalVerificationStatusReq()) 

536 assert status.status == postal_verification_pb2.POSTAL_VERIFICATION_STATUS_IN_PROGRESS 

537 

538 

539def test_postal_verification_code_case_insensitive(db): 

540 """Test that verification codes are case insensitive.""" 

541 user, token = generate_user() 

542 

543 # Initiate and confirm 

544 with postal_verification_session(token) as pv: 

545 res = pv.InitiatePostalVerification( 

546 postal_verification_pb2.InitiatePostalVerificationReq( 

547 address=postal_verification_pb2.PostalAddress( 

548 address_line_1="123 Main St", 

549 city="Test City", 

550 country_code="US", 

551 ) 

552 ) 

553 ) 

554 attempt_id = res.postal_verification_attempt_id 

555 

556 with postal_verification_session(token) as pv: 

557 pv.ConfirmPostalAddress( 

558 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

559 ) 

560 

561 # Process background job 

562 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

563 mock_send.return_value = 12345 

564 while process_job(): 

565 pass 

566 

567 # Get the code 

568 with session_scope() as session: 

569 attempt = session.execute( 

570 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

571 ).scalar_one() 

572 verification_code = attempt.verification_code 

573 assert verification_code 

574 

575 # Verify with lowercase code 

576 with postal_verification_session(token) as pv: 

577 res = pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code=verification_code.lower())) 

578 assert res.success 

579 

580 

581def test_postal_verification_attempt_not_found(db): 

582 """Test accessing non-existent attempts.""" 

583 user, token = generate_user() 

584 

585 with postal_verification_session(token) as pv: 

586 # Try to confirm non-existent attempt 

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

588 pv.ConfirmPostalAddress( 

589 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=999999) 

590 ) 

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

592 

593 # Try to cancel non-existent attempt 

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

595 pv.CancelPostalVerification( 

596 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=999999) 

597 ) 

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

599 

600 

601def test_postal_verification_other_user_attempt(db): 

602 """Test that users cannot access other users' attempts.""" 

603 user1, token1 = generate_user() 

604 user2, token2 = generate_user() 

605 

606 # User 1 creates an attempt 

607 with postal_verification_session(token1) as pv: 

608 res = pv.InitiatePostalVerification( 

609 postal_verification_pb2.InitiatePostalVerificationReq( 

610 address=postal_verification_pb2.PostalAddress( 

611 address_line_1="123 Main St", 

612 city="Test City", 

613 country_code="US", 

614 ) 

615 ) 

616 ) 

617 attempt_id = res.postal_verification_attempt_id 

618 

619 # User 2 tries to confirm user 1's attempt 

620 with postal_verification_session(token2) as pv: 

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

622 pv.ConfirmPostalAddress( 

623 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

624 ) 

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

626 

627 # User 2 tries to cancel user 1's attempt 

628 with postal_verification_session(token2) as pv: 

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

630 pv.CancelPostalVerification( 

631 postal_verification_pb2.CancelPostalVerificationReq(postal_verification_attempt_id=attempt_id) 

632 ) 

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

634 

635 

636def test_has_postal_verification_helper(db): 

637 """Test the has_postal_verification helper function.""" 

638 user, token = generate_user() 

639 

640 # Initially no verification 

641 with session_scope() as session: 

642 db_user = session.execute(select(User).where(User.id == user.id)).scalar_one() 

643 assert not has_postal_verification(session, db_user) 

644 

645 # Complete verification 

646 with postal_verification_session(token) as pv: 

647 res = pv.InitiatePostalVerification( 

648 postal_verification_pb2.InitiatePostalVerificationReq( 

649 address=postal_verification_pb2.PostalAddress( 

650 address_line_1="123 Main St", 

651 city="Test City", 

652 country_code="US", 

653 ) 

654 ) 

655 ) 

656 attempt_id = res.postal_verification_attempt_id 

657 

658 with postal_verification_session(token) as pv: 

659 pv.ConfirmPostalAddress( 

660 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

661 ) 

662 

663 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

664 mock_send.return_value = 12345 

665 while process_job(): 

666 pass 

667 

668 with session_scope() as session: 

669 attempt = session.execute( 

670 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

671 ).scalar_one() 

672 verification_code = attempt.verification_code 

673 

674 with postal_verification_session(token) as pv: 

675 pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code=verification_code)) 

676 

677 # Now should have verification 

678 with session_scope() as session: 

679 db_user = session.execute(select(User).where(User.id == user.id)).scalar_one() 

680 assert has_postal_verification(session, db_user) 

681 

682 

683def test_postal_verification_requires_donation(db): 

684 """Postcards cost money, so non-donors can't initiate. Mirrors phone verification.""" 

685 user, token = generate_user(last_donated=None) 

686 

687 with postal_verification_session(token) as pv: 

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

689 pv.InitiatePostalVerification( 

690 postal_verification_pb2.InitiatePostalVerificationReq( 

691 address=postal_verification_pb2.PostalAddress( 

692 address_line_1="123 Main St", 

693 city="Test City", 

694 country_code="US", 

695 ) 

696 ) 

697 ) 

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

699 assert e.value.details() == "You need to donate to Couchers.org before you can verify your address." 

700 

701 # No attempt should have been created 

702 with session_scope() as session: 

703 assert not session.execute( 

704 select(PostalVerificationAttempt).where(PostalVerificationAttempt.user_id == user.id) 

705 ).scalar_one_or_none() 

706 

707 

708def _confirmed_attempt_id(token: str) -> int: 

709 """Takes a user through to `in_progress`, i.e. ready for the postcard-sending job.""" 

710 with postal_verification_session(token) as pv: 

711 res = pv.InitiatePostalVerification( 

712 postal_verification_pb2.InitiatePostalVerificationReq( 

713 address=postal_verification_pb2.PostalAddress( 

714 address_line_1="123 Main St", 

715 city="Test City", 

716 state="CA", 

717 postal_code="12345", 

718 country_code="US", 

719 ) 

720 ) 

721 ) 

722 attempt_id: int = res.postal_verification_attempt_id 

723 

724 with postal_verification_session(token) as pv: 

725 pv.ConfirmPostalAddress( 

726 postal_verification_pb2.ConfirmPostalAddressReq(postal_verification_attempt_id=attempt_id) 

727 ) 

728 

729 return attempt_id 

730 

731 

732def test_bypass_emails_the_code_instead_of_posting(db, email_collector): 

733 """With the bypass set, no order is placed and the code is emailed instead.""" 

734 user, token = generate_user() 

735 attempt_id = _confirmed_attempt_id(token) 

736 

737 config.POSTAL_VERIFICATION_BYPASS_POST_AND_EMAIL_CODE_FOR_TESTING = True 

738 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

739 while process_job(): 

740 pass 

741 mock_send.assert_not_called() 

742 

743 with session_scope() as session: 

744 attempt = session.execute( 

745 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

746 ).scalar_one() 

747 assert attempt.status == PostalVerificationStatus.awaiting_verification 

748 assert attempt.postcard_sent_at is not None 

749 assert attempt.mypostcard_job_id is None 

750 verification_code = attempt.verification_code 

751 

752 email = email_collector.pop_for_recipient(user.email) 

753 assert "[TESTING]" in email.subject 

754 # It must be unmistakable, right at the top, that this is a testing email 

755 assert "should only be sent out in testing environments" in email.plain.split("\n\n")[0] 

756 assert "support@couchers.org" in email.plain 

757 assert verification_code in email.plain 

758 

759 assert len(email.attachments) == 1 

760 attachment = email.attachments[0] 

761 assert attachment.data[:4] == b"\x89PNG" 

762 assert 'filename="postcard.png"' in attachment.content_disposition 

763 

764 # The emailed code still works, so the whole flow can be tested 

765 with postal_verification_session(token) as pv: 

766 assert pv.VerifyPostalCode(postal_verification_pb2.VerifyPostalCodeReq(code=verification_code)).success 

767 

768 

769def test_postcard_is_posted_when_bypass_is_unset(db): 

770 """With the bypass unset, an order is placed and its job ID recorded.""" 

771 user, token = generate_user() 

772 attempt_id = _confirmed_attempt_id(token) 

773 

774 config.POSTAL_VERIFICATION_BYPASS_POST_AND_EMAIL_CODE_FOR_TESTING = False 

775 with patch("couchers.jobs.handlers.send_postcard") as mock_send: 

776 mock_send.return_value = 12345 

777 while process_job(): 

778 pass 

779 mock_send.assert_called_once() 

780 

781 with session_scope() as session: 

782 attempt = session.execute( 

783 select(PostalVerificationAttempt).where(PostalVerificationAttempt.id == attempt_id) 

784 ).scalar_one() 

785 assert attempt.status == PostalVerificationStatus.awaiting_verification 

786 assert attempt.mypostcard_job_id == 12345 

787 

788 

789def test_check_mypostcard_jobs_skipped_when_bypassing(db): 

790 """The reconciliation job must not call the API when we never placed any orders.""" 

791 config.POSTAL_VERIFICATION_BYPASS_POST_AND_EMAIL_CODE_FOR_TESTING = True 

792 with patch("couchers.jobs.handlers.get_order_ids") as mock_get_order_ids: 

793 check_mypostcard_jobs(empty_pb2.Empty()) 

794 mock_get_order_ids.assert_not_called() 

795 

796 

797def test_generate_postcard_images(): 

798 """ 

799 Generates sample postcard front and back images for visual inspection. 

800 

801 Output is written to test_artifacts/ (gitignored) and picked up by CI. 

802 """ 

803 code = "ABC123" 

804 front = get_postcard_front_image() 

805 back = _generate_back_left_side_png(code) 

806 

807 assert len(front) > 0 

808 assert len(back) > 0 

809 

810 output_path = Path(__file__).resolve().parents[2] / "test_artifacts" 

811 output_path.mkdir(parents=True, exist_ok=True) 

812 (output_path / "postcard_front.png").write_bytes(front) 

813 (output_path / "postcard_back.png").write_bytes(back)