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

from __future__ import annotations

import time

import pytest

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


def _dataset_ids(items: list[dict]) -> set[int]:
    return {int(item["id"]) for item in items if item.get("id") is not None}


def _app_ids(items: list[dict]) -> set[str]:
    return {str(item["appId"]) for item in items if item.get("appId") is not None}


@pytest.mark.release
def test_g5_rag_dataset_admin_scope_is_limited_to_own_tenant(
    admin_client: ReleaseApiClient,
    tenant_admin_api: ReleaseApiClient,
    tenant_admin_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_b: TenantSeed,
    make_release_dataset,
) -> None:
    dataset_a = make_release_dataset(
        tenant=tenant_a,
        dataset_name=f"Pytest G5 A {int(time.time())}",
    )
    dataset_b = make_release_dataset(
        tenant=tenant_b,
        dataset_name=f"Pytest G5 B {int(time.time())}",
    )

    global_admin_list = ReleaseApiClient.json_data(admin_client.get("/api/v3/rag/datasets/admin?page=1&pageSize=100"))
    ids_a = _dataset_ids(global_admin_list["data"])
    assert dataset_a.dataset_id in ids_a
    assert dataset_b.dataset_id in ids_a

    tenant_admin_list = ReleaseApiClient.json_data(
        tenant_admin_api.get("/api/v3/rag/datasets/admin?page=1&pageSize=100")
    )
    tenant_admin_ids = _dataset_ids(tenant_admin_list["data"])
    assert dataset_a.dataset_id in tenant_admin_ids
    assert dataset_b.dataset_id not in tenant_admin_ids

    forbidden_query = tenant_admin_api.get(
        f"/api/v3/rag/datasets/admin?page=1&pageSize=100&tenant_code={tenant_b.tenant_code}",
        expected_status=403,
    )
    assert "本地区知识库配置" in forbidden_query.text or "本租户知识库" in forbidden_query.text


@pytest.mark.release
def test_g5_rag_dataset_detail_and_update_respect_tenant_boundary(
    admin_client: ReleaseApiClient,
    tenant_admin_api: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_b: TenantSeed,
    make_release_dataset,
) -> None:
    dataset_a = make_release_dataset(
        tenant=tenant_a,
        dataset_name=f"Pytest G5 Detail A {int(time.time())}",
    )
    dataset_b = make_release_dataset(
        tenant=tenant_b,
        dataset_name=f"Pytest G5 Detail B {int(time.time())}",
    )

    own_detail = ReleaseApiClient.json_data(tenant_admin_api.get(f"/api/v3/rag/datasets/{dataset_a.dataset_id}"))
    assert int(own_detail["id"]) == dataset_a.dataset_id
    assert str(own_detail.get("tenantCode") or "") == tenant_a.tenant_code

    hidden_detail = ReleaseApiClient.json_data(tenant_admin_api.get(f"/api/v3/rag/datasets/{dataset_b.dataset_id}"))
    assert hidden_detail is None

    own_update = ReleaseApiClient.json_data(
        tenant_admin_api.patch(
            f"/api/v3/rag/datasets/{dataset_a.dataset_id}",
            json={"name": f"{dataset_a.dataset_name}-tenant-updated"},
            expected_status=200,
        )
    )
    assert str(own_update["name"]).endswith("-tenant-updated")

    hidden_update = ReleaseApiClient.json_data(
        tenant_admin_api.patch(
            f"/api/v3/rag/datasets/{dataset_b.dataset_id}",
            json={"name": "should-not-work"},
            expected_status=200,
        )
    )
    assert hidden_update is None

    admin_update = ReleaseApiClient.json_data(
        admin_client.patch(
            f"/api/v3/rag/datasets/{dataset_a.dataset_id}",
            json={"name": f"{dataset_a.dataset_name}-updated"},
            expected_status=200,
        )
    )
    assert str(admin_update["name"]).endswith("-updated")


@pytest.mark.release
def test_g5_rag_apps_and_public_dataset_visibility(
    tenant_admin_api: ReleaseApiClient,
    tenant_admin_api_b: ReleaseApiClient,
    common_api_a: ReleaseApiClient,
    common_api_b: ReleaseApiClient,
    tenant_a: TenantSeed,
    tenant_b: TenantSeed,
    make_release_dataset,
) -> None:
    dataset_a = make_release_dataset(
        tenant=tenant_a,
        dataset_name=f"Pytest G5 App A {int(time.time())}",
        is_default=True,
    )
    dataset_b = make_release_dataset(
        tenant=tenant_b,
        dataset_name=f"Pytest G5 App B {int(time.time())}",
        is_default=True,
    )
    public_dataset = make_release_dataset(
        tenant=tenant_a,
        dataset_name=f"Pytest G5 Public {int(time.time())}",
        is_public=True,
        is_default=False,
    )

    apps_a = ReleaseApiClient.json_data(common_api_a.get("/api/v3/rag/apps"))
    app_ids_a = _app_ids(apps_a["data"])
    assert dataset_a.app_id is not None
    assert str(dataset_a.app_id) in app_ids_a
    assert dataset_b.app_id is not None
    assert str(dataset_b.app_id) not in app_ids_a
    assert public_dataset.app_id is not None
    assert str(public_dataset.app_id) in app_ids_a

    apps_b = ReleaseApiClient.json_data(common_api_b.get("/api/v3/rag/apps"))
    app_ids_b = _app_ids(apps_b["data"])
    assert str(dataset_b.app_id) in app_ids_b
    assert str(dataset_a.app_id) not in app_ids_b
    assert str(public_dataset.app_id) in app_ids_b

    default_app_a = ReleaseApiClient.json_data(common_api_a.get("/api/v3/rag/apps/default"))
    assert default_app_a is not None
    assert str(default_app_a.get("tenantCode") or "") in {tenant_a.tenant_code, public_dataset.tenant_code}


@pytest.mark.release
def test_g5_common_user_cannot_access_rag_admin_endpoints(
    common_api_a: ReleaseApiClient,
    tenant_a: TenantSeed,
) -> None:
    forbidden_admin_list = common_api_a.get("/api/v3/rag/datasets/admin?page=1&pageSize=20", expected_status=403)
    assert "管理知识库权限" in forbidden_admin_list.text

    forbidden_create = common_api_a.post(
        "/api/v3/rag/datasets/admin",
        json={
            "tenant_code": tenant_a.tenant_code,
            "tenant_name": tenant_a.tenant_name,
            "area": tenant_a.tenant_name,
            "name": f"should-forbid-{int(time.time())}",
            "description": "forbidden",
            "status": 1,
        },
        expected_status=403,
    )
    assert "创建知识库权限" in forbidden_create.text