"""Forgot-password tests. Verifies the recovery flow is powered entirely by the generic verification framework via VerificationPurpose.PASSWORD_RESET: request, generic (non- enumerating) response, OTP issue/expiry/invalid/success, password update, old password rejected / new accepted, reset-code invalidation with purpose isolation, the Security timeline update, and the full unauthenticated route flow with no auto-login. """ import os import re import unittest from datetime import timedelta os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash from app import create_app from app.auth import otp_service from app.auth.otp_service import issue_verification_code from app.auth.password_reset import ( complete_password_reset, start_password_reset, verify_password_reset, ) from app.auth.services import authenticate from app.extensions import db, mail from app.models import EmailVerification, User, VerificationPurpose from app.models.base import utcnow RESET = VerificationPurpose.PASSWORD_RESET EMAIL = VerificationPurpose.EMAIL_VERIFICATION class PasswordResetTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() self.client = self.app.test_client() def tearDown(self): db.session.remove() db.drop_all() self.ctx.pop() def _user(self, email="u@c.edu", password="Passw0rd!", *, verified=True): u = User(email=email, password_hash=generate_password_hash(password), is_verified=verified) db.session.add(u) db.session.commit() return u def _by_email(self, email): return db.session.scalar(db.select(User).filter_by(email=email)) @staticmethod def _code_from(outbox): return re.search(r"(\d{6})", outbox[0].body).group(1) # --- request + generic response ------------------------------------------- def test_forgot_password_request_issues_code(self): user = self._user() with mail.record_messages() as outbox: resp = self.client.post("/forgot-password", data={"email": "u@c.edu"}) self.assertEqual(resp.status_code, 302) self.assertIn("/reset-password/verify", resp.headers["Location"]) self.assertTrue(otp_service.has_active_code(user, RESET)) self.assertEqual(len(outbox), 1) self.assertEqual(outbox[0].subject, "Reset your Student Bazaar password") self.assertNotIn("Passw0rd", outbox[0].body) # never leak passwords def test_unknown_email_generic_response(self): with mail.record_messages() as outbox: resp = self.client.post("/forgot-password", data={"email": "nobody@c.edu"}, follow_redirects=True) self.assertEqual(resp.status_code, 200) self.assertEqual(len(outbox), 0) # no email for unknown account # Same generic confirmation as the known-account path (no enumeration). self.assertIn("sent a password reset code", resp.get_data(as_text=True)) # --- verification ---------------------------------------------------------- def test_invalid_otp_rejected(self): user = self._user() issue_verification_code(user, RESET, enforce_cooldown=False) ok, _ = verify_password_reset(user, "000000") self.assertFalse(ok) def test_expired_otp_rejected(self): user = self._user() code = issue_verification_code(user, RESET, enforce_cooldown=False) row = db.session.scalar(db.select(EmailVerification).where(EmailVerification.user_id == user.id)) row.expires_at = utcnow() - timedelta(seconds=1) db.session.commit() ok, error = verify_password_reset(user, code) self.assertFalse(ok) self.assertIn("expired", error.lower()) def test_successful_verification(self): user = self._user() code = issue_verification_code(user, RESET, enforce_cooldown=False) ok, error = verify_password_reset(user, code) self.assertTrue(ok) self.assertIsNone(error) # --- completion: password updated, old rejected, new works ---------------- def test_password_updated_old_rejected_new_works(self): user = self._user(email="pw@c.edu", password="OldPassw0rd!") code = issue_verification_code(user, RESET, enforce_cooldown=False) self.assertTrue(verify_password_reset(user, code)[0]) ok, error = complete_password_reset(user, "NewPassw0rd!") self.assertTrue(ok) self.assertIsNone(error) self.assertIsNone(authenticate("pw@c.edu", "OldPassw0rd!")) # old rejected self.assertIsNotNone(authenticate("pw@c.edu", "NewPassw0rd!")) # new works def test_weak_password_rejected(self): user = self._user() ok, error = complete_password_reset(user, "weak") self.assertFalse(ok) self.assertIn("uppercase", error.lower()) # --- invalidation + purpose isolation ------------------------------------- def test_reset_codes_invalidated_isolation_preserved(self): user = self._user() email_code = issue_verification_code(user, EMAIL, enforce_cooldown=False) reset_code = issue_verification_code(user, RESET, enforce_cooldown=False) self.assertTrue(verify_password_reset(user, reset_code)[0]) complete_password_reset(user, "NewPassw0rd!") # All PASSWORD_RESET codes cleared... self.assertFalse(otp_service.has_active_code(user, RESET)) # ...but the EMAIL_VERIFICATION code is untouched (purpose isolation). self.assertTrue(otp_service.has_active_code(user, EMAIL)) # --- timeline -------------------------------------------------------------- def test_timeline_updated(self): user = self._user(email="tl@c.edu") self.assertIsNone(user.last_password_change) code = issue_verification_code(user, RESET, enforce_cooldown=False) verify_password_reset(user, code) complete_password_reset(user, "NewPassw0rd!") self.assertIsNotNone(user.last_password_change) # The Security timeline reflects the change. self.client.post("/login", data={"email": "tl@c.edu", "password": "NewPassw0rd!"}) html = self.client.get("/security").get_data(as_text=True) self.assertIn("Password changed", html) # --- full unauthenticated route flow + no auto-login ---------------------- def test_full_route_flow_no_autologin(self): self._user(email="flow@c.edu", password="OldPassw0rd!") with mail.record_messages() as outbox: self.client.post("/forgot-password", data={"email": "flow@c.edu"}) code = self._code_from(outbox) r1 = self.client.post("/reset-password/verify", data={"code": code}) self.assertEqual(r1.status_code, 302) self.assertIn("/reset-password", r1.headers["Location"]) r2 = self.client.post("/reset-password", data={ "new_password": "NewPassw0rd!", "confirm_password": "NewPassw0rd!"}) self.assertEqual(r2.status_code, 200) self.assertIn("Password Updated", r2.get_data(as_text=True)) # Not logged in — a protected page bounces to login. follow = self.client.get("/profile") self.assertEqual(follow.status_code, 302) self.assertIn("/login", follow.headers["Location"]) # New credentials work on a fresh login. self.assertIsNotNone(authenticate("flow@c.edu", "NewPassw0rd!")) def test_reset_password_requires_verified_code(self): # Jumping straight to step 3 without verifying redirects back to step 1. resp = self.client.get("/reset-password") self.assertEqual(resp.status_code, 302) self.assertIn("/forgot-password", resp.headers["Location"]) def test_login_page_shows_forgot_link(self): html = self.client.get("/login").get_data(as_text=True) self.assertIn("/forgot-password", html) if __name__ == "__main__": unittest.main()