"""Verify-first registration tests. Covers the overhaul: no user is created before verification, the account is created (and auto-logged-in) only after a correct OTP, session/code expiry, resend + cooldown, change-email, duplicate-email handling, send-failure rollback, temporary-registration cleanup, and branded HTML + plain-text email rendering. All OTP concerns are delegated to the shared framework. """ 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, registration from app.auth.services import authenticate from app.extensions import db, mail from app.models import TemporaryRegistration, User, VerificationPurpose from app.models.base import utcnow EMAIL = VerificationPurpose.EMAIL_VERIFICATION RESET = VerificationPurpose.PASSWORD_RESET class RegistrationFlowTests(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 _start(self, name="New User", email="new@c.edu", password="Passw0rd!"): with mail.record_messages() as outbox: temp, err = registration.start_registration(name, email, password) code = None if outbox: code = re.search(r"(\d{6})", outbox[0].body).group(1) return temp, err, code, outbox def _verified_user(self, email): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True) 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 _user_count(self): return db.session.scalar(db.select(db.func.count(User.id))) or 0 def _temp_count(self): return db.session.scalar(db.select(db.func.count(TemporaryRegistration.id))) or 0 # --- no user before verification ------------------------------------------ def test_no_user_created_before_verification(self): temp, err, code, outbox = self._start() self.assertIsNotNone(temp) self.assertIsNone(err) self.assertEqual(self._user_count(), 0) # zero users self.assertEqual(self._temp_count(), 1) # one pending registration self.assertEqual(len(outbox), 1) # --- user created only after successful verification ---------------------- def test_user_created_after_verification(self): temp, _, code, _ = self._start() user, err = registration.verify_registration(temp, code) self.assertIsNotNone(user) self.assertIsNone(err) self.assertTrue(user.is_verified) self.assertIsNotNone(user.verified_at) self.assertEqual(user.profile.full_name, "New User") self.assertEqual(self._user_count(), 1) self.assertEqual(self._temp_count(), 0) # pending row deleted self.assertIsNotNone(authenticate("new@c.edu", "Passw0rd!")) def test_invalid_code_rejected_no_user(self): temp, _, _, _ = self._start() user, err = registration.verify_registration(temp, "000000") self.assertIsNone(user) self.assertEqual(err, registration.ERR_BAD_CODE) self.assertEqual(self._user_count(), 0) # --- expiry ---------------------------------------------------------------- def test_expired_session_rejected_and_deleted(self): temp, _, code, _ = self._start() temp.expires_at = utcnow() - timedelta(seconds=1) db.session.commit() user, err = registration.verify_registration(temp, code) self.assertIsNone(user) self.assertEqual(err, registration.ERR_EXPIRED) self.assertEqual(self._temp_count(), 0) def test_expired_code_but_live_session(self): temp, _, code, _ = self._start() temp.code_expires_at = utcnow() - timedelta(seconds=1) db.session.commit() user, err = registration.verify_registration(temp, code) self.assertIsNone(user) self.assertEqual(err, registration.ERR_CODE_EXPIRED) # --- resend + cooldown ----------------------------------------------------- def test_resend_sends_new_code(self): temp, _, first, _ = self._start() temp.last_sent_at = utcnow() - timedelta(seconds=120) # bypass cooldown db.session.commit() with mail.record_messages() as ob: ok, err = registration.resend_code(temp) self.assertTrue(ok) second = re.search(r"(\d{6})", ob[0].body).group(1) self.assertNotEqual(first, second) # old code no longer valid, new one works self.assertIsNone(registration.verify_registration(temp, first)[0]) self.assertIsNotNone(registration.verify_registration(temp, second)[0]) def test_resend_cooldown(self): temp, _, _, _ = self._start() ok, err = registration.resend_code(temp) # just sent → cooldown active self.assertFalse(ok) self.assertEqual(err, registration.ERR_RATE_LIMITED) self.assertGreater(registration.seconds_until_resend(temp), 0) # --- change email ---------------------------------------------------------- def test_change_email(self): temp, _, _, _ = self._start(email="typo@c.edu") with mail.record_messages() as ob: ok, err = registration.change_email(temp, "fixed@c.edu") self.assertTrue(ok) self.assertEqual(temp.email, "fixed@c.edu") code = re.search(r"(\d{6})", ob[0].body).group(1) user, verr = registration.verify_registration(temp, code) self.assertIsNotNone(user) self.assertEqual(user.email, "fixed@c.edu") # --- duplicate email ------------------------------------------------------- def test_duplicate_verified_email_blocked(self): self._verified_user("taken@c.edu") temp, err, _, _ = self._start(email="taken@c.edu") self.assertIsNone(temp) self.assertEqual(err, registration.ERR_EXISTS) def test_expired_temp_registration_allows_new_registration(self): temp, _, _, _ = self._start(email="again@c.edu") temp.expires_at = utcnow() - timedelta(minutes=1) # stale pending row db.session.commit() temp2, err2, _, _ = self._start(email="again@c.edu") # cleanup + allow self.assertIsNotNone(temp2) self.assertIsNone(err2) # --- send-failure rollback ------------------------------------------------- def test_failed_email_delivery_rolls_back(self): original = otp_service.send_verification_email otp_service.send_verification_email = lambda *a, **k: False try: temp, err = registration.start_registration("X", "fail@c.edu", "Passw0rd!") finally: otp_service.send_verification_email = original self.assertIsNone(temp) self.assertEqual(err, registration.ERR_SEND_FAILED) self.assertEqual(self._temp_count(), 0) # rolled back, nothing left self.assertEqual(self._user_count(), 0) # --- cleanup --------------------------------------------------------------- def test_cleanup_expired(self): temp, _, _, _ = self._start(email="old@c.edu") temp.expires_at = utcnow() - timedelta(minutes=1) db.session.commit() self.assertEqual(registration.cleanup_expired(), 1) self.assertEqual(self._temp_count(), 0) # --- attempt lock ---------------------------------------------------------- def test_too_many_attempts_locks_code(self): temp, _, _, _ = self._start() max_attempts = self.app.config["OTP_MAX_VERIFY_ATTEMPTS"] err = None for _ in range(max_attempts + 1): _, err = registration.verify_registration(temp, "000000") self.assertEqual(err, registration.ERR_LOCKED) # --- email rendering (HTML + plain text) ---------------------------------- def test_html_and_text_email_render(self): subject, text, html = otp_service.render_verification_email(EMAIL, "246813", 5) self.assertEqual(subject, "Verify your Student Bazaar account") self.assertIn("246813", text) self.assertIn("246813", html) self.assertIn("<", html) # HTML markup present self.assertNotIn("<", text) # plain-text alternative has no tags self.assertIn("Student Bazaar", html) def test_password_reset_email_uses_its_own_template(self): subject, text, html = otp_service.render_verification_email(RESET, "111111", 5) self.assertEqual(subject, "Reset your Student Bazaar password") self.assertIn("password", html.lower()) self.assertIn("111111", html) def test_email_is_multipart_with_reply_to(self): with mail.record_messages() as outbox: registration.start_registration("MP", "mp@c.edu", "Passw0rd!") msg = outbox[0] self.assertTrue(msg.body) # plain text self.assertTrue(msg.html) # HTML alternative self.assertEqual(msg.reply_to, self.app.config["MAIL_REPLY_TO"]) # --- full route flow ------------------------------------------------------- def test_route_flow_creates_user_only_after_verify_and_autologs_in(self): with mail.record_messages() as ob: r = self.client.post("/register", data={ "full_name": "Route User", "email": "route@c.edu", "password": "Passw0rd!", "confirm_password": "Passw0rd!"}) self.assertEqual(r.status_code, 302) self.assertIn("/register/verify", r.headers["Location"]) self.assertIsNone(self._by_email("route@c.edu")) # no user yet code = re.search(r"(\d{6})", ob[0].body).group(1) r2 = self.client.post("/register/verify", data={"code": code}) self.assertEqual(r2.status_code, 200) self.assertIn("Email Verified", r2.get_data(as_text=True)) user = self._by_email("route@c.edu") self.assertIsNotNone(user) self.assertTrue(user.is_verified) # auto-logged-in → protected page reachable self.assertEqual(self.client.get("/profile").status_code, 200) def test_route_duplicate_email_shows_generic_exists(self): self._verified_user("dupe@c.edu") r = self.client.post("/register", data={ "full_name": "Dupe", "email": "dupe@c.edu", "password": "Passw0rd!", "confirm_password": "Passw0rd!"}, follow_redirects=True) self.assertIn("already exists", r.get_data(as_text=True)) self.assertEqual(self._temp_count(), 0) if __name__ == "__main__": unittest.main()