leaudit-platform-backend/tests/release/test_g5_rule_cross_review_matrix.py

from __future__ import annotations

import time

import pytest

from .conftest import SeededUser, TenantSeed
from .helpers import ReleaseApiClient


@pytest.mark.release
def test_g5_rule_sets_are_globally_readable_but_bindings_and_groups_follow_tenant_scope(
    admin_client: ReleaseApiClient,
    tenant_admin_api: ReleaseApiClient,
    common_api_a: ReleaseApiClient,
) -> None:
    admin_rule_sets = ReleaseApiClient.json_data(admin_client.get("/api/rule-sets"))
    tenant_rule_sets = ReleaseApiClient.json_data(tenant_admin_api.get("/api/rule-sets"))
    common_rule_sets = ReleaseApiClient.json_data(common_api_a.get("/api/rule-sets"))

    assert len(admin_rule_sets) > 0
    assert len(tenant_rule_sets) == len(admin_rule_sets)
    assert len(common_rule_sets) == len(admin_rule_sets)

    admin_bindings = ReleaseApiClient.json_data(admin_client.get("/api/rule-sets/bindings"))
    tenant_bindings = ReleaseApiClient.json_data(tenant_admin_api.get("/api/rule-sets/bindings"))
    common_bindings = ReleaseApiClient.json_data(common_api_a.get("/api/rule-sets/bindings"))
    assert isinstance(admin_bindings, list)
    assert len(admin_bindings) > 0
    assert tenant_bindings == []
    assert common_bindings == []

    admin_groups = admin_client.get("/api/v3/evaluation-point-groups/all")
    tenant_groups = tenant_admin_api.get("/api/v3/evaluation-point-groups/all")
    common_groups = common_api_a.get("/api/v3/evaluation-point-groups/all")
    assert ReleaseApiClient.json_data(admin_groups) != []
    assert ReleaseApiClient.json_data(tenant_groups) == []
    assert ReleaseApiClient.json_data(common_groups) == []


@pytest.mark.release
def test_g5_group_detail_and_children_follow_same_scope_boundary(
    admin_client: ReleaseApiClient,
    tenant_admin_api: ReleaseApiClient,
    common_api_a: ReleaseApiClient,
) -> None:
    admin_groups = ReleaseApiClient.json_data(admin_client.get("/api/v3/evaluation-point-groups/all"))
    assert admin_groups != []

    target_group = admin_groups[0]
    target_group_id = int(target_group["id"])

    admin_detail = ReleaseApiClient.json_data(
        admin_client.get(f"/api/v3/evaluation-point-groups/{target_group_id}")
    )
    assert int(admin_detail["id"]) == target_group_id

    admin_children = admin_client.get(
        f"/api/v3/evaluation-point-groups/{target_group_id}/children?page=1&page_size=20"
    ).json()
    assert isinstance(admin_children, dict)
    assert "data" in admin_children

    tenant_detail = tenant_admin_api.get(
        f"/api/v3/evaluation-point-groups/{target_group_id}",
        expected_status=404,
    )
    assert "规则分组不存在" in tenant_detail.text

    tenant_children = tenant_admin_api.get(
        f"/api/v3/evaluation-point-groups/{target_group_id}/children?page=1&page_size=20",
        expected_status=404,
    )
    assert "规则分组不存在" in tenant_children.text

    common_detail = common_api_a.get(
        f"/api/v3/evaluation-point-groups/{target_group_id}",
        expected_status=404,
    )
    assert "规则分组不存在" in common_detail.text


@pytest.mark.release
def test_g5_cross_review_rejects_cross_tenant_documents_and_members(
    tenant_admin_api: ReleaseApiClient,
    common_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_b: TenantSeed,
    tenant_common_user_a: SeededUser,
    tenant_common_user_b: SeededUser,
    make_release_document,
) -> None:
    doc_a = make_release_document(
        client=tenant_admin_api,
        tenant=tenant_a,
        file_name=f"pytest-cross-a-{int(time.time())}.txt",
    )
    doc_b = make_release_document(
        client=common_api_b,
        tenant=tenant_b,
        file_name=f"pytest-cross-b-{int(time.time())}.txt",
    )

    same_tenant = tenant_admin_api.post(
        "/api/v3/cross-review/tasks",
        json={
            "taskName": f"Pytest Cross Same {int(time.time())}",
            "taskType": "CITY",
            "memberUserIds": [tenant_common_user_a.user_id],
            "principalUserIds": [],
            "documentIds": [doc_a.document_id],
        },
        expected_status=200,
    )
    same_tenant_data = ReleaseApiClient.json_data(same_tenant)
    assert int(same_tenant_data["documentCount"]) == 1

    cross_doc = tenant_admin_api.post(
        "/api/v3/cross-review/tasks",
        json={
            "taskName": f"Pytest Cross Doc {int(time.time())}",
            "taskType": "CITY",
            "memberUserIds": [tenant_common_user_a.user_id],
            "principalUserIds": [],
            "documentIds": [doc_b.document_id],
        },
        expected_status=403,
    )
    assert "其他租户文档" in cross_doc.text

    cross_member = tenant_admin_api.post(
        "/api/v3/cross-review/tasks",
        json={
            "taskName": f"Pytest Cross Member {int(time.time())}",
            "taskType": "CITY",
            "memberUserIds": [tenant_common_user_b.user_id],
            "principalUserIds": [],
            "documentIds": [doc_a.document_id],
        },
        expected_status=403,
    )
    assert "其他租户用户" in cross_member.text


@pytest.mark.release
def test_g5_cross_review_task_visibility_progress_and_documents_follow_member_scope(
    tenant_admin_api: ReleaseApiClient,
    tenant_admin_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_common_user_a: SeededUser,
    make_release_document,
) -> None:
    doc_a = make_release_document(
        client=tenant_admin_api,
        tenant=tenant_a,
        file_name=f"pytest-cross-query-{int(time.time())}.txt",
    )
    created = tenant_admin_api.post(
        "/api/v3/cross-review/tasks",
        json={
            "taskName": f"Pytest Query Task {int(time.time())}",
            "taskType": "CITY",
            "memberUserIds": [tenant_common_user_a.user_id],
            "principalUserIds": [],
            "documentIds": [doc_a.document_id],
        },
        expected_status=200,
    )
    created_data = ReleaseApiClient.json_data(created)
    task_id = int(created_data["taskId"])

    queried = ReleaseApiClient.json_data(
        tenant_admin_api.post(
            "/api/v3/cross-review/tasks/query",
            json={"page": 1, "pageSize": 50, "keyword": "Pytest Query Task"},
            expected_status=200,
        )
    )
    task_ids = {int(item["taskId"]) for item in queried["items"]}
    assert task_id in task_ids

    created_item = next(item for item in queried["items"] if int(item["taskId"]) == task_id)
    evaluation_tenant_codes = {str(item.get("tenantCode") or "") for item in created_item.get("evaluationTenants") or []}
    assert tenant_a.tenant_code in evaluation_tenant_codes

    progress = ReleaseApiClient.json_data(
        tenant_admin_api.get(f"/api/v3/cross-review/tasks/{task_id}/progress", expected_status=200)
    )
    assert int(progress["taskId"]) == task_id
    assert int(progress["totalDocuments"]) == 1

    documents = ReleaseApiClient.json_data(
        tenant_admin_api.get(
            f"/api/v3/cross-review/tasks/{task_id}/documents?page=1&pageSize=20",
            expected_status=200,
        )
    )
    assert int(documents["taskId"]) == task_id
    returned_document_ids = {int(item["documentId"]) for item in documents["items"]}
    assert doc_a.document_id in returned_document_ids

    cross_tenant_query = ReleaseApiClient.json_data(
        tenant_admin_api_b.post(
            "/api/v3/cross-review/tasks/query",
            json={"page": 1, "pageSize": 50, "keyword": "Pytest Query Task"},
            expected_status=200,
        )
    )
    cross_tenant_task_ids = {int(item["taskId"]) for item in cross_tenant_query["items"]}
    assert task_id not in cross_tenant_task_ids

    cross_tenant_progress = tenant_admin_api_b.get(
        f"/api/v3/cross-review/tasks/{task_id}/progress",
        expected_status=403,
    )
    assert "当前用户不是交叉评查任务成员" in cross_tenant_progress.text

    cross_tenant_documents = tenant_admin_api_b.get(
        f"/api/v3/cross-review/tasks/{task_id}/documents?page=1&pageSize=20",
        expected_status=403,
    )
    assert "当前用户不是交叉评查任务成员" in cross_tenant_documents.text