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_verify_set_password_embeds_enabled_org_options(b2_client, portal_mock):
    # 匹配作战部 → 机构下拉仅含启用机构（不含停用部门），预选推断结果
    data = _verify(
        b2_client, "mock:9010:portal_orgs:机构列表用户:作战部"
    ).get_json()["data"]
    assert data["phase"] == "SET_PASSWORD"
    orgs = data["organizations"]
    assert {o["orgName"] for o in orgs} == {"机关", "作战部", "通信部"}
    assert all(
        set(o) == {"id", "orgCode", "orgName", "orgPath"} for o in orgs
    )
    assert all(isinstance(o["id"], str) for o in orgs)
    # 按 org_path 稳定排序：机关 → 作战部 → 通信部
    assert [o["orgName"] for o in orgs] == ["机关", "作战部", "通信部"]
    ops = next(o for o in orgs if o["orgName"] == "作战部")
    assert data["suggestedOrganizationId"] == ops["id"]


def test_verify_set_password_suggested_null_without_org(b2_client, portal_mock):
    # 无匹配机构且未配置默认 → 预选为空，机构列表仍内嵌
    data = _verify(
        b2_client, "mock:9011:portal_noorg2:无预选用户:不存在部门"
    ).get_json()["data"]
    assert data["phase"] == "SET_PASSWORD"
    assert data["suggestedOrganizationId"] is None
    assert len(data["organizations"]) == 3


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


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
    # 契约1.29：设密完成响应同登录下发空闲自动登出分钟数
    assert payload["idleLogoutMinutes"] == 5
    # 契约1.38：一次一码轮换回显（Python 版不实现轮换，恒 False）
    assert payload["tokenRotation"] is False
    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_setup_password_updates_organization_when_provided(
    b2_app, b2_client, portal_mock
):
    # 用户推断挂作战部，设密时自选通信部 → 机构改为通信部并记审计
    data = _verify(
        b2_client, "mock:9012:portal_pickorg:自选机构用户:作战部"
    ).get_json()["data"]
    comms = next(
        o for o in data["organizations"] if o["orgName"] == "通信部"
    )
    response = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9012",
            "keepSignedIn": False,
            "organizationId": comms["id"],
        },
    )
    assert response.status_code == 200
    assert response.get_json()["data"]["user"]["organizationName"] == "通信部"

    user = _find_user(b2_app, "portal_pickorg")
    assert user.organization_id == int(comms["id"])
    assert user.organization_name == "通信部"
    with b2_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(
                AuditLog.username == "portal_pickorg",
                AuditLog.action_type == "LOGIN",
            )
        )
        assert audit is not None
        assert audit.detail["organizationId"] == comms["id"]


def test_setup_password_keeps_inferred_org_when_omitted(
    b2_app, b2_client, portal_mock
):
    # 缺省（旧客户端）不改变用户机构
    _verify(b2_client, "mock:9013:portal_keeporg:缺省机构用户:作战部")
    data = _verify(
        b2_client, "mock:9013:portal_keeporg:缺省机构用户:作战部"
    ).get_json()["data"]
    response = b2_client.post(
        SETUP_PATH,
        json={
            "portalTicket": data["portalTicket"],
            "newPassword": "Portal#9013",
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 200
    user = _find_user(b2_app, "portal_keeporg")
    assert user.organization_name == "作战部"


def test_setup_password_rejects_invalid_organization(
    b2_app, b2_client, portal_mock
):
    data = _verify(
        b2_client, "mock:9014:portal_badorg:非法机构用户:作战部"
    ).get_json()["data"]

    def _setup(org_id) -> object:
        return b2_client.post(
            SETUP_PATH,
            json={
                "portalTicket": data["portalTicket"],
                "newPassword": "Portal#9014",
                "keepSignedIn": False,
                "organizationId": org_id,
            },
        )

    with b2_app.app_context():
        disabled = db.session.scalar(
            db.select(Organization).where(
                Organization.org_code == "ORG_DISABLED"
            )
        )
        assert disabled is not None
        disabled_id = str(disabled.id)

    # 停用机构拒绝
    disabled_response = _setup(disabled_id)
    assert disabled_response.status_code == 400
    assert (
        disabled_response.get_json()["code"] == "INVALID_ARGUMENT"
    )
    # 不存在的机构拒绝
    missing = _setup("999999")
    assert missing.status_code == 400
    assert missing.get_json()["code"] == "INVALID_ARGUMENT"
    # 非字符串拒绝
    non_string = _setup(123)
    assert non_string.status_code == 400

    # 上述拒绝均不烧毁票据，改用合法机构仍可设密
    comms = next(
        o for o in data["organizations"] if o["orgName"] == "通信部"
    )
    ok = _setup(comms["id"])
    assert ok.status_code == 200


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