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

from __future__ import annotations

import hashlib
from pathlib import Path

import pytest
from docx import Document as WordDocument


def auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def data(response):
    return response.get_json()["data"]


def by_name(items, name: str):
    return next(item for item in items if item["documentName"] == name)


def test_documents_require_authentication(b4_client):
    response = b4_client.get("/api/v1/documents")
    assert response.status_code == 401
    assert response.get_json()["code"] == "TOKEN_INVALID"


def test_admin_sees_all_plan_statuses_but_not_attachments(
    b4_client, token_for
):
    response = b4_client.get(
        "/api/v1/documents?pageSize=100", headers=auth(token_for("admin"))
    )
    assert response.status_code == 200
    items = data(response)["items"]
    assert any(item["status"] == "DRAFT" for item in items)
    assert all(item["documentType"] != "ATTACHMENT" for item in items)


def test_user_list_filters_status_security_and_acl(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/documents?pageSize=100", headers=auth(token_for("user"))
    )
    assert response.status_code == 200
    names = {item["documentName"] for item in data(response)["items"]}
    assert "B4_TEST_全部主案" in names
    assert "B4_TEST_组织主案" in names
    assert "B4_TEST_自定义主案" in names
    assert "B4_TEST_动态继承子案" not in names
    assert "B4_TEST_草稿主案" not in names
    assert "B4_TEST_绝密主案" not in names
    assert "B4_TEST_无授权主案" not in names


def test_historical_auditor_cannot_login(b4_client, b2_password):
    response = b4_client.post(
        "/api/v1/auth/login",
        json={
            "username": "auditor",
            "password": b2_password,
            "keepSignedIn": False,
        },
    )
    assert response.status_code == 401
    assert response.get_json()["code"] == "INVALID_CREDENTIALS"


@pytest.mark.parametrize(
    ("query", "expected"),
    [
        ("documentType=ATTACHMENT", 400),
        ("documentType=MAIN,ATTACHMENT", 400),
        ("sortBy=fileRelativePath", 400),
        ("sortDirection=sideways", 400),
        ("pageSize=101", 400),
        ("categoryId=not-an-id", 400),
        ("status=INVALID", 400),
    ],
)
def test_document_invalid_filters(b4_client, token_for, query, expected):
    response = b4_client.get(
        f"/api/v1/documents?{query}", headers=auth(token_for("admin"))
    )
    assert response.status_code == expected
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.parametrize(
    "sort_by",
    ["documentName", "createdAt", "updatedAt", "viewCount", "downloadCount"],
)
def test_document_sort_whitelist_accepts_contract_fields(
    b4_client, token_for, sort_by
):
    response = b4_client.get(
        f"/api/v1/documents?sortBy={sort_by}&sortDirection=asc",
        headers=auth(token_for("admin")),
    )
    assert response.status_code == 200


def test_dynamic_sub_plan_inherits_root_access(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/documents?documentType=SUB_PLAN",
        headers=auth(token_for("user")),
    )
    item = by_name(data(response)["items"], "B4_TEST_动态继承子案")
    assert "visibilityType" not in item
    assert item["rootDocumentId"].isdecimal()
    assert item["allowedActions"] == ["VIEW", "DOWNLOAD"]


def test_document_detail_has_contract_fields_and_no_storage_path(
    b4_app, b4_client, token_for
):
    document_id = b4_app.config["B4_IDS"]["main_all"]
    response = b4_client.get(
        f"/api/v1/documents/{document_id}", headers=auth(token_for("user"))
    )
    assert response.status_code == 200
    item = data(response)
    assert item["id"] == str(document_id)
    assert item["fileSize"] == 128
    assert item["permissionSummary"]["inheritedFromMainPlan"] is False
    assert "fileRelativePath" not in item
    assert "updatedByName" not in item
    assert isinstance(item["tags"], list)


def test_document_detail_increments_view_and_audits(
    b4_app, b4_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, Document

    document_id = b4_app.config["B4_IDS"]["main_all"]
    response = b4_client.get(
        f"/api/v1/documents/{document_id}", headers=auth(token_for("user"))
    )
    assert response.status_code == 200
    with b4_app.app_context():
        document = db.session.get(Document, document_id)
        audits = db.session.scalars(
            db.select(AuditLog).where(
                AuditLog.target_id == document_id,
                AuditLog.action_type == "VIEW_DOCUMENT",
            )
        ).all()
        assert document.view_count == 1
        assert len(audits) == 1
        assert audits[0].target_type == "DOCUMENT"


def test_attachment_cannot_use_document_detail(
    b4_app, b4_client, token_for
):
    attachment_id = b4_app.config["B4_IDS"]["attachment_one"]
    response = b4_client.get(
        f"/api/v1/documents/{attachment_id}", headers=auth(token_for("admin"))
    )
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"


def test_user_security_failure_is_specific(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/documents?pageSize=100", headers=auth(token_for("admin"))
    )
    top_secret = by_name(data(response)["items"], "B4_TEST_绝密主案")
    forbidden = b4_client.get(
        f"/api/v1/documents/{top_secret['id']}",
        headers=auth(token_for("user")),
    )
    assert forbidden.status_code == 403
    assert forbidden.get_json()["code"] == "SECURITY_LEVEL_FORBIDDEN"


def test_user_acl_failure_is_specific(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/documents?pageSize=100", headers=auth(token_for("admin"))
    )
    hidden = by_name(data(response)["items"], "B4_TEST_无授权主案")
    forbidden = b4_client.get(
        f"/api/v1/documents/{hidden['id']}",
        headers=auth(token_for("user")),
    )
    assert forbidden.status_code == 403
    assert forbidden.get_json()["code"] == "DOCUMENT_VIEW_FORBIDDEN"


def test_sub_plan_page_and_main_type_validation(b4_app, b4_client, token_for):
    main_id = b4_app.config["B4_IDS"]["main_custom"]
    response = b4_client.get(
        f"/api/v1/main-plans/{main_id}/sub-plans?keyword=动态&pageSize=1",
        headers=auth(token_for("user")),
    )
    assert response.status_code == 200
    assert data(response)["total"] == 1
    sub_id = b4_app.config["B4_IDS"]["sub"]
    invalid = b4_client.get(
        f"/api/v1/main-plans/{sub_id}/sub-plans",
        headers=auth(token_for("admin")),
    )
    assert invalid.status_code == 404
    assert invalid.get_json()["code"] == "MAIN_PLAN_NOT_FOUND"


def test_attachment_library_list_open_for_all_roles(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/attachments?pageSize=100",
        headers=auth(token_for("admin")),
    )
    assert response.status_code == 200
    assert data(response)["total"] == 2
    assert all(
        item["securityLevel"] == "PUBLIC" for item in data(response)["items"]
    )
    # 列表全量可见（全局检索需要）：普通用户返回全部标题，但操作位为空。
    user_response = b4_client.get(
        "/api/v1/attachments?pageSize=100",
        headers=auth(token_for("user")),
    )
    assert user_response.status_code == 200
    assert data(user_response)["total"] == 2
    assert all(
        item["allowedActions"] == [] for item in data(user_response)["items"]
    )


def test_attachment_filters_and_real_mount_count(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/attachments?attachmentType=TRAINING_MATERIAL&fileExtension=docx",
        headers=auth(token_for("admin")),
    )
    assert response.status_code == 200
    assert data(response)["total"] == 1
    assert data(response)["items"][0]["mountedPlanCount"] == 2


def test_attachment_detail_increments_view_and_attachment_audit(
    b4_app, b4_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, Document

    attachment_id = b4_app.config["B4_IDS"]["attachment_one"]
    response = b4_client.get(
        f"/api/v1/attachments/{attachment_id}",
        headers=auth(token_for("user")),
    )
    assert response.status_code == 200
    item = data(response)
    assert item["categoryId"] is None
    assert item["permissionSummary"] == {
        "borrowerCount": 0,
        "inheritedFromMainPlan": False,
    }
    assert "fileRelativePath" not in item
    with b4_app.app_context():
        assert db.session.get(Document, attachment_id).view_count == 1
        audit = db.session.scalar(
            db.select(AuditLog).where(AuditLog.target_id == attachment_id)
        )
        assert audit.target_type == "ATTACHMENT"


def test_non_attachment_returns_not_found(b4_app, b4_client, token_for):
    main_id = b4_app.config["B4_IDS"]["main_all"]
    response = b4_client.get(
        f"/api/v1/attachments/{main_id}", headers=auth(token_for("admin"))
    )
    assert response.status_code == 404


def test_attachment_main_plans_respect_plan_access(
    b4_app, b4_client, token_for
):
    attachment_id = b4_app.config["B4_IDS"]["attachment_one"]
    user_response = b4_client.get(
        f"/api/v1/attachments/{attachment_id}/main-plans",
        headers=auth(token_for("user")),
    )
    assert data(user_response)["total"] == 2
    admin_response = b4_client.get(
        f"/api/v1/attachments/{attachment_id}/main-plans",
        headers=auth(token_for("admin")),
    )
    assert data(admin_response)["total"] == 2


def test_main_plan_attachments_stable_binding_order(
    b4_app, b4_client, token_for
):
    main_id = b4_app.config["B4_IDS"]["main_all"]
    response = b4_client.get(
        f"/api/v1/main-plans/{main_id}/attachments",
        headers=auth(token_for("user")),
    )
    assert response.status_code == 200
    items = data(response)["items"]
    assert [item["bindingSortNo"] for item in items] == [10, 20]
    assert all(item["bindingId"].isdecimal() for item in items)


def test_allowed_actions_are_stable(b4_app, b4_client, token_for):
    main_id = b4_app.config["B4_IDS"]["main_all"]
    admin = b4_client.get(
        f"/api/v1/documents/{main_id}", headers=auth(token_for("admin"))
    )
    assert data(admin)["allowedActions"] == [
        "VIEW",
        "DOWNLOAD",
        "EDIT",
        "CONFIG_PERMISSION",
        "DELETE",
        "BIND_ATTACHMENT",
        "UNBIND_ATTACHMENT",
    ]
    attachment_id = b4_app.config["B4_IDS"]["attachment_one"]
    user = b4_client.get(
        f"/api/v1/attachments/{attachment_id}",
        headers=auth(token_for("user")),
    )
    assert data(user)["allowedActions"] == ["VIEW", "DOWNLOAD"]


def test_seed_documents_is_idempotent_and_creates_valid_files(
    b3_app, tmp_path: Path
):
    from dms.extensions import db
    from dms.models import Category, Document
    from dms.seed_documents import seed_documents

    with b3_app.app_context():
        # B3测试分类编码与开发初始化编码不同，补齐seed的必要依赖。
        for code, name in (
            ("SCENE_A", "B4_TEST_场景A"),
            ("SCENE_B", "B4_TEST_场景B"),
            ("SCENE_C", "B4_TEST_场景C"),
        ):
            db.session.add(
                Category(
                    category_code=code,
                    category_name=name,
                    category_type="SCENE",
                    category_path=f"/{name}",
                    sort_no=10,
                    document_count=0,
                    status="ENABLED",
                )
            )
        db.session.commit()
        scene_a = db.session.scalar(
            db.select(Category).where(Category.category_code == "SCENE_A")
        )
        db.session.add(
            Category(
                category_code="STYLE_A1",
                category_name="B4_TEST_样式A1",
                category_type="STYLE",
                parent_id=scene_a.id,
                category_path=f"{scene_a.category_path}/B4_TEST_样式A1",
                sort_no=10,
                document_count=0,
                status="ENABLED",
            )
        )
        db.session.commit()
        first = seed_documents(tmp_path / "storage")
        second = seed_documents(tmp_path / "storage")
        assert first == (9, 0, 0)
        assert second == (0, 9, 0)
        documents = db.session.scalars(
            db.select(Document).where(
                Document.document_name.in_(
                    ["2024年度综合应急预案", "装备保障工作规范"]
                )
            )
        ).all()
        assert len(documents) == 2
        for document in documents:
            path = tmp_path / "storage" / Path(document.file_relative_path)
            assert path.is_file()
            WordDocument(path)
            assert hashlib.sha256(path.read_bytes()).hexdigest() == document.file_hash