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

"""方案关联关系：预置配置/载荷校验单元测试 + HTTP 集成测试。"""

from __future__ import annotations

import io
import json

import pytest

from dms.common.errors import InvalidArgumentError
from dms.config import DEFAULT_DOCUMENT_RELATION_PRESET_CONFIG_PATH
from dms.services.document_relation_service import (
    DocumentRelationPresetError,
    _relations_payload,
    load_relation_presets,
)


# ---------------------------------------------------------------------------
# 纯单元测试（无数据库依赖）
# ---------------------------------------------------------------------------


def test_default_relation_presets_load_in_sorted_order():
    presets = load_relation_presets(DEFAULT_DOCUMENT_RELATION_PRESET_CONFIG_PATH)
    assert [(p.code, p.label, p.sort_no) for p in presets] == [
        ("BASIS", "依据", 10),
        ("SUPPORT", "支撑", 20),
        ("REFERENCE", "参照", 30),
    ]


def _write(tmp_path, content: str):
    path = tmp_path / "presets.json"
    path.write_text(content, encoding="utf-8")
    return path


@pytest.mark.parametrize(
    "content",
    [
        '{"presets": []}',  # 空清单
        '{"presets": [{"code": "A", "label": "依据"}]}',  # 缺sortNo
        '{"presets": [{"code": "A", "label": "依据", "sortNo": 1, "extra": 1}]}',  # 多余字段
        '{"presets": [{"code": "", "label": "依据", "sortNo": 1}]}',  # 空code
        '{"presets": [{"code": "A", "label": " ", "sortNo": 1}]}',  # 空label
        '{"presets": [{"code": "A", "label": "依据", "sortNo": "10"}]}',  # 非整数
        '{"items": []}',  # 根节点形状不符
        "not-json",  # 非JSON
    ],
)
def test_load_relation_presets_rejects_invalid(tmp_path, content):
    with pytest.raises(DocumentRelationPresetError):
        load_relation_presets(_write(tmp_path, content))


def test_load_relation_presets_rejects_duplicate_code_or_label(tmp_path):
    content = json.dumps(
        {
            "presets": [
                {"code": "A", "label": "依据", "sortNo": 1},
                {"code": "A", "label": "支撑", "sortNo": 2},
            ]
        },
        ensure_ascii=False,
    )
    with pytest.raises(DocumentRelationPresetError):
        load_relation_presets(_write(tmp_path, content))
    content = json.dumps(
        {
            "presets": [
                {"code": "A", "label": "依据", "sortNo": 1},
                {"code": "B", "label": "依据", "sortNo": 2},
            ]
        },
        ensure_ascii=False,
    )
    with pytest.raises(DocumentRelationPresetError):
        load_relation_presets(_write(tmp_path, content))


def test_load_relation_presets_missing_file(tmp_path):
    with pytest.raises(DocumentRelationPresetError):
        load_relation_presets(tmp_path / "absent.json")


@pytest.mark.parametrize(
    "payload",
    [
        None,
        [],
        {"relations": []},  # 缺rowVersion
        {"rowVersion": 0},  # 缺relations
        {"relations": [], "rowVersion": 0, "extra": 1},  # 多余字段
        {"relations": {}, "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "1"}], "rowVersion": 0},  # 缺relationType
        {"relations": [{"relationType": "依据"}], "rowVersion": 0},  # 缺relatedDocumentId
        {"relations": [{"relatedDocumentId": 1, "relationType": "依据"}], "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "x", "relationType": "依据"}], "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "1", "relationType": ""}], "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "1", "relationType": " "}], "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "1", "relationType": "依" * 65}], "rowVersion": 0},
        {"relations": [{"relatedDocumentId": "1", "relationType": "依据"}, {"relatedDocumentId": "1", "relationType": "支撑"}], "rowVersion": 0},  # 重复目标
        {"relations": [{"relatedDocumentId": "1", "relationType": "依据"}], "rowVersion": -1},
        {"relations": [{"relatedDocumentId": "1", "relationType": "依据"}], "rowVersion": "0"},
    ],
)
def test_relations_payload_rejects_invalid(payload):
    with pytest.raises(InvalidArgumentError):
        _relations_payload(payload)


def test_relations_payload_accepts_empty_and_custom_text():
    assert _relations_payload({"relations": [], "rowVersion": 3}) == ([], 3)
    pairs, version = _relations_payload(
        {
            "relations": [
                {"relatedDocumentId": "12", "relationType": " 依据 "},
                {"relatedDocumentId": "34", "relationType": "前置输入条件"},
            ],
            "rowVersion": 0,
        }
    )
    assert pairs == [(12, "依据"), (34, "前置输入条件")]
    assert version == 0


# ---------------------------------------------------------------------------
# HTTP 集成测试（依赖独立测试库，本机无MySQL时自动跳过）
# ---------------------------------------------------------------------------


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


def _put_relations(client, document_id, relations, row_version, headers):
    return client.put(
        f"/api/v1/documents/{document_id}/relations",
        json={"relations": relations, "rowVersion": row_version},
        headers=headers,
    )


def _get_relations(client, document_id, headers):
    response = client.get(
        f"/api/v1/documents/{document_id}/relations", headers=headers
    )
    assert response.status_code == 200, response.get_json()
    return response.get_json()["data"]["items"]


def test_relation_presets_endpoint(b4_client, token_for):
    response = b4_client.get(
        "/api/v1/documents/relation-presets", headers=auth(token_for("user"))
    )
    assert response.status_code == 200
    presets = response.get_json()["data"]["presets"]
    assert [item["code"] for item in presets] == ["BASIS", "SUPPORT", "REFERENCE"]
    assert presets[0]["label"] == "依据"


def test_save_list_replace_and_clear_relations(b4_client, b4_app, token_for):
    ids = b4_app.config["B4_IDS"]
    main_id, target_id = ids["main_all"], ids["main_org"]
    admin = auth(token_for("admin"))

    # 初始为空
    assert _get_relations(b4_client, main_id, admin) == []

    # 保存一条预置关系 + 一条自定义关系
    saved = _put_relations(
        b4_client,
        main_id,
        [
            {"relatedDocumentId": str(target_id), "relationType": "依据"},
            {"relatedDocumentId": str(ids["main_custom"]), "relationType": "前置输入"},
        ],
        0,
        admin,
    )
    assert saved.status_code == 200, saved.get_json()
    data = saved.get_json()["data"]
    assert data["rowVersion"] == 1
    assert [
        (item["relatedDocumentId"], item["relationType"])
        for item in data["relations"]
    ] == [(str(target_id), "依据"), (str(ids["main_custom"]), "前置输入")]

    # GET 返回被关联方案名称
    items = _get_relations(b4_client, main_id, admin)
    assert items[0]["relatedDocumentName"] == "B4_TEST_组织主案"

    # 整体替换为空 = 清空；rowVersion 再次自增
    cleared = _put_relations(b4_client, main_id, [], data["rowVersion"], admin)
    assert cleared.status_code == 200
    assert cleared.get_json()["data"]["relations"] == []
    assert cleared.get_json()["data"]["rowVersion"] == 2
    assert _get_relations(b4_client, main_id, admin) == []


def test_save_relations_stale_row_version_conflicts(b4_client, b4_app, token_for):
    ids = b4_app.config["B4_IDS"]
    admin = auth(token_for("admin"))
    response = _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "依据"}],
        0,
        admin,
    )
    assert response.status_code == 200
    stale = _put_relations(
        b4_client,
        ids["main_all"],
        [],
        0,  # 旧版本号
        admin,
    )
    assert stale.status_code == 409


def test_save_relations_target_must_be_main_and_not_self(
    b4_client, b4_app, token_for
):
    ids = b4_app.config["B4_IDS"]
    admin = auth(token_for("admin"))
    base = _get_relations(b4_client, ids["main_all"], admin)
    version = 0

    # 自关联被拒
    self_relation = _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["main_all"]), "relationType": "依据"}],
        version,
        admin,
    )
    assert self_relation.status_code == 400
    # 子方案/附件不能作为被关联对象
    sub_target = _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["sub"]), "relationType": "依据"}],
        version,
        admin,
    )
    assert sub_target.status_code == 404
    attachment_target = _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["attachment_one"]), "relationType": "依据"}],
        version,
        admin,
    )
    assert attachment_target.status_code == 404
    assert _get_relations(b4_client, ids["main_all"], admin) == base


def test_sub_plan_may_declare_relations_but_attachment_may_not(
    b4_client, b4_app, token_for
):
    ids = b4_app.config["B4_IDS"]
    admin = auth(token_for("admin"))

    # 子方案可作为发起方
    ok = _put_relations(
        b4_client,
        ids["sub"],
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "支撑"}],
        0,
        admin,
    )
    assert ok.status_code == 200, ok.get_json()

    # 附件不可作为发起方
    rejected = _put_relations(
        b4_client,
        ids["attachment_one"],
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "支撑"}],
        0,
        admin,
    )
    assert rejected.status_code == 404


def test_save_relations_requires_admin(b4_client, b4_app, token_for):
    ids = b4_app.config["B4_IDS"]
    response = _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "依据"}],
        0,
        auth(token_for("user")),
    )
    assert response.status_code == 403


def test_user_with_view_access_can_list_relations(b4_client, b4_app, token_for):
    ids = b4_app.config["B4_IDS"]
    admin = auth(token_for("admin"))
    _put_relations(
        b4_client,
        ids["main_all"],
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "依据"}],
        0,
        admin,
    )
    items = _get_relations(b4_client, ids["main_all"], auth(token_for("user")))
    assert len(items) == 1
    assert items[0]["relationType"] == "依据"


def test_locked_main_relations_require_edit_password(
    b4_client, b4_app, token_for
):
    from dms.extensions import db
    from dms.models import Category

    with b4_app.app_context():
        category = db.session.scalar(
            db.select(Category).where(Category.category_code == "CHILD_A1")
        )
        category_id = str(category.id)

    name = "REL_TEST_锁定主案"
    metadata = {
        "documentName": name,
        "documentType": "MAIN",
        "summary": "关联关系密码锁测试",
        "categoryId": category_id,
        "securityLevel": "INTERNAL",
        "tags": ["REL"],
        "password": "rel-pw-123",
        "protectedActions": ["EDIT"],
    }
    created = b4_client.post(
        "/api/v1/documents",
        data={
            "file": (io.BytesIO(b"%PDF-1.4\n%%EOF"), name + ".pdf"),
            "metadata": json.dumps(metadata, ensure_ascii=False),
        },
        headers=auth(token_for("admin")),
        content_type="multipart/form-data",
    )
    assert created.status_code == 201, created.get_json()
    locked_id = created.get_json()["data"]["id"]

    ids = b4_app.config["B4_IDS"]
    blocked = _put_relations(
        b4_client,
        locked_id,
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "依据"}],
        0,
        auth(token_for("admin")),
    )
    assert blocked.status_code == 403
    assert blocked.get_json()["code"] == "DOCUMENT_PASSWORD_REQUIRED"

    headers = auth(token_for("admin"))
    headers["X-DMS-Document-Password"] = "rel-pw-123"
    ok = _put_relations(
        b4_client,
        locked_id,
        [{"relatedDocumentId": str(ids["main_org"]), "relationType": "依据"}],
        0,
        headers,
    )
    assert ok.status_code == 200, ok.get_json()