feat(auth): add email verification flow

test_email_verification.py

@@ -0,0 +1,90 @@
+import re
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import app as application
+from werkzeug.security import generate_password_hash
+
+
+class EmailVerificationTestCase(unittest.TestCase):
+    def setUp(self):
+        self.temp_directory = tempfile.TemporaryDirectory()
+        self.database_path = Path(self.temp_directory.name) / "test.db"
+        self.original_config = application.app.config.copy()
+        application.app.config.update(
+            DATABASE=self.database_path,
+            MAIL_HOST="smtp.example.com",
+            PUBLIC_URL="https://example.com",
+            TESTING=True,
+            WTF_CSRF_ENABLED=False,
+        )
+        with application.app.app_context():
+            application.init_db()
+        self.client = application.app.test_client()
+
+    def tearDown(self):
+        application.app.config.update(self.original_config)
+        self.temp_directory.cleanup()
+
+    def verification_token_from_message(self, message):
+        match = re.search(r"/email/verify/([A-Za-z0-9_-]+)", message.get_content())
+        self.assertIsNotNone(match)
+        return match.group(1)
+
+    def test_registration_requires_email_verification_before_login(self):
+        with patch("app.smtplib.SMTP") as smtp:
+            response = self.client.post(
+                "/register",
+                data={"username": "newmember", "email": "new@example.com", "password": "secure-password"},
+            )
+
+        self.assertEqual(response.status_code, 302)
+        message = smtp.return_value.__enter__.return_value.send_message.call_args.args[0]
+        token = self.verification_token_from_message(message)
+        with application.app.app_context():
+            user = application.get_db().execute("SELECT id, email_verified FROM users WHERE username = ?", ("newmember",)).fetchone()
+            self.assertEqual(user["email_verified"], 0)
+            application.get_db().execute("UPDATE users SET is_approved = 1 WHERE id = ?", (user["id"],))
+            application.get_db().commit()
+
+        response = self.client.post("/login", data={"username": "newmember", "password": "secure-password"})
+        self.assertIn(b"Verify your email address before you can sign in.", response.data)
+        self.assertEqual(self.client.post(f"/email/verify/{token}").status_code, 302)
+        with application.app.app_context():
+            self.assertEqual(application.get_db().execute("SELECT email_verified FROM users WHERE id = ?", (user["id"],)).fetchone()[0], 1)
+
+    def test_email_change_keeps_current_address_until_confirmation(self):
+        with application.app.app_context():
+            db = application.get_db()
+            cursor = db.execute(
+                """INSERT INTO users (username, email, email_verified, password_hash, created_at, is_approved, role, mfa_enabled)
+                   VALUES (?, ?, 1, ?, ?, 1, 'member', 1)""",
+                ("member", "old@example.com", generate_password_hash("secure-password"), "2026-09-01T00:00:00+00:00"),
+            )
+            db.commit()
+            user_id = cursor.lastrowid
+        with self.client.session_transaction() as session:
+            session["user_id"] = user_id
+            session["mfa_verified"] = True
+
+        with patch("app.smtplib.SMTP") as smtp:
+            response = self.client.post(
+                "/account/email",
+                data={"email": "new@example.com", "current_password": "secure-password"},
+            )
+
+        self.assertEqual(response.status_code, 302)
+        message = smtp.return_value.__enter__.return_value.send_message.call_args.args[0]
+        token = self.verification_token_from_message(message)
+        with application.app.app_context():
+            self.assertEqual(application.get_db().execute("SELECT email FROM users WHERE id = ?", (user_id,)).fetchone()[0], "old@example.com")
+
+        self.assertEqual(self.client.post(f"/email/verify/{token}").status_code, 302)
+        with application.app.app_context():
+            self.assertEqual(application.get_db().execute("SELECT email FROM users WHERE id = ?", (user_id,)).fetchone()[0], "new@example.com")
+
+
+if __name__ == "__main__":
+    unittest.main()