"""方案关联关系：预置配置/载荷校验单元测试 + 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()