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

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest


def _headers(token: str, *, origin: bool = False) -> dict[str, str]:
    headers = {"Authorization": f"Bearer {token}"}
    if origin:
        headers["Origin"] = "http://127.0.0.1:9346"
    return headers


def _ids(app) -> dict[str, int]:
    return app.config["B4_IDS"]


def _expires_at(days: int = 3) -> str:
    return (datetime.now(timezone.utc) + timedelta(days=days)).strftime(
        "%Y-%m-%dT%H:%M:%S.000Z"
    )


def _permission(subject_id: int, **overrides):
    values = {
        "canView": True,
        "canDownload": True,
        "expiresAt": _expires_at(),
    }
    values.update(overrides)
    return {"subjectId": str(subject_id), **values}


def _save(client, token, document_id, version, entries):
    return client.put(
        f"/api/v1/documents/{document_id}/permissions",
        json={
            "documentRowVersion": version,
            "entries": entries,
        },
        headers=_headers(token),
    )


def test_bind_deduplicates_preserves_order_and_is_idempotent(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, AuditLog, Document

    ids = _ids(b6_app)
    requested = [
        str(ids["attachment_two"]),
        str(ids["attachment_one"]),
        str(ids["attachment_two"]),
    ]
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={"attachmentIds": requested, "mainPlanRowVersion": 0},
        headers=_headers(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {
        "createdCount": 2,
        "existingCount": 0,
        "attachmentCount": 2,
        "mainPlanRowVersion": 1,
    }
    with b6_app.app_context():
        rows = db.session.scalars(
            db.select(AttachmentBinding)
            .where(
                AttachmentBinding.main_document_id == ids["main_custom"],
                AttachmentBinding.is_deleted.is_(False),
            )
            .order_by(AttachmentBinding.sort_no)
        ).all()
        assert [row.attachment_document_id for row in rows] == [
            ids["attachment_two"],
            ids["attachment_one"],
        ]
        assert [row.sort_no for row in rows] == [10, 20]
        assert db.session.get(Document, ids["main_custom"]).attachment_count == 2
        assert (
            db.session.scalar(
                db.select(db.func.count(AuditLog.id)).where(
                    AuditLog.action_type == "BIND_ATTACHMENT"
                )
            )
            == 1
        )

    again = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={"attachmentIds": requested, "mainPlanRowVersion": 1},
        headers=_headers(login_token),
    )
    assert again.status_code == 200
    assert again.get_json()["data"] == {
        "createdCount": 0,
        "existingCount": 2,
        "attachmentCount": 2,
        "mainPlanRowVersion": 1,
    }


def test_bind_counts_existing_and_appends_without_renumbering(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/bind",
        json={
            "attachmentIds": [
                str(ids["attachment_one"]),
                str(ids["attachment_two"]),
            ],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {
        "createdCount": 0,
        "existingCount": 2,
        "attachmentCount": 2,
        "mainPlanRowVersion": 0,
    }


@pytest.mark.parametrize(
    "payload",
    [
        {"attachmentIds": [], "mainPlanRowVersion": 0},
        {"attachmentIds": [1], "mainPlanRowVersion": 0},
        {"attachmentIds": ["1"], "mainPlanRowVersion": True},
        {"attachmentIds": ["1"], "mainPlanRowVersion": -1},
        {"attachmentIds": ["1"], "mainPlanRowVersion": 0, "extra": 1},
        {"attachmentIds": "1", "mainPlanRowVersion": 0},
    ],
)
def test_bind_rejects_invalid_payloads(
    payload, b6_app, b6_client, login_token
):
    response = b6_client.post(
        f"/api/v1/main-plans/{_ids(b6_app)['main_custom']}/attachments/bind",
        json=payload,
        headers=_headers(login_token),
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.parametrize("username", ["user"])
def test_binding_management_requires_admin(
    username, b6_app, b6_client, token_for
):
    ids = _ids(b6_app)
    token = token_for(username)
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"])],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(token),
    )
    assert response.status_code == 403
    assert response.get_json()["code"] == "FORBIDDEN"


def test_bind_invalid_attachment_rolls_back_everything(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document

    ids = _ids(b6_app)
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"]), "999999999"],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert response.status_code == 404
    with b6_app.app_context():
        assert (
            db.session.scalar(
                db.select(db.func.count(AttachmentBinding.id)).where(
                    AttachmentBinding.main_document_id == ids["main_custom"],
                    AttachmentBinding.is_deleted.is_(False),
                )
            )
            == 0
        )
        main = db.session.get(Document, ids["main_custom"])
        assert (main.attachment_count, main.row_version) == (0, 0)


def test_bind_rejects_wrong_document_types_and_version_conflict(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    wrong_main = b6_client.post(
        f"/api/v1/main-plans/{ids['sub']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"])],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert wrong_main.status_code == 404
    wrong_attachment = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["main_all"])],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert wrong_attachment.status_code == 404
    conflict = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"])],
            "mainPlanRowVersion": 9,
        },
        headers=_headers(login_token),
    )
    assert conflict.status_code == 409
    assert conflict.get_json()["details"]["currentRowVersion"] == 0


def test_unbind_is_logical_keeps_attachment_and_updates_both_queries(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, AuditLog, Document

    ids = _ids(b6_app)
    response = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=0",
        headers=_headers(login_token),
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {
        "attachmentCount": 1,
        "mainPlanRowVersion": 1,
    }
    forward = b6_client.get(
        f"/api/v1/main-plans/{ids['main_all']}/attachments",
        headers=_headers(login_token),
    )
    assert ids["attachment_one"] not in {
        int(item["id"]) for item in forward.get_json()["data"]["items"]
    }
    reverse = b6_client.get(
        f"/api/v1/attachments/{ids['attachment_one']}/main-plans",
        headers=_headers(login_token),
    )
    assert ids["main_all"] not in {
        int(item["id"]) for item in reverse.get_json()["data"]["items"]
    }
    with b6_app.app_context():
        binding = db.session.scalar(
            db.select(AttachmentBinding).where(
                AttachmentBinding.main_document_id == ids["main_all"],
                AttachmentBinding.attachment_document_id
                == ids["attachment_one"],
            )
        )
        assert binding.is_deleted is True
        attachment = db.session.get(Document, ids["attachment_one"])
        assert attachment is not None and not attachment.is_deleted
        audit = db.session.scalar(
            db.select(AuditLog).where(
                AuditLog.action_type == "UNBIND_ATTACHMENT"
            )
        )
        assert audit is not None
        assert audit.target_id == binding.id
        assert audit.operation_detail["attachmentCount"] == 1
        assert audit.operation_detail["mainPlanRowVersion"] == 1


def test_unbind_missing_relation_and_second_unbind_return_404(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    missing = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=0",
        headers=_headers(login_token),
    )
    assert missing.status_code == 404
    first = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=0",
        headers=_headers(login_token),
    )
    assert first.status_code == 200
    second = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=1",
        headers=_headers(login_token),
    )
    assert second.status_code == 404
    assert second.get_json()["code"] == "RESOURCE_NOT_FOUND"


def test_unbind_then_rebind_creates_new_binding_and_appends_sort(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding

    ids = _ids(b6_app)
    b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_two']}"
        "?mainPlanRowVersion=0",
        headers=_headers(login_token),
    )
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_two"])],
            "mainPlanRowVersion": 1,
        },
        headers=_headers(login_token),
    )
    assert response.status_code == 200
    with b6_app.app_context():
        rows = db.session.scalars(
            db.select(AttachmentBinding)
            .where(
                AttachmentBinding.main_document_id == ids["main_all"],
                AttachmentBinding.attachment_document_id
                == ids["attachment_two"],
            )
            .order_by(AttachmentBinding.id)
        ).all()
        assert len(rows) == 2
        assert rows[0].is_deleted is True
        assert rows[1].is_deleted is False
        assert rows[1].sort_no == 30


def test_binding_audit_failure_rolls_back(
    monkeypatch, b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document
    import dms.services.attachment_binding_service as service

    ids = _ids(b6_app)
    monkeypatch.setattr(
        service, "business_audit", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError())
    )
    response = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"])],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert response.status_code == 500
    with b6_app.app_context():
        assert (
            db.session.scalar(
                db.select(db.func.count(AttachmentBinding.id)).where(
                    AttachmentBinding.main_document_id == ids["main_custom"],
                    AttachmentBinding.is_deleted.is_(False),
                )
            )
            == 0
        )
        assert db.session.get(Document, ids["main_custom"]).row_version == 0


@pytest.mark.parametrize("username", ["user"])
def test_unbind_requires_admin(username, b6_app, b6_client, token_for):
    ids = _ids(b6_app)
    response = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=0",
        headers=_headers(token_for(username)),
    )
    assert response.status_code == 403


@pytest.mark.parametrize(
    "query",
    [
        "",
        "?mainPlanRowVersion=-1",
        "?mainPlanRowVersion=true",
        "?mainPlanRowVersion=0&extra=1",
    ],
)
def test_unbind_rejects_invalid_version_query(
    query, b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    response = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        f"{query}",
        headers=_headers(login_token),
    )
    assert response.status_code == 400


def test_unbind_version_conflict_and_audit_failure_rollback(
    monkeypatch, b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document
    import dms.services.attachment_binding_service as service

    ids = _ids(b6_app)
    conflict = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=8",
        headers=_headers(login_token),
    )
    assert conflict.status_code == 409
    assert conflict.get_json()["details"]["currentRowVersion"] == 0
    monkeypatch.setattr(
        service, "business_audit", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError())
    )
    failed = b6_client.delete(
        f"/api/v1/main-plans/{ids['main_all']}/attachments/{ids['attachment_one']}"
        "?mainPlanRowVersion=0",
        headers=_headers(login_token),
    )
    assert failed.status_code == 500
    with b6_app.app_context():
        binding = db.session.scalar(
            db.select(AttachmentBinding).where(
                AttachmentBinding.main_document_id == ids["main_all"],
                AttachmentBinding.attachment_document_id
                == ids["attachment_one"],
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        assert binding is not None
        main = db.session.get(Document, ids["main_all"])
        assert (main.attachment_count, main.row_version) == (2, 0)


def test_permission_get_main_sub_inheritance_and_attachment_error(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    main = b6_client.get(
        f"/api/v1/documents/{ids['main_custom']}/permissions",
        headers=_headers(login_token),
    )
    assert main.status_code == 200
    main_data = main.get_json()["data"]
    assert main_data["documentId"] == str(ids["main_custom"])
    assert main_data["sourceDocumentId"] == str(ids["main_custom"])
    assert main_data["inherited"] is False
    assert len(main_data["entries"]) == 1
    assert set(main_data["entries"][0]) == {
        "id",
        "subjectId",
        "subjectName",
        "canView",
        "canDownload",
        "expiresAt",
        "voucherFileName",
        "voucherFileSize",
        "voucherUploadedAt",
    }
    sub = b6_client.get(
        f"/api/v1/documents/{ids['sub']}/permissions",
        headers=_headers(login_token),
    )
    sub_data = sub.get_json()["data"]
    assert sub_data["documentId"] == str(ids["sub"])
    assert sub_data["sourceDocumentId"] == str(ids["main_custom"])
    assert sub_data["inherited"] is True
    assert sub_data["entries"] == main_data["entries"]
    attachment = b6_client.get(
        f"/api/v1/documents/{ids['attachment_one']}/permissions",
        headers=_headers(login_token),
    )
    assert attachment.status_code == 400
    assert attachment.get_json()["code"] == "ATTACHMENT_HAS_NO_ACL"


@pytest.mark.parametrize("username", ["user"])
def test_permission_management_requires_admin(
    username, b6_app, b6_client, token_for
):
    response = b6_client.get(
        f"/api/v1/documents/{_ids(b6_app)['main_custom']}/permissions",
        headers=_headers(token_for(username)),
    )
    assert response.status_code == 403


def test_permission_save_multiple_users_and_stable_sort(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import User

    ids = _ids(b6_app)
    with b6_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        user = db.session.scalar(db.select(User).where(User.username == "user"))
        assert admin and user
        admin_id, user_id = admin.id, user.id
    response = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        [
            _permission(user_id, canDownload=False),
            _permission(admin_id),
        ],
    )
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["documentRowVersion"] == 1
    assert "visibilityType" not in data
    # 条目按借阅用户ID升序稳定排序
    assert [entry["subjectId"] for entry in data["entries"]] == [
        str(min(admin_id, user_id)),
        str(max(admin_id, user_id)),
    ]


def test_permission_save_rejects_malformed_entries(b6_app, b6_client, login_token):
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        [{"bad": True}],
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.parametrize("invalid_boolean", [0, 1, "true", None])
def test_permission_actions_are_strict_booleans(
    invalid_boolean, b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import User

    with b6_app.app_context():
        user_id = db.session.scalar(
            db.select(User.id).where(User.username == "user")
        )
    entry = _permission(user_id)
    entry["canView"] = invalid_boolean
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        [entry],
    )
    assert response.status_code == 400


def test_permission_duplicate_subject(b6_app, b6_client, login_token):
    from dms.extensions import db
    from dms.models import User

    with b6_app.app_context():
        user_id = db.session.scalar(
            db.select(User.id).where(User.username == "user")
        )
    duplicate = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        [_permission(user_id), _permission(user_id)],
    )
    assert duplicate.status_code == 400
    assert duplicate.get_json()["code"] == "DUPLICATE_PERMISSION_SUBJECT"


def test_permission_save_rejects_invalid_subject_and_unknown_fields(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    invalid = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        [_permission(99999999)],
    )
    assert invalid.status_code == 404
    entry = _permission(1)
    entry["subjectName"] = "客户端伪造"
    unknown = _save(
        b6_client, login_token, ids["main_all"], 0, [entry]
    )
    assert unknown.status_code == 400


def test_permission_rejects_disabled_subjects(b6_app, b6_client, login_token):
    from dms.extensions import db
    from dms.models import User

    with b6_app.app_context():
        subject_id = db.session.scalar(
            db.select(User.id).where(User.username == "disabled")
        )
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        [_permission(subject_id)],
    )
    assert response.status_code == 404


def test_permission_sub_save_conflict_attachment_save_error_and_version_conflict(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    sub = _save(b6_client, login_token, ids["sub"], 0, [])
    assert sub.status_code == 409
    assert sub.get_json()["code"] == "SUB_PLAN_PERMISSION_INHERITED"
    attachment = _save(
        b6_client,
        login_token,
        ids["attachment_one"],
        0,
        [],
    )
    assert attachment.status_code == 400
    assert attachment.get_json()["code"] == "ATTACHMENT_HAS_NO_ACL"
    conflict = _save(b6_client, login_token, ids["main_all"], 99, [])
    assert conflict.status_code == 409
    assert conflict.get_json()["details"]["currentRowVersion"] == 0


def test_permission_full_set_logical_delete_and_readd_creates_new_record(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import Permission, User

    ids = _ids(b6_app)
    with b6_app.app_context():
        user_id = db.session.scalar(
            db.select(User.id).where(User.username == "user")
        )
        old_id = db.session.scalar(
            db.select(Permission.id).where(
                Permission.document_id == ids["main_custom"],
                Permission.subject_id == user_id,
                Permission.is_deleted.is_(False),
            )
        )
    removed = _save(b6_client, login_token, ids["main_custom"], 0, [])
    assert removed.status_code == 200
    assert removed.get_json()["data"]["entries"] == []
    readded = _save(
        b6_client,
        login_token,
        ids["main_custom"],
        1,
        [_permission(user_id)],
    )
    assert readded.status_code == 200
    new_id = int(readded.get_json()["data"]["entries"][0]["id"])
    assert new_id != old_id
    with b6_app.app_context():
        old = db.session.get(Permission, old_id)
        assert old.is_deleted is True


def test_permission_change_audit_contains_diff_and_actor_snapshots(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import AuditLog, User

    ids = _ids(b6_app)
    with b6_app.app_context():
        user = db.session.scalar(db.select(User).where(User.username == "user"))
    expires = _expires_at(3)
    response = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        [_permission(user.id, expiresAt=expires)],
    )
    assert response.status_code == 200
    with b6_app.app_context():
        audit = db.session.scalar(
            db.select(AuditLog).where(
                AuditLog.action_type == "CHANGE_PERMISSION",
                AuditLog.target_id == ids["main_all"],
            )
        )
        assert audit is not None
        assert audit.username == "admin"
        assert audit.real_name == "系统管理员"
        assert audit.organization_name == "机关"
        assert audit.request_id
        assert "visibilityTypeBefore" not in audit.operation_detail
        assert "visibilityTypeAfter" not in audit.operation_detail
        # 种子借阅为7天，本次缩短为3天，仅时限变化记入updated
        assert audit.operation_detail["added"] == []
        assert audit.operation_detail["updated"] == [
            {
                "subjectId": str(user.id),
                "subjectName": "普通用户",
                "canView": True,
                "canDownload": True,
                "expiresAt": expires,
            }
        ]
        assert audit.operation_detail["deleted"] == []
        assert audit.operation_detail["documentRowVersion"] == 1


def test_permission_audit_failure_rolls_back(
    monkeypatch, b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import Document, Permission
    import dms.services.permission_service as service

    ids = _ids(b6_app)
    monkeypatch.setattr(
        service, "business_audit", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError())
    )
    response = _save(b6_client, login_token, ids["main_custom"], 0, [])
    assert response.status_code == 500
    with b6_app.app_context():
        document = db.session.get(Document, ids["main_custom"])
        assert document.row_version == 0
        assert (
            db.session.scalar(
                db.select(db.func.count(Permission.id)).where(
                    Permission.document_id == ids["main_custom"],
                    Permission.is_deleted.is_(False),
                )
            )
            == 1
        )


def test_permission_authorization_linkage_and_dynamic_sub_inheritance(
    b6_app, b6_client, login_token, token_for
):
    from dms.extensions import db
    from dms.models import Document, User

    ids = _ids(b6_app)
    user_token = token_for("user")
    with b6_app.app_context():
        user_id = db.session.scalar(
            db.select(User.id).where(User.username == "user")
        )
        main = db.session.get(Document, ids["main_custom"])
        relative = "original/b6/linkage.pdf"
        path = Path(b6_app.config["DMS_STORAGE_ROOT"]) / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(b"%PDF-1.4\nB6\n%%EOF")
        main.file_relative_path = relative
        main.original_file_name = "B6联动.pdf"
        main.file_extension = "pdf"
        main.mime_type = "application/pdf"
        sub = db.session.get(Document, ids["sub"])
        sub.file_relative_path = relative
        sub.original_file_name = "B6子案.pdf"
        sub.file_extension = "pdf"
        sub.mime_type = "application/pdf"
        db.session.commit()

    denied = _save(
        b6_client,
        login_token,
        ids["main_custom"],
        0,
        [_permission(user_id, canView=True, canDownload=False)],
    )
    assert denied.status_code == 200
    detail = b6_client.get(
        f"/api/v1/documents/{ids['main_custom']}",
        headers=_headers(user_token),
    )
    assert detail.status_code == 200
    assert detail.get_json()["data"]["allowedActions"] == ["VIEW"]
    download = b6_client.get(
        f"/api/v1/documents/{ids['main_custom']}/download",
        headers=_headers(user_token),
    )
    assert download.status_code == 403
    sub_permissions = b6_client.get(
        f"/api/v1/documents/{ids['sub']}/permissions",
        headers=_headers(login_token),
    )
    assert sub_permissions.get_json()["data"]["documentRowVersion"] == 1

    allowed = _save(
        b6_client,
        login_token,
        ids["main_custom"],
        1,
        [_permission(user_id, canView=True, canDownload=True)],
    )
    assert allowed.status_code == 200
    download = b6_client.get(
        f"/api/v1/documents/{ids['sub']}/download",
        headers=_headers(user_token),
    )
    assert download.status_code == 200


def test_borrowed_plan_and_mounted_attachment_are_visible_to_user(
    b6_app, b6_client, token_for
):
    ids = _ids(b6_app)
    # 种子为main_all配置了user的生效借阅：主案与挂载附件对借阅用户均可见
    assert (
        b6_client.get(
            f"/api/v1/documents/{ids['main_all']}",
            headers=_headers(token_for("user")),
        ).status_code
        == 200
    )
    assert (
        b6_client.get(
            f"/api/v1/attachments/{ids['attachment_one']}",
            headers=_headers(token_for("user")),
        ).status_code
        == 200
    )


def test_admin_cannot_bypass_security_for_b6(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import User

    ids = _ids(b6_app)
    with b6_app.app_context():
        admin = db.session.scalar(db.select(User).where(User.username == "admin"))
        admin.security_level = "PUBLIC"
        db.session.commit()
    permission = b6_client.get(
        f"/api/v1/documents/{ids['main_custom']}/permissions",
        headers=_headers(login_token),
    )
    assert permission.status_code == 403
    assert permission.get_json()["code"] == "SECURITY_LEVEL_FORBIDDEN"
    binding = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={
            "attachmentIds": [str(ids["attachment_one"])],
            "mainPlanRowVersion": 0,
        },
        headers=_headers(login_token),
    )
    assert binding.status_code == 403


def test_b6_cors_request_ids_and_ai_scope(
    b6_app, b6_client, login_token
):
    ids = _ids(b6_app)
    success = b6_client.get(
        f"/api/v1/documents/{ids['main_custom']}/permissions",
        headers=_headers(login_token, origin=True),
    )
    assert success.headers["Access-Control-Expose-Headers"] == (
        "Content-Disposition, X-Request-Id"
    )
    assert success.headers["X-Request-Id"] == success.get_json()["requestId"]
    error = b6_client.post(
        f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
        json={"attachmentIds": [], "mainPlanRowVersion": 0},
        headers=_headers(login_token, origin=True),
    )
    assert error.headers["Access-Control-Expose-Headers"] == (
        "Content-Disposition, X-Request-Id"
    )
    assert error.headers["X-Request-Id"] == error.get_json()["requestId"]
    ai = b6_client.get(
        "/api/health", headers={"Origin": "http://127.0.0.1:9346"}
    )
    assert "Access-Control-Expose-Headers" not in ai.headers


def test_concurrent_same_version_bind_has_one_effect(
    b6_app, login_token
):
    ids = _ids(b6_app)

    def call():
        with b6_app.test_client() as client:
            return client.post(
                f"/api/v1/main-plans/{ids['main_custom']}/attachments/bind",
                json={
                    "attachmentIds": [str(ids["attachment_one"])],
                    "mainPlanRowVersion": 0,
                },
                headers=_headers(login_token),
            ).status_code

    with ThreadPoolExecutor(max_workers=2) as executor:
        statuses = sorted(executor.map(lambda _value: call(), range(2)))
    assert statuses == [200, 409]


def test_concurrent_permission_save_does_not_silently_overwrite(
    b6_app, login_token
):
    ids = _ids(b6_app)

    def call(_value):
        with b6_app.test_client() as client:
            return _save(client, login_token, ids["main_all"], 0, []).status_code

    with ThreadPoolExecutor(max_workers=2) as executor:
        statuses = sorted(executor.map(call, range(2)))
    assert statuses == [200, 409]