document-management/backend/tests/dms/test_portal_auth.py

"""门户跳转按需登录（mock provider）全分支测试。"""

from __future__ import annotations

from sqlalchemy import delete

import pytest

from dms.extensions import db
from dms.models import AuditLog, IpEntry, OrgIpAssignment, Organization, User


VERIFY_PATH = "/api/v1/auth/portal/verify"
SETUP_PATH = "/api/v1/auth/portal/setup-password"


@pytest.fixture()
def portal_mock(monkeypatch):
    monkeypatch.setenv("DMS_PORTAL_PROVIDER", "mock")


@pytest.fixture(autouse=True)
def _clean_ip_tables(b2_app):
    """清空IP两表，保证setup-password的verify_login_ip恒走无分配放行分支。"""
    with b2_app.app_context():
        db.session.execute(delete(OrgIpAssignment))
        db.session.execute(delete(IpEntry))
        db.session.commit()


def _verify(client, token: str):
    return client.post(VERIFY_PATH, json={"query": f"token={token}"})


def _find_user(app, username: str) -> User:
    with app.app_context():
        user = db.session.scalar(
            db.select(User).where(User.username == username)
        )
        assert user is not None
        return user


# ---------------------------------------------------------------- 开关与入参


def test_verify_rejected_when_portal_disabled(b2_client, monkeypatch):
    monkeypatch.delenv("DMS_PORTAL_PROVIDER", raising=False)
    response = b2_client.post(
        VERIFY_PATH, json={"query": "token=mock:1:a:b"}
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_verify_rejects_payload_with_extra_fields(b2_client, portal_mock):
    response = b2_client.post(
        VERIFY_PATH, json={"query": "token=mock:1:a:b", "extra": 1}
    )
    assert response.status_code == 400


def test_verify_rejects_missing_token(b2_client, portal_mock):
    response = b2_client.post(VERIFY_PATH, json={"query": "foo=1&bar=2"})
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_verify_rejects_invalid_portal_token(b2_client, portal_mock):
    response = _verify(b2_client, "not-a-mock-token")
    assert response.status_code == 401
    assert response.get_json()["code"] == "PORTAL_TOKEN_INVALID"


# ---------------------------------------------------------------- 新用户落库


def test_verify_provisions_new_user_and_issues_setup_ticket(
    b2_app, b2_client, portal_mock
):
    response = _verify(b2_client, "mock:9001:portal_first:门户新用户:作战部")
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["phase"] == "SET_PASSWORD"
    assert data["username"] == "portal_first"
    assert data["realName"] == "门户新用户"
    assert data["portalTicket"]

    user = _find_user(b2_app, "portal_first")
    assert user.external_id == "9001"
    assert user.organization_name == "作战部"
    assert user.last_login_at is None
    assert user.status == "ENABLED"
    assert user.role_code == "USER"
    with b2_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(
                AuditLog.username == "portal_first",
                AuditLog.action_type == "CREATE_USER",
            )
        )
        assert audit is not None


def test_provisioned_user_cannot_login_before_setup(
    b2_client, b2_password, portal_mock
):
    # 占位密码随机不可知：设密前任何密码都无法登录
    _verify(b2_client, "mock:9002:portal_locked:未设密用户")
    response = b2_client.post(
        "/api/v1/auth/login",
        json={
            "username": "portal_locked",
            "password": b2_password,
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 401
    assert response.get_json()["code"] == "INVALID_CREDENTIALS"


def test_organization_falls_back_to_none_when_no_match(
    b2_app, b2_client, portal_mock
):
    _verify(b2_client, "mock:9003:portal_noorg:无组织用户:不存在部门")
    user = _find_user(b2_app, "portal_noorg")
    assert user.organization_id is None
    assert user.organization_name is None


def test_organization_falls_back_to_configured_default(
    b2_app, b2_client, portal_mock, monkeypatch
):
    with b2_app.app_context():
        root = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert root is not None
        monkeypatch.setenv("DMS_PORTAL_DEFAULT_ORG_ID", str(root.id))
    _verify(b2_client, "mock:9004:portal_deforg:默认组织用户:不存在部门")
    user = _find_user(b2_app, "portal_deforg")
    assert user.organization_name == "机关"


# ---------------------------------------------------------------- 既有用户匹配


def test_existing_username_matched_and_backfilled(b2_app, b2_client, portal_mock):
    # b2_app预置username="user"（作战部，从未登录）→ 按用户名匹配并同步展示信息
    response = _verify(b2_client, "mock:8001:user:门户同步姓名:通信部")
    data = response.get_json()["data"]
    assert data["phase"] == "SET_PASSWORD"
    assert data["username"] == "user"

    user = _find_user(b2_app, "user")
    assert user.external_id == "8001"
    assert user.real_name == "门户同步姓名"
    assert user.organization_name == "通信部"
    with b2_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(
                AuditLog.username == "user",
                AuditLog.action_type == "EDIT_USER",
            )
        )
        assert audit is not None


def test_disabled_user_rejected(b2_client, portal_mock):
    # b2_app预置username="disabled"（DISABLED）
    response = _verify(b2_client, "mock:8002:disabled:停用用户")
    assert response.status_code == 401
    assert response.get_json()["code"] == "USER_DISABLED"


# ---------------------------------------------------------------- 设密与登录闭环


def test_setup_password_completes_login_and_enables_normal_login(
    b2_client, portal_mock
):
    data = _verify(
        b2_client, "mock:9005:portal_setup:门户设密用户"
    ).get_json()["data"]
    response = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9005",
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 200
    payload = response.get_json()["data"]
    assert payload["tokenType"] == "Bearer"
    assert payload["expiresIn"] == 7200
    assert payload["user"]["username"] == "portal_setup"

    # 独立密码设好后可走普通用户名密码登录
    login = b2_client.post(
        "/api/v1/auth/login",
        json={
            "username": "portal_setup",
            "password": "Portal#9005",
            "keepSignedIn": False,
        },
    )
    assert login.status_code == 200

    # 票据一次性：设密后auth_version已变更，复用被拒
    replay = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9095",
            "keepSignedIn": False,
        },
    )
    assert replay.status_code == 401
    assert replay.get_json()["code"] == "PORTAL_TICKET_INVALID"


def test_setup_password_keep_signed_in_extends_expiry(b2_client, portal_mock):
    data = _verify(
        b2_client, "mock:9006:portal_keep:保持登录用户"
    ).get_json()["data"]
    response = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9006",
            "keepSignedIn": True,
        },
    )
    assert response.status_code == 200
    assert response.get_json()["data"]["expiresIn"] == 604800


def test_setup_password_rejects_weak_password_without_burning_ticket(
    b2_client, portal_mock
):
    data = _verify(
        b2_client, "mock:9007:portal_weak:弱密码用户"
    ).get_json()["data"]
    weak = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "short",
            "keepSignedIn": False,
        },
    )
    assert weak.status_code == 400
    assert weak.get_json()["code"] == "INVALID_ARGUMENT"

    # 弱密码校验先于票据消费，票据未烧毁仍可正常设密
    retry = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9007",
            "keepSignedIn": False,
        },
    )
    assert retry.status_code == 200


def test_setup_password_rejects_garbage_ticket(b2_client, portal_mock):
    response = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": "not-a-jwt",
            "newPassword": "Portal#9000",
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 401
    assert response.get_json()["code"] == "PORTAL_TICKET_INVALID"


def test_second_launch_returns_login_phase(b2_client, portal_mock):
    token = "mock:9008:portal_again:二次跳转用户"
    first = _verify(b2_client, token).get_json()["data"]
    assert first["phase"] == "SET_PASSWORD"

    setup = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": first["portalTicket"],
            "newPassword": "Portal#9008",
            "keepSignedIn": False,
        },
    )
    assert setup.status_code == 200

    # 设密登录后（last_login_at非空），再次跳转直接进入独立密码登录态
    second = _verify(b2_client, token).get_json()["data"]
    assert second["phase"] == "LOGIN"
    assert second["username"] == "portal_again"
    assert "portalTicket" not in second