"""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) 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, Organization, Permission, 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) with app.app_context(): db.session.execute(delete(AuditLog)) 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="AUDITOR", security_level="TOP_SECRET", status="ENABLED", ), 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 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, Organization, Permission, User, ) def document( name: str, document_type: str, *, category: Category | None = None, status: str = "PUBLISHED", security: str = "INTERNAL", visibility: str = "ALL_AUTHENTICATED", 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, visibility_type=visibility, 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")) root_org = db.session.scalar( db.select(Organization).where(Organization.org_code == "ORG_ROOT") ) assert ( category is not None and admin is not None and user is not None and root_org is not None ) main_all = document("B4_TEST_全部主案", "MAIN", category=category) main_org = document( "B4_TEST_组织主案", "MAIN", category=category, security="SECRET", visibility="ORGANIZATION", ) main_custom = document( "B4_TEST_自定义主案", "MAIN", category=category, security="SECRET", visibility="CUSTOM", ) document( "B4_TEST_草稿主案", "MAIN", category=category, status="DRAFT", ) document( "B4_TEST_绝密主案", "MAIN", category=category, security="TOP_SECRET", ) document( "B4_TEST_无授权主案", "MAIN", category=category, visibility="CUSTOM", ) sub = document( "B4_TEST_动态继承子案", "SUB_PLAN", category=category, parent=main_custom, root=main_custom, security="SECRET", # 故意与根主案不同，验证输出和授权动态继承根主案。 visibility="ALL_AUTHENTICATED", ) main_custom.child_count = 1 attachment_one = document( "B4_TEST_规范附件", "ATTACHMENT", security="PUBLIC", visibility="ALL_AUTHENTICATED", attachment_type="WORK_STANDARD", ) attachment_two = document( "B4_TEST_表格附件", "ATTACHMENT", security="PUBLIC", visibility="ALL_AUTHENTICATED", attachment_type="TABLE", ) document( "B4_TEST_已删除附件", "ATTACHMENT", security="PUBLIC", visibility="ALL_AUTHENTICATED", attachment_type="OTHER", deleted=True, ) db.session.add_all( [ Permission( document_id=main_org.id, subject_type="ORG", subject_id=root_org.id, subject_name=root_org.org_name, can_view=True, can_download=True, created_by=admin.id, updated_by=admin.id, ), Permission( document_id=main_custom.id, subject_type="USER", subject_id=user.id, subject_name=user.real_name, can_view=True, can_download=True, 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()