124 lines
6.2 KiB
Python
124 lines
6.2 KiB
Python
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", "real_name": "New Member", "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")
|
|
|
|
def test_profile_name_is_used_for_byline_with_handle_fallback(self):
|
|
with application.app.app_context():
|
|
db = application.get_db()
|
|
named_author = db.execute(
|
|
"""INSERT INTO users (username, email, email_verified, password_hash, created_at, is_approved, role, mfa_enabled)
|
|
VALUES (?, ?, 1, ?, ?, 1, 'member', 1)""",
|
|
("named", "named@example.com", generate_password_hash("secure-password"), "2026-09-01T00:00:00+00:00"),
|
|
).lastrowid
|
|
legacy_author = db.execute(
|
|
"""INSERT INTO users (username, email, email_verified, password_hash, created_at, is_approved, role, mfa_enabled)
|
|
VALUES (?, ?, 1, ?, ?, 1, 'member', 1)""",
|
|
("legacy", "legacy@example.com", generate_password_hash("secure-password"), "2026-09-01T00:00:00+00:00"),
|
|
).lastrowid
|
|
db.execute(
|
|
"""INSERT INTO posts (title, slug, excerpt, body, category, author_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
("Named post", "named-post", "Named excerpt", "Named body", "Code", named_author, "2026-09-01T00:00:00+00:00", "2026-09-01T00:00:00+00:00"),
|
|
)
|
|
db.execute(
|
|
"""INSERT INTO posts (title, slug, excerpt, body, category, author_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
("Legacy post", "legacy-post", "Legacy excerpt", "Legacy body", "Code", legacy_author, "2026-09-01T00:00:00+00:00", "2026-09-01T00:00:00+00:00"),
|
|
)
|
|
db.commit()
|
|
with self.client.session_transaction() as session:
|
|
session["user_id"] = named_author
|
|
session["mfa_verified"] = True
|
|
|
|
response = self.client.post("/account/profile", data={"real_name": "Ada Lovelace"})
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn(b"BY Ada Lovelace", self.client.get("/").data)
|
|
self.assertIn(b"WRITTEN BY Ada Lovelace", self.client.get("/post/named-post").data)
|
|
self.assertIn(b"BY legacy", self.client.get("/").data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |