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

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
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 _permission(subject_type: str, subject_id: int, **actions):
    values = {
        "canView": True,
        "canDownload": True,
        "canEdit": False,
        "canManagePermission": False,
        "canDelete": False,
    }
    values.update(actions)
    return {
        "subjectType": subject_type,
        "subjectId": str(subject_id),
        **values,
    }


def _save(client, token, document_id, version, visibility, entries):
    return client.put(
        f"/api/v1/documents/{document_id}/permissions",
        json={
            "visibilityType": visibility,
            "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",
        "subjectType",
        "subjectId",
        "subjectName",
        *{
            "canView",
            "canDownload",
            "canEdit",
            "canManagePermission",
            "canDelete",
        },
    }
    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_custom_mixed_and_stable_sort(
    b6_app, b6_client, login_token
):
    from dms.extensions import db
    from dms.models import Organization, User

    ids = _ids(b6_app)
    with b6_app.app_context():
        root = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        user = db.session.scalar(db.select(User).where(User.username == "user"))
        assert root and user
        root_id, user_id = root.id, user.id
    response = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        "CUSTOM",
        [
            _permission("USER", user_id, canEdit=True),
            _permission("ORG", root_id, canDownload=False),
        ],
    )
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["documentRowVersion"] == 1
    assert data["visibilityType"] == "CUSTOM"
    assert [(entry["subjectType"], entry["subjectId"]) for entry in data["entries"]] == [
        ("ORG", str(root_id)),
        ("USER", str(user_id)),
    ]


@pytest.mark.parametrize(
    ("visibility", "entries", "code"),
    [
        ("ALL_AUTHENTICATED", [{"bad": True}], "INVALID_ARGUMENT"),
        ("CUSTOM", [{"bad": True}], "INVALID_ARGUMENT"),
    ],
)
def test_permission_save_rejects_malformed_entries(
    visibility, entries, code, b6_app, b6_client, login_token
):
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        visibility,
        entries,
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == code


@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 Organization

    with b6_app.app_context():
        root_id = db.session.scalar(
            db.select(Organization.id).where(Organization.org_code == "ORG_ROOT")
        )
    entry = _permission("ORG", root_id)
    entry["canView"] = invalid_boolean
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        "CUSTOM",
        [entry],
    )
    assert response.status_code == 400


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

    with b6_app.app_context():
        org_id = db.session.scalar(
            db.select(Organization.id).where(Organization.org_code == "ORG_ROOT")
        )
        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,
        "CUSTOM",
        [_permission("ORG", org_id), _permission("ORG", org_id)],
    )
    assert duplicate.status_code == 400
    assert duplicate.get_json()["code"] == "DUPLICATE_PERMISSION_SUBJECT"
    all_nonempty = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        "ALL_AUTHENTICATED",
        [_permission("ORG", org_id)],
    )
    assert all_nonempty.status_code == 400
    org_with_user = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        "ORGANIZATION",
        [_permission("USER", user_id)],
    )
    assert org_with_user.status_code == 400
    empty_organization = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        "ORGANIZATION",
        [],
    )
    assert empty_organization.status_code == 400


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,
        "CUSTOM",
        [_permission("ORG", 99999999)],
    )
    assert invalid.status_code == 404
    entry = _permission("ORG", 1)
    entry["subjectName"] = "客户端伪造"
    unknown = _save(
        b6_client, login_token, ids["main_all"], 0, "CUSTOM", [entry]
    )
    assert unknown.status_code == 400


@pytest.mark.parametrize(
    ("subject_type", "lookup_model", "lookup_field", "lookup_value"),
    [
        ("ORG", "Organization", "org_code", "ORG_DISABLED"),
        ("USER", "User", "username", "disabled"),
    ],
)
def test_permission_rejects_disabled_subjects(
    subject_type,
    lookup_model,
    lookup_field,
    lookup_value,
    b6_app,
    b6_client,
    login_token,
):
    from dms.extensions import db
    from dms.models import Organization, User

    model = Organization if lookup_model == "Organization" else User
    with b6_app.app_context():
        subject_id = db.session.scalar(
            db.select(model.id).where(
                getattr(model, lookup_field) == lookup_value
            )
        )
    response = _save(
        b6_client,
        login_token,
        _ids(b6_app)["main_all"],
        0,
        "CUSTOM",
        [_permission(subject_type, 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, "ALL_AUTHENTICATED", []
    )
    assert sub.status_code == 409
    assert sub.get_json()["code"] == "SUB_PLAN_PERMISSION_INHERITED"
    attachment = _save(
        b6_client,
        login_token,
        ids["attachment_one"],
        0,
        "ALL_AUTHENTICATED",
        [],
    )
    assert attachment.status_code == 400
    assert attachment.get_json()["code"] == "ATTACHMENT_HAS_NO_ACL"
    conflict = _save(
        b6_client, login_token, ids["main_all"], 99, "ALL_AUTHENTICATED", []
    )
    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_type == "USER",
                Permission.subject_id == user_id,
                Permission.is_deleted.is_(False),
            )
        )
    removed = _save(
        b6_client,
        login_token,
        ids["main_custom"],
        0,
        "ALL_AUTHENTICATED",
        [],
    )
    assert removed.status_code == 200
    assert removed.get_json()["data"]["entries"] == []
    readded = _save(
        b6_client,
        login_token,
        ids["main_custom"],
        1,
        "CUSTOM",
        [_permission("USER", 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, Organization

    ids = _ids(b6_app)
    with b6_app.app_context():
        root_id = db.session.scalar(
            db.select(Organization.id).where(Organization.org_code == "ORG_ROOT")
        )
    response = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        "ORGANIZATION",
        [_permission("ORG", root_id)],
    )
    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 audit.operation_detail["visibilityTypeBefore"] == (
            "ALL_AUTHENTICATED"
        )
        assert audit.operation_detail["visibilityTypeAfter"] == "ORGANIZATION"
        assert audit.operation_detail["added"] == [
            {"subjectType": "ORG", "subjectId": str(root_id)}
        ]
        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,
        "ALL_AUTHENTICATED",
        [],
    )
    assert response.status_code == 500
    with b6_app.app_context():
        document = db.session.get(Document, ids["main_custom"])
        assert document.row_version == 0
        assert document.visibility_type == "CUSTOM"
        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, Organization

    ids = _ids(b6_app)
    user_token = token_for("user")
    with b6_app.app_context():
        ops_id = db.session.scalar(
            db.select(Organization.id).where(Organization.org_code == "ORG_OPS")
        )
        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,
        "CUSTOM",
        [_permission("ORG", ops_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,
        "ORGANIZATION",
        [_permission("ORG", ops_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_all_authenticated_plan_and_mounted_attachment_are_visible_to_user(
    b6_app, b6_client, login_token, token_for
):
    ids = _ids(b6_app)
    saved = _save(
        b6_client,
        login_token,
        ids["main_all"],
        0,
        "ALL_AUTHENTICATED",
        [],
    )
    assert saved.status_code == 200
    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(visibility):
        with b6_app.test_client() as client:
            return _save(
                client,
                login_token,
                ids["main_all"],
                0,
                visibility,
                [],
            ).status_code

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