feat: add email-based authentication
This commit is contained in:
+79
-557
@@ -2,581 +2,103 @@ from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unittest
|
||||
from datetime import datetime, timedelta
|
||||
from tempfile import TemporaryDirectory
|
||||
from urllib.parse import urlparse
|
||||
from unittest.mock import patch
|
||||
|
||||
from app import create_app
|
||||
from app.config import Config
|
||||
from app.extensions import db
|
||||
from app.models import AppSetting, PasswordResetToken, UploadRecord, User
|
||||
from app.time_utils import utc_now
|
||||
from app.models import EmailVerificationCode, TrustedDevice, User
|
||||
|
||||
|
||||
class RegistrationRoutesTest(unittest.TestCase):
|
||||
def create_test_app(
|
||||
self,
|
||||
temp_dir: str,
|
||||
*,
|
||||
allow_registration: bool,
|
||||
app_timezone: str = "Asia/Shanghai",
|
||||
):
|
||||
class EmailAuthenticationTest(unittest.TestCase):
|
||||
def create_app(self, directory: str):
|
||||
class TestConfig(Config):
|
||||
TESTING = True
|
||||
SECRET_KEY = "test-secret"
|
||||
SECRET_KEY_GENERATED = False
|
||||
SQLALCHEMY_DATABASE_URI = f"sqlite:///{temp_dir}/test.db"
|
||||
ALLOW_REGISTRATION = allow_registration
|
||||
APP_TIMEZONE = app_timezone
|
||||
SQLALCHEMY_DATABASE_URI = f"sqlite:///{directory}/test.db"
|
||||
ALLOW_REGISTRATION = True
|
||||
ADMIN_PASSWORD = None
|
||||
|
||||
ADMIN_EMAIL = ""
|
||||
RESEND_API_KEY = "test"
|
||||
RESEND_FROM_EMAIL = "no-reply@waternetwork.cn"
|
||||
SESSION_COOKIE_SECURE = False
|
||||
return create_app(TestConfig, load_model_on_start=False)
|
||||
|
||||
def csrf_token_from(self, html: bytes) -> str:
|
||||
match = re.search(rb'name="csrf_token" value="([^"]+)"', html)
|
||||
self.assertIsNotNone(match)
|
||||
return match.group(1).decode()
|
||||
def csrf(self, response) -> str:
|
||||
return re.search(rb'name="csrf_token" value="([^"]+)"', response.data).group(1).decode()
|
||||
|
||||
def create_user(self, app, username: str, password: str, *, is_admin: bool = False) -> int:
|
||||
with app.app_context():
|
||||
user = User(username=username, is_admin=is_admin)
|
||||
user.set_password(password)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return user.id
|
||||
def form(self, client, path: str, **data):
|
||||
page = client.get(path)
|
||||
data["csrf_token"] = self.csrf(page)
|
||||
return client.post(path, data=data)
|
||||
|
||||
def add_upload_records(self, app, username: str, count: int) -> None:
|
||||
with app.app_context():
|
||||
user = User.query.filter_by(username=username).one()
|
||||
base_time = datetime(2026, 1, 1, 12, 0, 0)
|
||||
for index in range(count):
|
||||
db.session.add(
|
||||
UploadRecord(
|
||||
user_id=user.id,
|
||||
original_filename=f"{username}-file-{index:02d}.xlsx",
|
||||
saved_path=f"/tmp/{username}-original-{index:02d}.xlsx",
|
||||
prediction_path=f"/tmp/{username}-prediction-{index:02d}.xlsx",
|
||||
image_path=f"/tmp/{username}-image-{index:02d}.png",
|
||||
upload_time=base_time + timedelta(minutes=index),
|
||||
)
|
||||
)
|
||||
db.session.commit()
|
||||
|
||||
def login(self, client, username: str, password: str) -> None:
|
||||
response = client.get("/login")
|
||||
token = self.csrf_token_from(response.data)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
|
||||
login_response = client.post(
|
||||
"/login",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": username,
|
||||
"password": password,
|
||||
"captcha": captcha,
|
||||
},
|
||||
)
|
||||
self.assertEqual(login_response.status_code, 302)
|
||||
|
||||
def login_attempt(self, client, username: str, password: str):
|
||||
response = client.get("/login")
|
||||
token = self.csrf_token_from(response.data)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
|
||||
return client.post(
|
||||
"/login",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": username,
|
||||
"password": password,
|
||||
"captcha": captcha,
|
||||
},
|
||||
)
|
||||
|
||||
def create_reset_link(self, app, client, user_id: int) -> str:
|
||||
token = self.csrf_token_from(client.get("/admin").data)
|
||||
response = client.post(
|
||||
f"/admin/users/{user_id}/password-reset-link",
|
||||
data={"csrf_token": token},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
return urlparse(response.get_json()["reset_url"]).path
|
||||
|
||||
def test_login_page_always_shows_register_entry_when_registration_is_closed(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
|
||||
response = app.test_client().get("/login")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn('href="/register"', response.get_data(as_text=True))
|
||||
|
||||
def test_register_page_is_visible_but_disabled_when_registration_is_closed(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
|
||||
response = app.test_client().get("/register")
|
||||
html = response.get_data(as_text=True)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("当前未开放自助注册", html)
|
||||
self.assertIn("disabled", html)
|
||||
|
||||
def test_register_page_shows_captcha(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
client = app.test_client()
|
||||
|
||||
response = client.get("/register")
|
||||
html = response.get_data(as_text=True)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn('name="captcha"', html)
|
||||
self.assertIn(captcha, html)
|
||||
|
||||
def test_template_download_uses_chinese_filename(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
|
||||
response = app.test_client().get("/download_template")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("attachment", response.headers["Content-Disposition"])
|
||||
self.assertIn(
|
||||
"filename*=UTF-8''%E7%AE%A1%E9%81%93%E9%A2%84%E6%B5%8B%E6%95%B0%E6%8D%AE%E6%A8%A1%E6%9D%BF.xlsx",
|
||||
response.headers["Content-Disposition"],
|
||||
)
|
||||
|
||||
def test_register_post_does_not_create_user_when_registration_is_closed(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/register").data)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": "new-user",
|
||||
"password": "secret123",
|
||||
"captcha": captcha,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 403)
|
||||
self.assertIn("当前未开放自助注册", response.get_data(as_text=True))
|
||||
@patch("app.routes.send_transactional_email")
|
||||
@patch("app.routes.secrets.randbelow", return_value=123456)
|
||||
def test_registration_requires_and_consumes_email_code(self, _random, _send):
|
||||
with TemporaryDirectory() as directory:
|
||||
app = self.create_app(directory); client = app.test_client()
|
||||
page = client.get("/register")
|
||||
with client.session_transaction() as state: captcha = state["captcha"]
|
||||
response = client.post("/register", data={"csrf_token": self.csrf(page), "username": "Alice", "email": "Alice@example.com", "password": "password-1234", "captcha": captcha})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
with app.app_context():
|
||||
self.assertIsNone(User.query.filter_by(username="new-user").first())
|
||||
user = User.query.filter_by(email="alice@example.com").one()
|
||||
self.assertFalse(user.is_active_account)
|
||||
self.assertEqual(EmailVerificationCode.query.count(), 1)
|
||||
verify = client.get("/verify/register")
|
||||
response = client.post("/verify/register", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
with app.app_context(): self.assertTrue(User.query.filter_by(email="alice@example.com").one().is_active_account)
|
||||
|
||||
def test_register_post_checks_captcha_before_registration_setting(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
@patch("app.routes.send_transactional_email")
|
||||
@patch("app.routes.secrets.randbelow", return_value=123456)
|
||||
def test_unknown_device_requires_email_mfa_and_creates_trusted_device(self, _random, _send):
|
||||
with TemporaryDirectory() as directory:
|
||||
app = self.create_app(directory)
|
||||
with app.app_context():
|
||||
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit()
|
||||
client = app.test_client(); page = client.get("/login")
|
||||
with client.session_transaction() as state: captcha = state["captcha"]
|
||||
response = client.post("/login", data={"csrf_token": self.csrf(page), "email": "alice@example.com", "password": "password-1234", "captcha": captcha})
|
||||
self.assertEqual(response.location, "/verify/login")
|
||||
verify = client.get("/verify/login")
|
||||
response = client.post("/verify/login", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
with app.app_context(): self.assertEqual(TrustedDevice.query.count(), 1)
|
||||
|
||||
@patch("app.routes.send_transactional_email")
|
||||
@patch("app.routes.secrets.randbelow", return_value=123456)
|
||||
def test_password_reset_revokes_trusted_devices(self, _random, _send):
|
||||
with TemporaryDirectory() as directory:
|
||||
app = self.create_app(directory)
|
||||
with app.app_context():
|
||||
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit(); db.session.add(TrustedDevice(user_id=user.id, token_hash="a" * 64, auth_version=1, expires_at=__import__('app.time_utils', fromlist=['utc_now']).utc_now())); db.session.commit()
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/register").data)
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": "new-user",
|
||||
"password": "secret123",
|
||||
"captcha": "WRONG",
|
||||
},
|
||||
)
|
||||
response = self.form(client, "/forgot-password", email="alice@example.com")
|
||||
self.assertEqual(response.status_code, 302)
|
||||
verify = client.get("/verify/reset")
|
||||
response = client.post("/verify/reset", data={"csrf_token": self.csrf(verify), "code": "123456"})
|
||||
self.assertEqual(response.location, "/set-password")
|
||||
page = client.get("/set-password")
|
||||
response = client.post("/set-password", data={"csrf_token": self.csrf(page), "password": "new-password-1234", "password_confirm": "new-password-1234"})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
with app.app_context():
|
||||
user = User.query.filter_by(email="alice@example.com").one()
|
||||
self.assertTrue(user.check_password("new-password-1234")); self.assertEqual(TrustedDevice.query.count(), 0); self.assertEqual(user.auth_version, 2)
|
||||
|
||||
@patch("app.routes.send_transactional_email")
|
||||
@patch("app.routes.secrets.randbelow", return_value=123456)
|
||||
def test_invalid_code_attempts_are_limited(self, _random, _send):
|
||||
with TemporaryDirectory() as directory:
|
||||
app = self.create_app(directory); client = app.test_client()
|
||||
with app.app_context():
|
||||
user = User(username="Alice", email="alice@example.com", is_active_account=True); user.set_password("password-1234"); db.session.add(user); db.session.commit()
|
||||
page = client.get("/login")
|
||||
with client.session_transaction() as state: captcha = state["captcha"]
|
||||
client.post("/login", data={"csrf_token": self.csrf(page), "email": "alice@example.com", "password": "password-1234", "captcha": captcha})
|
||||
for _ in range(5):
|
||||
page = client.get("/verify/login"); client.post("/verify/login", data={"csrf_token": self.csrf(page), "code": "000000"})
|
||||
page = client.get("/verify/login"); response = client.post("/verify/login", data={"csrf_token": self.csrf(page), "code": "123456"})
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn("验证码错误", response.get_data(as_text=True))
|
||||
with app.app_context():
|
||||
self.assertIsNone(User.query.filter_by(username="new-user").first())
|
||||
|
||||
def test_register_post_rejects_wrong_captcha(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/register").data)
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": "new-user",
|
||||
"password": "secret123",
|
||||
"captcha": "WRONG",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn("验证码错误", response.get_data(as_text=True))
|
||||
with app.app_context():
|
||||
self.assertIsNone(User.query.filter_by(username="new-user").first())
|
||||
|
||||
def test_register_post_creates_user_when_registration_is_open(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/register").data)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
|
||||
response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": "new-user",
|
||||
"password": "secret123",
|
||||
"captcha": captcha,
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("注册成功，请登录", response.get_data(as_text=True))
|
||||
with app.app_context():
|
||||
user = User.query.filter_by(username="new-user").first()
|
||||
self.assertIsNotNone(user)
|
||||
self.assertFalse(user.is_admin)
|
||||
|
||||
def test_admin_can_enable_registration_from_admin_panel(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
token = self.csrf_token_from(client.get("/admin").data)
|
||||
|
||||
response = client.post(
|
||||
"/admin/registration",
|
||||
data={"csrf_token": token, "allow_registration": "on"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("已开放用户自助注册", response.get_data(as_text=True))
|
||||
with app.app_context():
|
||||
self.assertTrue(AppSetting.get_bool("allow_registration"))
|
||||
|
||||
register_page = client.get("/register").get_data(as_text=True)
|
||||
self.assertNotIn("当前未开放自助注册", register_page)
|
||||
|
||||
def test_admin_can_disable_registration_from_admin_panel(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
token = self.csrf_token_from(client.get("/admin").data)
|
||||
|
||||
response = client.post(
|
||||
"/admin/registration",
|
||||
data={"csrf_token": token},
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("已关闭用户自助注册", response.get_data(as_text=True))
|
||||
with app.app_context():
|
||||
self.assertFalse(AppSetting.get_bool("allow_registration", True))
|
||||
|
||||
def test_admin_can_create_password_reset_link_for_regular_user(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
|
||||
token = self.csrf_token_from(client.get("/admin").data)
|
||||
response = client.post(
|
||||
f"/admin/users/{user_id}/password-reset-link",
|
||||
data={"csrf_token": token},
|
||||
)
|
||||
data = response.get_json()
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("/password-reset/", data["reset_url"])
|
||||
self.assertEqual(data["username"], "alice")
|
||||
with app.app_context():
|
||||
self.assertEqual(PasswordResetToken.query.count(), 1)
|
||||
|
||||
def test_admin_password_reset_section_lists_registered_users_without_uploads(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=True)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
client = app.test_client()
|
||||
|
||||
register_page = client.get("/register")
|
||||
token = self.csrf_token_from(register_page.data)
|
||||
with client.session_transaction() as session:
|
||||
captcha = session["captcha"]
|
||||
register_response = client.post(
|
||||
"/register",
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"username": "registered-user",
|
||||
"password": "secret123",
|
||||
"captcha": captcha,
|
||||
},
|
||||
)
|
||||
self.assertEqual(register_response.status_code, 200)
|
||||
|
||||
self.login(client, "admin", "secret123")
|
||||
admin_page = client.get("/admin").get_data(as_text=True)
|
||||
|
||||
self.assertIn("用户密码重置", admin_page)
|
||||
self.assertIn("registered-user", admin_page)
|
||||
self.assertIn("生成重置链接", admin_page)
|
||||
|
||||
def test_password_reset_link_requires_admin(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/login").data)
|
||||
|
||||
anonymous_response = client.post(
|
||||
f"/admin/users/{user_id}/password-reset-link",
|
||||
data={"csrf_token": token},
|
||||
)
|
||||
|
||||
self.assertEqual(anonymous_response.status_code, 302)
|
||||
|
||||
self.login(client, "alice", "oldpass")
|
||||
token = self.csrf_token_from(client.get("/home").data)
|
||||
user_response = client.post(
|
||||
f"/admin/users/{user_id}/password-reset-link",
|
||||
data={"csrf_token": token},
|
||||
)
|
||||
|
||||
self.assertEqual(user_response.status_code, 403)
|
||||
|
||||
def test_admin_cannot_create_password_reset_link_for_admin_user(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
admin_id = self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
token = self.csrf_token_from(client.get("/admin").data)
|
||||
|
||||
response = client.post(
|
||||
f"/admin/users/{admin_id}/password-reset-link",
|
||||
data={"csrf_token": token},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 403)
|
||||
self.assertIn("管理员账号", response.get_json()["error"])
|
||||
|
||||
def test_password_reset_changes_password_and_consumes_link(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
admin_client = app.test_client()
|
||||
self.login(admin_client, "admin", "secret123")
|
||||
reset_path = self.create_reset_link(app, admin_client, user_id)
|
||||
|
||||
client = app.test_client()
|
||||
reset_page = client.get(reset_path)
|
||||
token = self.csrf_token_from(reset_page.data)
|
||||
response = client.post(
|
||||
reset_path,
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"password": "newpass123",
|
||||
"password_confirm": "newpass123",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("密码已重置", response.get_data(as_text=True))
|
||||
self.assertEqual(self.login_attempt(client, "alice", "oldpass").status_code, 400)
|
||||
self.assertEqual(self.login_attempt(client, "alice", "newpass123").status_code, 302)
|
||||
self.assertEqual(client.get(reset_path).status_code, 400)
|
||||
with app.app_context():
|
||||
reset_token = PasswordResetToken.query.one()
|
||||
self.assertIsNotNone(reset_token.used_at)
|
||||
|
||||
def test_password_reset_page_displays_expiry_in_configured_timezone(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
admin_client = app.test_client()
|
||||
self.login(admin_client, "admin", "secret123")
|
||||
reset_path = self.create_reset_link(app, admin_client, user_id)
|
||||
|
||||
with app.app_context():
|
||||
reset_token = PasswordResetToken.query.one()
|
||||
reset_token.expires_at = datetime(2027, 1, 1, 0, 0, 0)
|
||||
db.session.commit()
|
||||
|
||||
response = app.test_client().get(reset_path)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("2027-01-01 08:00:00", response.get_data(as_text=True))
|
||||
|
||||
def test_expired_password_reset_link_cannot_change_password(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
admin_client = app.test_client()
|
||||
self.login(admin_client, "admin", "secret123")
|
||||
reset_path = self.create_reset_link(app, admin_client, user_id)
|
||||
|
||||
with app.app_context():
|
||||
reset_token = PasswordResetToken.query.one()
|
||||
reset_token.expires_at = utc_now() - timedelta(minutes=1)
|
||||
db.session.commit()
|
||||
|
||||
client = app.test_client()
|
||||
token = self.csrf_token_from(client.get("/login").data)
|
||||
response = client.post(
|
||||
reset_path,
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"password": "newpass123",
|
||||
"password_confirm": "newpass123",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(self.login_attempt(client, "alice", "oldpass").status_code, 302)
|
||||
|
||||
def test_new_password_reset_link_invalidates_previous_link(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
|
||||
first_path = self.create_reset_link(app, client, user_id)
|
||||
second_path = self.create_reset_link(app, client, user_id)
|
||||
|
||||
self.assertEqual(client.get(first_path).status_code, 400)
|
||||
self.assertEqual(client.get(second_path).status_code, 200)
|
||||
|
||||
def test_password_reset_validation_does_not_consume_link(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
user_id = self.create_user(app, "alice", "oldpass")
|
||||
admin_client = app.test_client()
|
||||
self.login(admin_client, "admin", "secret123")
|
||||
reset_path = self.create_reset_link(app, admin_client, user_id)
|
||||
|
||||
client = app.test_client()
|
||||
reset_page = client.get(reset_path)
|
||||
token = self.csrf_token_from(reset_page.data)
|
||||
short_response = client.post(
|
||||
reset_path,
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"password": "short",
|
||||
"password_confirm": "short",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(short_response.status_code, 400)
|
||||
|
||||
reset_page = client.get(reset_path)
|
||||
token = self.csrf_token_from(reset_page.data)
|
||||
mismatch_response = client.post(
|
||||
reset_path,
|
||||
data={
|
||||
"csrf_token": token,
|
||||
"password": "newpass123",
|
||||
"password_confirm": "different",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(mismatch_response.status_code, 400)
|
||||
self.assertEqual(client.get(reset_path).status_code, 200)
|
||||
with app.app_context():
|
||||
reset_token = PasswordResetToken.query.one()
|
||||
self.assertIsNone(reset_token.used_at)
|
||||
|
||||
def test_history_page_paginates_upload_records(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "alice", "secret123")
|
||||
self.add_upload_records(app, "alice", 12)
|
||||
client = app.test_client()
|
||||
self.login(client, "alice", "secret123")
|
||||
|
||||
first_page = client.get("/history").get_data(as_text=True)
|
||||
second_page = client.get("/history?page=2").get_data(as_text=True)
|
||||
|
||||
self.assertIn("共 12 条", first_page)
|
||||
self.assertIn("alice-file-11.xlsx", first_page)
|
||||
self.assertIn("alice-file-02.xlsx", first_page)
|
||||
self.assertNotIn("alice-file-01.xlsx", first_page)
|
||||
self.assertIn("第 <span class=\"font-bold text-textMain\">2</span> / 2 页", second_page)
|
||||
self.assertIn("alice-file-01.xlsx", second_page)
|
||||
self.assertIn("alice-file-00.xlsx", second_page)
|
||||
|
||||
def test_history_page_displays_utc_upload_time_in_configured_timezone(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(
|
||||
temp_dir,
|
||||
allow_registration=False,
|
||||
app_timezone="America/New_York",
|
||||
)
|
||||
self.create_user(app, "alice", "secret123")
|
||||
self.add_upload_records(app, "alice", 1)
|
||||
client = app.test_client()
|
||||
self.login(client, "alice", "secret123")
|
||||
|
||||
html = client.get("/history").get_data(as_text=True)
|
||||
|
||||
self.assertIn("2026-01-01 07:00:00", html)
|
||||
|
||||
def test_admin_page_paginates_upload_records(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, allow_registration=False)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
self.create_user(app, "alice", "secret123")
|
||||
self.add_upload_records(app, "alice", 12)
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
|
||||
first_page = client.get("/admin").get_data(as_text=True)
|
||||
second_page = client.get("/admin?page=2").get_data(as_text=True)
|
||||
|
||||
self.assertIn("共 12 条", first_page)
|
||||
self.assertIn("alice-file-11.xlsx", first_page)
|
||||
self.assertIn("alice-file-02.xlsx", first_page)
|
||||
self.assertNotIn("alice-file-01.xlsx", first_page)
|
||||
self.assertIn("alice-file-01.xlsx", second_page)
|
||||
self.assertIn("alice-file-00.xlsx", second_page)
|
||||
|
||||
def test_admin_page_displays_utc_times_in_configured_timezone(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(
|
||||
temp_dir,
|
||||
allow_registration=False,
|
||||
app_timezone="America/New_York",
|
||||
)
|
||||
self.create_user(app, "admin", "secret123", is_admin=True)
|
||||
self.create_user(app, "alice", "secret123")
|
||||
with app.app_context():
|
||||
alice = User.query.filter_by(username="alice").one()
|
||||
alice.created_at = datetime(2026, 1, 1, 0, 0, 0)
|
||||
db.session.commit()
|
||||
self.add_upload_records(app, "alice", 1)
|
||||
client = app.test_client()
|
||||
self.login(client, "admin", "secret123")
|
||||
|
||||
html = client.get("/admin").get_data(as_text=True)
|
||||
|
||||
self.assertIn("2025-12-31 19:00:00", html)
|
||||
self.assertIn("2026-01-01 07:00:00", html)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
|
||||
from app import create_app
|
||||
from app.config import Config
|
||||
from app.email import EmailConfigurationError, send_transactional_email
|
||||
|
||||
|
||||
class TransactionalEmailTest(unittest.TestCase):
|
||||
def create_test_app(self, temp_dir: str, *, configured: bool):
|
||||
class TestConfig(Config):
|
||||
TESTING = True
|
||||
SECRET_KEY = "test-secret"
|
||||
SECRET_KEY_GENERATED = False
|
||||
SQLALCHEMY_DATABASE_URI = f"sqlite:///{temp_dir}/test.db"
|
||||
ADMIN_PASSWORD = None
|
||||
RESEND_API_KEY = "re_test_key" if configured else ""
|
||||
RESEND_FROM_EMAIL = "no-reply@auth.example.com" if configured else ""
|
||||
|
||||
return create_app(TestConfig, load_model_on_start=False)
|
||||
|
||||
def test_send_requires_resend_configuration(self) -> None:
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, configured=False)
|
||||
with app.app_context(), self.assertRaises(EmailConfigurationError):
|
||||
send_transactional_email(
|
||||
to="user@example.com",
|
||||
subject="测试",
|
||||
html="<p>测试</p>",
|
||||
)
|
||||
|
||||
@patch("app.email.resend.Emails.send", return_value={"id": "email_123"})
|
||||
def test_send_uses_configured_sender(self, send):
|
||||
with TemporaryDirectory() as temp_dir:
|
||||
app = self.create_test_app(temp_dir, configured=True)
|
||||
with app.app_context():
|
||||
result = send_transactional_email(
|
||||
to="user@example.com",
|
||||
subject="密码重置",
|
||||
html="<p>重置链接</p>",
|
||||
)
|
||||
|
||||
self.assertEqual(result, {"id": "email_123"})
|
||||
send.assert_called_once_with(
|
||||
{
|
||||
"from": "no-reply@auth.example.com",
|
||||
"to": ["user@example.com"],
|
||||
"subject": "密码重置",
|
||||
"html": "<p>重置链接</p>",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user