"""文档归属部门行为测试（T1.2）。 覆盖： - 上传默认取创建者当前部门；显式 organizationId 可指定其他启用部门 - 无效/停用部门 → 404 RESOURCE_NOT_FOUND - PUT /documents/{id} 修改归属部门 - 编辑缺省 organizationId 沿用现有归属，不被编辑人部门覆盖 - 非管理员原样回传其他部门归属视为未变更；改挂其他部门 → 400 INVALID_ARGUMENT - 组织改名级联 Document.organization_name 冗余字段 - 用户调岗不联动历史文档归属 - 附件库条目输出 organizationName """ from __future__ import annotations import io import json import pytest from docx import Document as WordDocument from dms.extensions import db from dms.models import Document, Organization, User def _auth(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} def _docx_bytes() -> bytes: stream = io.BytesIO() document = WordDocument() document.add_paragraph("DO组织行为验证") document.save(stream) return stream.getvalue() def _upload(client, token: str, path: str, metadata: dict): name = metadata["documentName"] return client.post( f"/api/v1/{path}", data={ "file": (io.BytesIO(_docx_bytes()), f"{name}.docx"), "metadata": json.dumps(metadata, ensure_ascii=False), }, headers=_auth(token), content_type="multipart/form-data", ) def _leaf_category_id(app) -> str: from dms.models import Category with app.app_context(): category = db.session.scalar( db.select(Category).where(Category.category_code == "CHILD_A1") ) assert category is not None return str(category.id) def _org_by_code(app, code: str): with app.app_context(): org = db.session.scalar( db.select(Organization).where(Organization.org_code == code) ) assert org is not None return org.id, org.org_name, org.parent_id, org.row_version def _main_metadata(leaf_id: str, name: str, **extra) -> dict: return { "documentName": name, "documentType": "MAIN", "summary": "DO摘要", "categoryId": leaf_id, "securityLevel": "INTERNAL", "tags": ["DO"], **extra, } @pytest.fixture() def do_app(b3_app): with b3_app.app_context(): root = db.session.scalar( db.select(Organization).where(Organization.org_code == "ORG_ROOT") ) assert root is not None org_a = Organization( org_code="ORG_DO_A", org_name="DO甲部门", org_path=f"{root.org_path}/DO甲部门", parent_id=root.id, sort_no=80, status="ENABLED", ) org_disabled = Organization( org_code="ORG_DO_DISABLED", org_name="DO停用部门", org_path=f"{root.org_path}/DO停用部门", parent_id=root.id, sort_no=81, status="DISABLED", ) db.session.add_all([org_a, org_disabled]) db.session.commit() b3_app.config["DO_IDS"] = { "org_a": org_a.id, "org_disabled": org_disabled.id, } return b3_app @pytest.fixture() def do_client(do_app): return do_app.test_client() def test_upload_defaults_to_creator_organization(do_app, do_client, token_for): response = _upload( do_client, token_for("admin"), "documents", _main_metadata(_leaf_category_id(do_app), "DO_TEST_默认部门主案"), ) assert response.status_code == 201 item = response.get_json()["data"] with do_app.app_context(): admin = db.session.scalar(db.select(User).where(User.username == "admin")) assert item["organizationId"] == str(admin.organization_id) assert item["organizationName"] == admin.organization_name def test_upload_with_explicit_enabled_organization(do_app, do_client, token_for): org_a_id = do_app.config["DO_IDS"]["org_a"] response = _upload( do_client, token_for("admin"), "documents", _main_metadata( _leaf_category_id(do_app), "DO_TEST_指定部门主案", organizationId=str(org_a_id), ), ) assert response.status_code == 201 item = response.get_json()["data"] assert item["organizationId"] == str(org_a_id) assert item["organizationName"] == "DO甲部门" @pytest.mark.parametrize( "org_code", ["missing", "ORG_DO_DISABLED"], ) def test_upload_rejects_invalid_or_disabled_organization( do_app, do_client, token_for, org_code ): raw_id = ( "99999999" if org_code == "missing" else str(do_app.config["DO_IDS"][org_code]) ) response = _upload( do_client, token_for("admin"), "documents", _main_metadata( _leaf_category_id(do_app), f"DO_TEST_非法部门主案_{org_code}", organizationId=raw_id, ), ) assert response.status_code == 404 assert response.get_json()["code"] == "RESOURCE_NOT_FOUND" def test_update_document_organization(do_app, do_client, token_for): token = token_for("admin") org_a_id = do_app.config["DO_IDS"]["org_a"] created = _upload( do_client, token, "documents", _main_metadata(_leaf_category_id(do_app), "DO_TEST_修改部门主案"), ).get_json()["data"] assert created["organizationName"] != "DO甲部门" updated = do_client.put( f"/api/v1/documents/{created['id']}", json={ "documentName": created["documentName"], "summary": created["summary"], "categoryId": created["categoryId"], "securityLevel": created["securityLevel"], "tags": created["tags"], "rowVersion": created["rowVersion"], "organizationId": str(org_a_id), }, headers=_auth(token), ) assert updated.status_code == 200 assert updated.get_json()["data"]["organizationName"] == "DO甲部门" listed = do_client.get( "/api/v1/documents", query_string={ "keyword": "DO_TEST_修改部门主案", "organizationKeyword": "DO甲部", }, headers=_auth(token), ).get_json()["data"] assert [item["documentName"] for item in listed["items"]] == [ "DO_TEST_修改部门主案" ] def _edit_payload(item: dict, **extra) -> dict: return { "documentName": item["documentName"], "summary": item["summary"], "categoryId": item["categoryId"], "securityLevel": item["securityLevel"], "tags": item["tags"], "rowVersion": item["rowVersion"], **extra, } def test_update_without_organization_keeps_current( do_app, do_client, token_for ): """编辑未携带 organizationId 时沿用文档现有归属，不被编辑人部门覆盖。""" token = token_for("admin") org_a_id = do_app.config["DO_IDS"]["org_a"] created = _upload( do_client, token, "documents", _main_metadata( _leaf_category_id(do_app), "DO_TEST_编辑不改部门主案", organizationId=str(org_a_id), ), ).get_json()["data"] updated = do_client.put( f"/api/v1/documents/{created['id']}", json=_edit_payload(created), headers=_auth(token), ) assert updated.status_code == 200 data = updated.get_json()["data"] assert data["organizationId"] == str(org_a_id) assert data["organizationName"] == "DO甲部门" def test_non_admin_can_resubmit_other_organization_but_not_reassign( do_app, do_client, token_for ): """非管理员原样回传其他部门归属视为未变更；改挂其他部门被拒绝。""" admin_token = token_for("admin") user_token = token_for("user") org_a_id = do_app.config["DO_IDS"]["org_a"] comms_id, _, _, _ = _org_by_code(do_app, "ORG_COMMS") created = _upload( do_client, admin_token, "documents", _main_metadata( _leaf_category_id(do_app), "DO_TEST_普通用户编辑他部门主案", organizationId=str(org_a_id), ), ).get_json()["data"] with do_app.app_context(): user_id = db.session.scalar( db.select(User.id).where(User.username == "user") ) # 授予普通用户该主案EDIT权限（CUSTOM可见范围+用户授权） granted = do_client.put( f"/api/v1/documents/{created['id']}/permissions", json={ "visibilityType": "CUSTOM", "documentRowVersion": created["rowVersion"], "entries": [ { "subjectType": "USER", "subjectId": str(user_id), "canView": True, "canDownload": True, "canEdit": True, "canManagePermission": False, "canDelete": False, } ], }, headers=_auth(admin_token), ) assert granted.status_code == 200, granted.get_json() # 保存权限会推进rowVersion，重新读取详情 latest = do_client.get( f"/api/v1/documents/{created['id']}", headers=_auth(user_token) ).get_json()["data"] assert "EDIT" in latest["allowedActions"] # 原样回传其他部门（DO甲部门）归属：视为未变更，允许保存 unchanged = do_client.put( f"/api/v1/documents/{created['id']}", json=_edit_payload(latest, organizationId=str(org_a_id)), headers=_auth(user_token), ) assert unchanged.status_code == 200, unchanged.get_json() assert unchanged.get_json()["data"]["organizationId"] == str(org_a_id) # 改挂到通信部：非管理员被拒绝 reassigned = do_client.put( f"/api/v1/documents/{created['id']}", json=_edit_payload( unchanged.get_json()["data"], organizationId=str(comms_id) ), headers=_auth(user_token), ) assert reassigned.status_code == 400 body = reassigned.get_json() assert body["code"] == "INVALID_ARGUMENT" assert body["message"] == "仅管理员可将文档调整到其他部门" def test_organization_rename_cascades_document_name(do_app, do_client, token_for): token = token_for("admin") org_a_id, _, parent_id, row_version = _org_by_code(do_app, "ORG_DO_A") created = _upload( do_client, token, "documents", _main_metadata( _leaf_category_id(do_app), "DO_TEST_改名级联主案", organizationId=str(org_a_id), ), ).get_json()["data"] assert created["organizationName"] == "DO甲部门" def rename(name: str, version: int) -> int: response = do_client.put( f"/api/v1/organizations/{org_a_id}", json={ "orgName": name, "parentId": str(parent_id), "sortNo": 80, "status": "ENABLED", "rowVersion": version, }, headers=_auth(token), ) assert response.status_code == 200, response.get_json() return response.get_json()["data"]["rowVersion"] try: row_version = rename("DO甲部门新名", row_version) listed = do_client.get( "/api/v1/documents", query_string={"keyword": "DO_TEST_改名级联主案"}, headers=_auth(token), ).get_json()["data"] assert listed["items"][0]["organizationName"] == "DO甲部门新名" finally: rename("DO甲部门", row_version) def test_user_transfer_does_not_cascade_documents(do_app, do_client, token_for): token = token_for("admin") org_a_id = do_app.config["DO_IDS"]["org_a"] comms_id, comms_name, _, _ = _org_by_code(do_app, "ORG_COMMS") created_user = do_client.post( "/api/v1/users", json={ "username": "do_transfer_user", "realName": "调岗验证用户", "initialPassword": "Initial@2026", "organizationId": str(org_a_id), "roleCode": "USER", "securityLevel": "INTERNAL", "status": "ENABLED", }, headers=_auth(token), ) assert created_user.status_code == 201, created_user.get_json() user_id = created_user.get_json()["data"]["id"] with do_app.app_context(): leaf = db.session.scalar( db.select(Document).where( Document.document_name == "DO_TEST_修改部门主案", Document.is_deleted.is_(False), ) ) assert leaf is not None document = Document( document_name="DO_TEST_调岗不联动主案", summary="调岗不联动", document_type="MAIN", document_status="PUBLISHED", security_level="INTERNAL", visibility_type="CUSTOM", organization_id=org_a_id, organization_name="DO甲部门", tags=["DO"], original_file_name="DO_TEST_调岗不联动主案.docx", file_relative_path="original/DO_TEST_调岗不联动主案.docx", file_extension="docx", mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", file_size=64, file_hash="d" * 64, search_text="DO_TEST_调岗不联动主案", created_by=int(user_id), updated_by=int(user_id), created_by_name="调岗验证用户", updated_by_name="调岗验证用户", category_id=leaf.category_id, ) db.session.add(document) user = db.session.get(User, int(user_id)) row_version = user.row_version db.session.commit() transferred = do_client.put( f"/api/v1/users/{user_id}", json={ "realName": "调岗验证用户", "organizationId": str(comms_id), "roleCode": "USER", "securityLevel": "INTERNAL", "status": "ENABLED", "rowVersion": row_version, }, headers=_auth(token), ) assert transferred.status_code == 200, transferred.get_json() assert transferred.get_json()["data"]["organizationName"] == comms_name with do_app.app_context(): current = db.session.scalar( db.select(Document).where( Document.document_name == "DO_TEST_调岗不联动主案", Document.is_deleted.is_(False), ) ) assert current.organization_id == org_a_id assert current.organization_name == "DO甲部门" def test_attachment_library_exposes_organization_name( do_app, do_client, token_for ): token = token_for("admin") created = _upload( do_client, token, "attachments", { "documentName": "DO_TEST_部门附件", "attachmentType": "OTHER", "summary": "DO附件", "tags": ["DO"], }, ) assert created.status_code == 201 listed = do_client.get( "/api/v1/attachments", query_string={"keyword": "DO_TEST_部门附件"}, headers=_auth(token), ).get_json()["data"] assert len(listed["items"]) == 1 with do_app.app_context(): admin = db.session.scalar(db.select(User).where(User.username == "admin")) assert listed["items"][0]["organizationId"] == str(admin.organization_id) assert listed["items"][0]["organizationName"] == admin.organization_name