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

"""B1测试公共配置。"""

from __future__ import annotations

import sys
import os
import tempfile
from pathlib import Path

import pytest


BACKEND_ROOT = Path(__file__).resolve().parents[2]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(BACKEND_ROOT))


@pytest.fixture(scope="session")
def flask_app():
    from app import app

    app.config.update(TESTING=True)
    # 测试环境关闭登录验证码（安全需求4）：验证码逻辑由 test_captcha.py 专项覆盖
    app.config["DMS_CAPTCHA_ENABLED"] = False
    return app


@pytest.fixture()
def client(flask_app):
    return flask_app.test_client()


@pytest.fixture()
def b2_app():
    database_url = os.environ.get("DMS_TEST_DATABASE_URL")
    jwt_secret = os.environ.get("DMS_TEST_JWT_SECRET")
    password = os.environ.get("DMS_TEST_USER_PASSWORD")
    if not database_url or not database_url.rsplit("/", 1)[-1].split("?", 1)[0].endswith("_test"):
        pytest.skip("DMS_TEST_DATABASE_URL必须指向以_test结尾的独立测试库")
    if not jwt_secret or not password:
        pytest.skip("缺少B2测试专用密钥或密码环境变量")

    from flask import Flask
    from flask_cors import CORS
    from sqlalchemy import delete, update

    from dms import init_dms
    from dms.extensions import db
    from dms.models import (
        AttachmentBinding,
        AuditLog,
        Category,
        Document,
        DocumentRelation,
        Organization,
        Permission,
        Role,
        RolePermission,
        SubPlanBinding,
        User,
    )
    from dms.security.passwords import hash_password

    test_storage = tempfile.TemporaryDirectory(prefix="dms-test-storage-")
    os.environ["DMS_DATABASE_URL"] = database_url
    os.environ["DMS_JWT_SECRET"] = jwt_secret
    os.environ["DMS_STORAGE_ROOT"] = test_storage.name
    app = Flask("dms-b2-test")
    CORS(app)
    init_dms(app)
    app.config.update(TESTING=True)
    # 测试环境关闭登录验证码（安全需求4），登录回归三键请求体
    app.config["DMS_CAPTCHA_ENABLED"] = False

    with app.app_context():
        db.session.execute(delete(AuditLog))
        # 角色两表：先解除用户 role_id 引用，避免外键残留（角色测试
        # 结束后 role_cleanup 无 teardown，残行会卡死机构清理）
        db.session.execute(update(User).values(role_id=None))
        db.session.execute(delete(RolePermission))
        db.session.execute(delete(Role))
        db.session.execute(delete(SubPlanBinding))
        db.session.execute(delete(DocumentRelation))
        db.session.execute(delete(Permission))
        db.session.execute(delete(AttachmentBinding))
        db.session.execute(
            delete(Document).where(Document.document_type == "SUB_PLAN")
        )
        db.session.execute(delete(Document))
        db.session.execute(update(Category).values(parent_id=None))
        db.session.execute(delete(Category))
        db.session.execute(delete(User))
        db.session.execute(update(Organization).values(parent_id=None))
        db.session.execute(delete(Organization))
        db.session.commit()

        root = Organization(
            org_code="ORG_ROOT",
            org_name="机关",
            org_path="/机关",
            sort_no=10,
            status="ENABLED",
        )
        db.session.add(root)
        db.session.flush()
        ops = Organization(
            org_code="ORG_OPS",
            org_name="作战部",
            org_path="/机关/作战部",
            parent_id=root.id,
            sort_no=20,
            status="ENABLED",
        )
        comms = Organization(
            org_code="ORG_COMMS",
            org_name="通信部",
            org_path="/机关/通信部",
            parent_id=root.id,
            sort_no=30,
            status="ENABLED",
        )
        disabled_org = Organization(
            org_code="ORG_DISABLED",
            org_name="停用部门",
            org_path="/机关/停用部门",
            parent_id=root.id,
            sort_no=40,
            status="DISABLED",
        )
        db.session.add_all([ops, comms, disabled_org])
        db.session.flush()
        db.session.add_all(
            [
                User(
                    username="admin",
                    password_hash=hash_password(password),
                    real_name="系统管理员",
                    organization_id=root.id,
                    organization_name=root.org_name,
                    role_code="ADMIN",
                    security_level="TOP_SECRET",
                    status="ENABLED",
                ),
                User(
                    username="auditor",
                    password_hash=hash_password(password),
                    real_name="历史审计员",
                    organization_id=root.id,
                    organization_name=root.org_name,
                    role_code="USER",
                    security_level="TOP_SECRET",
                    status="DISABLED",
                    auth_version=1,
                    is_deleted=True,
                ),
                User(
                    username="user",
                    password_hash=hash_password(password),
                    real_name="普通用户",
                    organization_id=ops.id,
                    organization_name=ops.org_name,
                    role_code="USER",
                    security_level="SECRET",
                    status="ENABLED",
                ),
                User(
                    username="disabled",
                    password_hash=hash_password(password),
                    real_name="停用用户",
                    organization_id=comms.id,
                    organization_name=comms.org_name,
                    role_code="USER",
                    security_level="SECRET",
                    status="DISABLED",
                ),
            ]
        )
        db.session.commit()

    yield app

    with app.app_context():
        db.session.remove()
    test_storage.cleanup()


@pytest.fixture()
def b2_client(b2_app):
    return b2_app.test_client()


@pytest.fixture()
def b2_password():
    password = os.environ.get("DMS_TEST_USER_PASSWORD")
    if not password:
        pytest.skip("缺少B2测试专用密码环境变量DMS_TEST_USER_PASSWORD")
    return password


@pytest.fixture()
def login_token(b2_client, b2_password):
    response = b2_client.post(
        "/api/v1/auth/login",
        json={
            "username": "admin",
            "password": b2_password,
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 200
    return response.get_json()["data"]["accessToken"]


@pytest.fixture()
def token_for(b2_client, b2_password):
    def issue(username: str) -> str:
        response = b2_client.post(
            "/api/v1/auth/login",
            json={
                "username": username,
                "password": b2_password,
                "keepSignedIn": False,
            },
        )
        assert response.status_code == 200
        return response.get_json()["data"]["accessToken"]

    return issue


@pytest.fixture()
def role_cleanup(b2_app):
    """清空角色两表并解除用户角色关联，供角色管理/矩阵判定测试隔离。"""
    from sqlalchemy import delete, update

    from dms.extensions import db
    from dms.models import Role, RolePermission, User

    with b2_app.app_context():
        db.session.execute(update(User).values(role_id=None))
        db.session.execute(delete(RolePermission))
        db.session.execute(delete(Role))
        db.session.commit()


@pytest.fixture()
def b3_app(b2_app):
    from dms.extensions import db
    from dms.models import Category

    with b2_app.app_context():
        root_a = Category(
            category_code="ROOT_A",
            category_name="场景根A",
            category_type="SCENE",
            category_path="/场景根A",
            sort_no=10,
            document_count=0,
            status="ENABLED",
        )
        root_b = Category(
            category_code="ROOT_B",
            category_name="场景根B",
            category_type="SCENE",
            category_path="/场景根B",
            sort_no=10,
            document_count=0,
            status="ENABLED",
        )
        root_c = Category(
            category_code="ROOT_C",
            category_name="场景根C",
            category_type="SCENE",
            category_path="/场景根C",
            sort_no=30,
            document_count=0,
            status="ENABLED",
        )
        disabled = Category(
            category_code="DISABLED_ROOT",
            category_name="停用分类",
            category_type="OTHER",
            category_path="/停用分类",
            sort_no=40,
            document_count=0,
            status="DISABLED",
        )
        deleted = Category(
            category_code="DELETED_CODE",
            category_name="已删除分类",
            category_type="OTHER",
            category_path="/已删除分类",
            sort_no=50,
            document_count=0,
            status="ENABLED",
            is_deleted=True,
        )
        db.session.add_all([root_a, root_b, root_c, disabled, deleted])
        db.session.flush()
        db.session.add_all(
            [
                Category(
                    category_code="CHILD_A1",
                    category_name="样式A1",
                    category_type="STYLE",
                    parent_id=root_a.id,
                    category_path="/场景根A/样式A1",
                    sort_no=10,
                    document_count=0,
                    status="ENABLED",
                ),
                Category(
                    category_code="CHILD_A2",
                    category_name="样式A2",
                    category_type="STYLE",
                    parent_id=root_a.id,
                    category_path="/场景根A/样式A2",
                    sort_no=20,
                    document_count=0,
                    status="ENABLED",
                ),
            ]
        )
        db.session.commit()
    return b2_app


@pytest.fixture()
def b3_client(b3_app):
    return b3_app.test_client()


@pytest.fixture()
def b4_app(b3_app):
    from dms.extensions import db
    from dms.models import (
        AttachmentBinding,
        Category,
        Document,
        Permission,
        User,
    )

    def document(
        name: str,
        document_type: str,
        *,
        category: Category | None = None,
        status: str = "PUBLISHED",
        security: str = "INTERNAL",
        parent: Document | None = None,
        root: Document | None = None,
        attachment_type: str | None = None,
        deleted: bool = False,
    ) -> Document:
        item = Document(
            document_name=name,
            summary=f"{name}摘要",
            document_type=document_type,
            document_status=status,
            security_level=security,
            attachment_type=attachment_type,
            category_id=category.id if category else None,
            category_name=category.category_name if category else None,
            category_path=category.category_path if category else None,
            parent_document_id=parent.id if parent else None,
            root_document_id=root.id if root else None,
            tags=["B4", "测试"],
            original_file_name=f"{name}.docx",
            file_relative_path=f"original/B4_TEST_{name}.docx",
            file_extension="docx",
            mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            file_size=128,
            file_hash="a" * 64,
            search_text=f"{name} B4 测试",
            created_by_name="系统管理员",
            updated_by_name="系统管理员",
            is_deleted=deleted,
        )
        db.session.add(item)
        db.session.flush()
        return item

    with b3_app.app_context():
        category = db.session.scalar(
            db.select(Category).where(Category.category_code == "ROOT_A")
        )
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        user = db.session.scalar(db.select(User).where(User.username == "user"))
        assert (
            category is not None
            and admin is not None
            and user is not None
        )
        from datetime import datetime, timedelta, timezone

        borrow_expires_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(days=7)

        main_all = document("B4_TEST_全部主案", "MAIN", category=category)
        main_org = document(
            "B4_TEST_组织主案",
            "MAIN",
            category=category,
            security="SECRET",
        )
        main_custom = document(
            "B4_TEST_自定义主案",
            "MAIN",
            category=category,
            security="SECRET",
        )
        document(
            "B4_TEST_草稿主案",
            "MAIN",
            category=category,
            status="DRAFT",
        )
        document(
            "B4_TEST_绝密主案",
            "MAIN",
            category=category,
            security="TOP_SECRET",
        )
        document(
            "B4_TEST_无授权主案",
            "MAIN",
            category=category,
        )
        sub = document(
            "B4_TEST_动态继承子案",
            "SUB_PLAN",
            category=category,
            parent=main_custom,
            root=main_custom,
            security="SECRET",
        )
        main_custom.child_count = 1
        attachment_one = document(
            "B4_TEST_规范附件",
            "ATTACHMENT",
            security="PUBLIC",
            attachment_type="TRAINING_MATERIAL",
        )
        attachment_two = document(
            "B4_TEST_表格附件",
            "ATTACHMENT",
            security="PUBLIC",
            attachment_type="SUPPORT_ACHIEVEMENT",
        )
        document(
            "B4_TEST_已删除附件",
            "ATTACHMENT",
            security="PUBLIC",
            attachment_type="TYPICAL_CASE",
            deleted=True,
        )
        db.session.add_all(
            [
                # 借阅语义：仅用户授权，限时生效（7天内），替代原组织授权。
                Permission(
                    document_id=main_all.id,
                    subject_id=user.id,
                    subject_name=user.real_name,
                    can_view=True,
                    can_download=True,
                    expires_at=borrow_expires_at,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
                Permission(
                    document_id=main_org.id,
                    subject_id=user.id,
                    subject_name=user.real_name,
                    can_view=True,
                    can_download=True,
                    expires_at=borrow_expires_at,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
                Permission(
                    document_id=main_custom.id,
                    subject_id=user.id,
                    subject_name=user.real_name,
                    can_view=True,
                    can_download=True,
                    expires_at=borrow_expires_at,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
                AttachmentBinding(
                    main_document_id=main_all.id,
                    attachment_document_id=attachment_one.id,
                    sort_no=20,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
                AttachmentBinding(
                    main_document_id=main_all.id,
                    attachment_document_id=attachment_two.id,
                    sort_no=10,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
                AttachmentBinding(
                    main_document_id=main_org.id,
                    attachment_document_id=attachment_one.id,
                    sort_no=10,
                    created_by=admin.id,
                    updated_by=admin.id,
                ),
            ]
        )
        main_all.attachment_count = 2
        main_org.attachment_count = 1
        db.session.commit()
        b3_app.config["B4_IDS"] = {
            "main_all": main_all.id,
            "main_org": main_org.id,
            "main_custom": main_custom.id,
            "sub": sub.id,
            "attachment_one": attachment_one.id,
            "attachment_two": attachment_two.id,
        }
    return b3_app


@pytest.fixture()
def b4_client(b4_app):
    return b4_app.test_client()


@pytest.fixture()
def b5_app(b3_app):
    return b3_app


@pytest.fixture()
def b5_client(b5_app):
    return b5_app.test_client()


@pytest.fixture()
def b6_app(b4_app):
    return b4_app


@pytest.fixture()
def b6_client(b6_app):
    return b6_app.test_client()


@pytest.fixture()
def b7_app(b4_app):
    from datetime import datetime

    from dms.extensions import db
    from dms.models import AuditLog, User

    with b4_app.app_context():
        users = {
            item.username: item
            for item in db.session.scalars(db.select(User)).all()
        }

        def audit(
            username,
            created_at,
            action,
            *,
            result="SUCCESS",
            target_type="DOCUMENT",
            target_id=101,
            target_name="B7目标",
            real_name=None,
            organization_name=None,
            failure_reason=None,
            request_id=None,
            client_ip="127.0.0.1",
        ):
            user = users.get(username) if username else None
            row = AuditLog(
                user_id=user.id if user else None,
                username=username,
                real_name=real_name or (user.real_name if user else None),
                organization_id=user.organization_id if user else None,
                organization_name=(
                    organization_name
                    or (user.organization_name if user else None)
                ),
                action_type=action,
                target_type=target_type,
                target_id=target_id,
                target_name=target_name,
                operation_result=result,
                failure_reason=failure_reason,
                client_ip=client_ip,
                request_id=request_id,
                created_at=created_at,
            )
            db.session.add(row)
            db.session.flush()
            return row

        rows = [
            audit(
                "admin",
                datetime(2026, 1, 5, 0, 0),
                "VIEW_DOCUMENT",
                real_name="管理员旧名",
                organization_name="旧组织",
                target_name="关键目标",
                request_id="b7-request-alpha",
            ),
            audit(
                "admin",
                datetime(2026, 1, 5, 1, 0),
                "DOWNLOAD_DOCUMENT",
                result="FAILURE",
                failure_reason="关键失败原因",
                request_id="b7-request-beta",
            ),
            audit(
                "auditor",
                datetime(2026, 1, 6, 2, 0),
                "LOGIN",
                target_type="AUTH",
                target_id=users["auditor"].id,
                target_name="auditor",
                request_id="b7-request-gamma",
            ),
            audit(
                "user",
                datetime(2026, 1, 7, 0, 0),
                "VIEW_DOCUMENT",
                target_id=102,
                request_id="b7-request-delta",
            ),
            audit(
                "user",
                datetime(2026, 1, 7, 1, 0),
                "VIEW_DOCUMENT",
                target_id=103,
                request_id="b7-request-epsilon",
            ),
            audit(
                "admin",
                datetime(2026, 1, 7, 1, 0),
                "VIEW_DOCUMENT",
                target_id=104,
                real_name="管理员新名",
                organization_name="新组织",
                request_id="b7-request-zeta",
            ),
            audit(
                "admin",
                datetime(2026, 1, 7, 2, 0),
                "VIEW_DOCUMENT",
                result="FAILURE",
                target_id=105,
                real_name="管理员新名",
                organization_name="新组织",
                request_id="b7-request-eta",
            ),
            audit(
                None,
                datetime(2026, 1, 8, 0, 0),
                "LOGIN",
                result="FAILURE",
                target_type="AUTH",
                target_id=None,
                target_name="unknown",
                failure_reason="用户名不存在",
                request_id="b7-request-theta",
            ),
            audit(
                "auditor",
                datetime(2026, 2, 2, 0, 0),
                "LOGOUT",
                target_type="AUTH",
                target_id=users["auditor"].id,
                target_name="auditor",
                request_id="b7-request-iota",
            ),
        ]
        db.session.commit()
        b4_app.config["B7_AUDIT_IDS"] = [row.id for row in rows]
        b4_app.config["B7_USER_IDS"] = {
            name: item.id for name, item in users.items()
        }
    return b4_app


@pytest.fixture()
def b7_client(b7_app):
    return b7_app.test_client()