gsfdc/server/tests/test_member_email_auth.py

"""门户邮箱账号：注册、登录、重置密码与令牌隔离。"""

from datetime import datetime, timezone

import jwt
import pytest
from fastapi.testclient import TestClient

from app.core.config import get_settings
from app.core.security import create_access_token
from app.services.email_code import _key
from app.services.redis_client import get_redis


@pytest.fixture
def client(monkeypatch):
    monkeypatch.setenv("JWT_SECRET", "test-secret-must-be-32-characters!")
    monkeypatch.setenv("API_PAYLOAD_ENCRYPT", "false")
    monkeypatch.setenv("SMTP_HOST", "")
    get_settings.cache_clear()
    if not get_settings().database_url.strip():
        pytest.skip("DATABASE_URL 为空")
    try:
        get_redis().ping()
    except Exception:
        pytest.skip("Redis 不可达")

    def _noop_send(**kwargs):
        return None

    monkeypatch.setattr("app.services.email_code.send_email", _noop_send)
    from app.main import create_app

    with TestClient(create_app()) as api:
        yield api
    get_settings.cache_clear()


def _code(purpose: str, email: str) -> str:
    stored = get_redis().get(_key(purpose, email))
    assert stored is not None, f"missing code for {purpose}/{email}"
    if isinstance(stored, bytes):
        stored = stored.decode("utf-8")
    return stored


def _unique_email() -> str:
    import time

    return f"member-{int(time.time() * 1000)}@example.com"


def test_member_token_lasts_seven_days(client):
    settings = get_settings()
    token = create_access_token("42", "member")
    payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
    exp = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
    remaining = (exp - datetime.now(timezone.utc)).total_seconds()
    assert remaining > 6.5 * 24 * 3600
    assert remaining < 7.5 * 24 * 3600


def test_register_keeps_code_when_password_invalid(client):
    """弱密码拒收后，同一验证码仍可用于再次提交。"""
    email = _unique_email()
    sent = client.post("/api/site/auth/email-code", json={"email": email, "purpose": "register"})
    assert sent.status_code == 200, sent.text
    code = _code("register", email)

    weak = client.post(
        "/api/site/auth/register",
        json={"email": email, "code": code, "password": "short", "privacyAccepted": True},
    )
    assert weak.status_code == 400, weak.text
    assert "验证码" not in weak.json().get("detail", "")

    registered = client.post(
        "/api/site/auth/register",
        json={"email": email, "code": code, "password": "Passw0rd1", "privacyAccepted": True},
    )
    assert registered.status_code == 200, registered.text
    assert registered.json()["tokenType"] == "member"


def test_register_login_reset_and_token_isolation(client):
    email = _unique_email()
    password = "Passw0rd1"
    new_password = "Passw0rd2"

    no_privacy = client.post(
        "/api/site/auth/register",
        json={"email": email, "code": "000000", "password": password, "privacyAccepted": False},
    )
    assert no_privacy.status_code == 400
    assert "隐私政策" in no_privacy.text

    sent = client.post("/api/site/auth/email-code", json={"email": email, "purpose": "register"})
    assert sent.status_code == 200, sent.text
    code = _code("register", email)

    registered = client.post(
        "/api/site/auth/register",
        json={"email": email, "code": code, "password": password, "privacyAccepted": True},
    )
    assert registered.status_code == 200, registered.text
    token = registered.json()["accessToken"]
    assert registered.json()["tokenType"] == "member"

    me = client.get("/api/site/member/me", headers={"Authorization": f"Bearer {token}"})
    assert me.status_code == 200, me.text
    assert me.json()["email"] == email.lower()

    again = client.post("/api/site/auth/email-code", json={"email": email, "purpose": "register"})
    assert again.status_code == 400

    bad_login = client.post(
        "/api/site/auth/login",
        json={"email": email, "password": "WrongPass1"},
    )
    assert bad_login.status_code == 401

    login = client.post(
        "/api/site/auth/login",
        json={"email": email, "password": password},
    )
    assert login.status_code == 200, login.text
    login_token = login.json()["accessToken"]

    reset_sent = client.post("/api/site/auth/email-code", json={"email": email, "purpose": "reset"})
    assert reset_sent.status_code == 200, reset_sent.text
    # 上一发可能还在 60s 频控内；若失败则直接用 redis 写入
    if reset_sent.status_code == 200:
        reset_code = _code("reset", email)
    else:
        pytest.fail(reset_sent.text)

    # 若仍被频控挡住了第一次 reset 请求以外的路径，上面已成功
    reset = client.post(
        "/api/site/auth/reset-password",
        json={"email": email, "code": reset_code, "password": new_password},
    )
    assert reset.status_code == 200, reset.text

    old_login = client.post(
        "/api/site/auth/login",
        json={"email": email, "password": password},
    )
    assert old_login.status_code == 401

    new_login = client.post(
        "/api/site/auth/login",
        json={"email": email, "password": new_password},
    )
    assert new_login.status_code == 200, new_login.text

    member_on_admin = client.get(
        "/api/admin/home",
        headers={"Authorization": f"Bearer {login_token}"},
    )
    assert member_on_admin.status_code in (401, 403)

    staff_on_member = client.get(
        "/api/site/member/me",
        headers={"Authorization": f"Bearer {create_access_token('1', 'staff')}"},
    )
    assert staff_on_member.status_code in (401, 403)