"""Email verification tests (via the generic verification framework). Covers the email-verification purpose end to end: sending, invalid/expired/ successful verification, the unverified-login block, verified login, resend cooldown, hourly resend limit, the success page, and that codes are hashed at rest. Framework-level behaviour (purpose isolation, generic issue/verify) lives in test_verification_framework.py. """ import os 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 ( OtpRateLimited, issue_verification_code, seconds_until_resend, verify_code, ) from app.auth.services import mark_email_verified from app.extensions import db, mail from app.models import EmailVerification, User, VerificationPurpose from app.models.base import utcnow EMAIL = VerificationPurpose.EMAIL_VERIFICATION PENDING_KEY = "pending_verification_user_id" class EmailVerificationTests(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() # --- helpers --------------------------------------------------------------- def _user(self, email="u@c.edu", *, verified=True): u = User( email=email, password_hash=generate_password_hash("Passw0rd!"), 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)) def _rows(self, user): return db.session.scalars( db.select(EmailVerification).where(EmailVerification.user_id == user.id) ).all() def _set_pending(self, user): with self.client.session_transaction() as sess: sess[PENDING_KEY] = user.id # --- registration sends OTP but creates NO user (verify-first) ------------ def test_registration_sends_otp_without_creating_user(self): with mail.record_messages() as outbox: resp = self.client.post("/register", data={ "full_name": "New User", "email": "new@c.edu", "password": "Passw0rd!", "confirm_password": "Passw0rd!", }) self.assertEqual(resp.status_code, 302) self.assertIn("/register/verify", resp.headers["Location"]) # Verify-first: no account exists until the code is confirmed. self.assertIsNone(self._by_email("new@c.edu")) self.assertEqual(len(outbox), 1) self.assertEqual(outbox[0].subject, "Verify your Student Bazaar account") self.assertIn("verification code", outbox[0].body.lower()) self.assertNotIn("Passw0rd", outbox[0].body) # never leak the password # --- OTP hashed, single table, purpose recorded --------------------------- def test_otp_is_hashed_not_plaintext(self): user = self._user(verified=False) code = issue_verification_code(user, EMAIL, enforce_cooldown=False) row = self._rows(user)[-1] self.assertNotEqual(row.verification_token, code) # not plaintext self.assertEqual(len(row.verification_token), 64) # HMAC-SHA256 hex self.assertEqual(row.purpose, EMAIL) # stored purpose self.assertIsNone(row.verified_at) # unredeemed # --- invalid OTP rejected -------------------------------------------------- def test_invalid_otp_rejected(self): user = self._user(verified=False) issue_verification_code(user, EMAIL, enforce_cooldown=False) ok, error = verify_code(user, EMAIL, "000000") self.assertFalse(ok) self.assertIn("Incorrect", error) self.assertFalse(user.is_verified) # --- expired OTP rejected -------------------------------------------------- def test_expired_otp_rejected(self): user = self._user(verified=False) code = issue_verification_code(user, EMAIL, enforce_cooldown=False) row = self._rows(user)[-1] row.expires_at = utcnow() - timedelta(seconds=1) db.session.commit() ok, error = verify_code(user, EMAIL, code) self.assertFalse(ok) self.assertIn("expired", error.lower()) self.assertFalse(user.is_verified) # --- successful verification (framework consumes; feature applies) -------- def test_successful_verification(self): user = self._user(verified=False) code = issue_verification_code(user, EMAIL, enforce_cooldown=False) ok, error = verify_code(user, EMAIL, code) self.assertTrue(ok) self.assertIsNone(error) # single-use: the same code can't be redeemed again ok2, _ = verify_code(user, EMAIL, code) self.assertFalse(ok2) # feature layer applies the email-verification consequence mark_email_verified(user) self.assertTrue(user.is_verified) self.assertIsNotNone(user.verified_at) # --- successful verification shows success page + auto-login (route) ------ def test_verify_route_shows_success_and_logs_in(self): user = self._user("new@c.edu", verified=False) code = issue_verification_code(user, EMAIL, enforce_cooldown=False) self._set_pending(user) resp = self.client.post("/verify-email", data={"code": code}) self.assertEqual(resp.status_code, 200) # success page, not redirect self.assertIn("Email Verified", resp.get_data(as_text=True)) self.assertTrue(self._by_email("new@c.edu").is_verified) # session is now authenticated → a login-required page is reachable self.assertEqual(self.client.get("/profile").status_code, 200) # --- resend invalidates the previous code --------------------------------- def test_resend_invalidates_previous_code(self): user = self._user(verified=False) first = issue_verification_code(user, EMAIL, enforce_cooldown=False) second = issue_verification_code(user, EMAIL, enforce_cooldown=False) self.assertNotEqual(first, second) ok_old, _ = verify_code(user, EMAIL, first) self.assertFalse(ok_old) # previous code no longer valid ok_new, _ = verify_code(user, EMAIL, second) self.assertTrue(ok_new) # --- login blocked when unverified ---------------------------------------- def test_login_blocked_when_unverified(self): self._user("pending@c.edu", verified=False) resp = self.client.post("/login", data={ "email": "pending@c.edu", "password": "Passw0rd!"}) self.assertEqual(resp.status_code, 302) self.assertIn("/verify-email", resp.headers["Location"]) follow = self.client.get("/profile") self.assertEqual(follow.status_code, 302) self.assertIn("/login", follow.headers["Location"]) # --- verified user can log in --------------------------------------------- def test_verified_user_can_log_in(self): self._user("ok@c.edu", verified=True) resp = self.client.post("/login", data={ "email": "ok@c.edu", "password": "Passw0rd!"}) self.assertEqual(resp.status_code, 302) self.assertNotIn("/verify-email", resp.headers["Location"]) self.assertEqual(self.client.get("/profile").status_code, 200) # --- resend cooldown ------------------------------------------------------- def test_resend_cooldown(self): user = self._user(verified=False) issue_verification_code(user, EMAIL, enforce_cooldown=False) self.assertGreater(seconds_until_resend(user, EMAIL), 0) with self.assertRaises(OtpRateLimited): issue_verification_code(user, EMAIL) # cooldown enforced # --- hourly resend limit --------------------------------------------------- def test_resend_hourly_limit(self): user = self._user(verified=False) limit = self.app.config["OTP_MAX_PER_HOUR"] for _ in range(limit): issue_verification_code(user, EMAIL, enforce_cooldown=False) with self.assertRaises(OtpRateLimited): issue_verification_code(user, EMAIL, enforce_cooldown=False) self.assertEqual(otp_service.requests_last_hour(user, EMAIL), limit) # --- too many wrong attempts disables the code (route) -------------------- def test_verify_attempt_cap_invalidates_code(self): user = self._user(verified=False) issue_verification_code(user, EMAIL, enforce_cooldown=False) self._set_pending(user) max_attempts = self.app.config["OTP_MAX_VERIFY_ATTEMPTS"] for _ in range(max_attempts): self.client.post("/verify-email", data={"code": "111111"}) self.assertFalse(otp_service.has_active_code(user, EMAIL)) # --- change email before verifying (route) -------------------------------- def test_change_email_before_verifying(self): user = self._user("typo@c.edu", verified=False) issue_verification_code(user, EMAIL, enforce_cooldown=False) self._set_pending(user) resp = self.client.post("/verify-email/change", data={"email": "fixed@c.edu"}) self.assertEqual(resp.status_code, 302) db.session.refresh(user) self.assertEqual(user.email, "fixed@c.edu") self.assertTrue(otp_service.has_active_code(user, EMAIL)) # --- verify page renders with a masked email ------------------------------ def test_verify_page_renders_masked_email(self): user = self._user("student@campus.edu", verified=False) issue_verification_code(user, EMAIL, enforce_cooldown=False) self._set_pending(user) html = self.client.get("/verify-email").get_data(as_text=True) self.assertIn("Verify your email", html) self.assertIn("s*****t@campus.edu", html) # --- email_verified property mirrors is_verified -------------------------- def test_email_verified_property_alias(self): user = self._user(verified=False) self.assertFalse(user.email_verified) user.email_verified = True self.assertTrue(user.is_verified) if __name__ == "__main__": unittest.main()