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

from __future__ import annotations

import os
import time
from dataclasses import dataclass
from typing import Any, Callable

import pytest

from .helpers import ReleaseApiClient, ReleaseTestConfig


@dataclass(frozen=True)
class TenantSeed:
    tenant_code: str
    tenant_name: str


@dataclass(frozen=True)
class SeededUser:
    user_id: int
    sub: str
    username: str
    nickname: str
    role_key: str
    tenant_code: str
    token: str


@dataclass(frozen=True)
class DocumentTypeSeed:
    type_id: int
    type_code: str
    type_name: str


@dataclass(frozen=True)
class ReleaseDocumentSeed:
    document_id: int
    file_id: int | None
    type_id: int
    type_code: str
    tenant_code: str
    tenant_name: str
    file_name: str


@dataclass(frozen=True)
class ReleaseDatasetSeed:
    dataset_id: int
    dataset_name: str
    tenant_code: str
    tenant_name: str
    app_id: int | None
    app_name: str | None


def _env(name: str, default: str) -> str:
    return str(os.getenv(name, default)).strip()


@pytest.fixture(scope="session")
def release_config() -> ReleaseTestConfig:
    return ReleaseTestConfig(
        base_url=_env("LEAUDIT_TEST_BASE_URL", "http://127.0.0.1:8096").rstrip("/"),
        admin_username=_env("LEAUDIT_TEST_ADMIN_USERNAME", "000"),
        admin_password=_env("LEAUDIT_TEST_ADMIN_PASSWORD", "admin06111"),
        timeout_seconds=float(_env("LEAUDIT_TEST_TIMEOUT_SECONDS", "30")),
        tenant_a_code=_env("LEAUDIT_TEST_TENANT_A_CODE", "PTA01"),
        tenant_a_name=_env("LEAUDIT_TEST_TENANT_A_NAME", "Pytest租户A"),
        tenant_b_code=_env("LEAUDIT_TEST_TENANT_B_CODE", "PTB01"),
        tenant_b_name=_env("LEAUDIT_TEST_TENANT_B_NAME", "Pytest租户B"),
        module_name=_env("LEAUDIT_TEST_MODULE_NAME", f"Pytest发布验收入口模块-{time.time_ns()}"),
        module_path=_env("LEAUDIT_TEST_MODULE_PATH", "/documents"),
    )


@pytest.fixture(scope="session")
def anonymous_client(release_config: ReleaseTestConfig) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config)
    yield client
    client.close()


@pytest.fixture(scope="session")
def admin_auth(release_config: ReleaseTestConfig, anonymous_client: ReleaseApiClient) -> dict[str, Any]:
    return anonymous_client.login_password(release_config.admin_username, release_config.admin_password)


@pytest.fixture(scope="session")
def admin_client(release_config: ReleaseTestConfig, admin_auth: dict[str, Any]) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config, token=str(admin_auth["access_token"]))
    yield client
    client.close()


@pytest.fixture(scope="session")
def role_map(admin_client: ReleaseApiClient) -> dict[str, int]:
    response = admin_client.get("/api/v3/rbac/roles?page=1&page_size=200")
    items = ReleaseApiClient.json_data(response)["items"]
    mapping = {str(item["role_key"]): int(item["id"]) for item in items}
    for role_key in ("super_admin", "provincial_admin", "admin", "common"):
        assert role_key in mapping, f"缺少角色种子: {role_key}"
    return mapping


def _ensure_tenant(admin_client: ReleaseApiClient, tenant_code: str, tenant_name: str) -> TenantSeed:
    detail_response = admin_client.get(f"/api/v3/tenants/{tenant_code}", expected_status=None)
    body = detail_response.json()
    desired_payload = {
        "tenant_name": tenant_name,
        "tenant_short_name": tenant_name,
        "tenant_type": "CUSTOM",
        "parent_tenant_code": None,
        "display_order": 500,
        "is_public": False,
        "can_host_entry_module": True,
        "can_host_documents": True,
        "can_host_rag": True,
        "can_host_templates": True,
        "feature_keys": ["home.entry_module", "documents.upload", "rag.dataset"],
        "alias_values": [tenant_name],
        "ext": {"created_by": "pytest-release"},
    }
    if detail_response.status_code == 404:
        create_payload = {
            "tenant_code": tenant_code,
            "is_enabled": True,
            **desired_payload,
        }
        admin_client.post("/api/v3/tenants", json=create_payload, expected_status=200)
    else:
        assert detail_response.status_code == 200, body
        admin_client.put(f"/api/v3/tenants/{tenant_code}", json=desired_payload, expected_status=200)
        admin_client.patch(f"/api/v3/tenants/{tenant_code}/status", json={"is_enabled": True}, expected_status=200)
    return TenantSeed(tenant_code=tenant_code, tenant_name=tenant_name)


@pytest.fixture(scope="session")
def tenant_a(admin_client: ReleaseApiClient, release_config: ReleaseTestConfig) -> TenantSeed:
    return _ensure_tenant(admin_client, release_config.tenant_a_code, release_config.tenant_a_name)


@pytest.fixture(scope="session")
def tenant_b(admin_client: ReleaseApiClient, release_config: ReleaseTestConfig) -> TenantSeed:
    return _ensure_tenant(admin_client, release_config.tenant_b_code, release_config.tenant_b_name)


def _ensure_user(
    *,
    anonymous_client: ReleaseApiClient,
    admin_client: ReleaseApiClient,
    role_map: dict[str, int],
    sub: str,
    username: str,
    nickname: str,
    role_key: str,
    tenant: TenantSeed,
) -> SeededUser:
    login_data = anonymous_client.login_oauth(
        sub=sub,
        username=username,
        nickname=nickname,
        ou_id=f"pytest-{tenant.tenant_code.lower()}",
        ou_name=f"{tenant.tenant_name}测试组织",
        area=tenant.tenant_name,
    )
    user_info = login_data["user_info"]
    user_id = int(user_info["user_id"])
    admin_client.post(
        f"/api/v3/rbac/users/{user_id}/roles",
        json={"role_ids": [role_map[role_key]]},
        expected_status=200,
    )
    admin_client.put(
        f"/api/v3/rbac/users/{user_id}/tenant",
        json={"tenant_code": tenant.tenant_code},
        expected_status=200,
    )
    refreshed = anonymous_client.login_oauth(
        sub=sub,
        username=username,
        nickname=nickname,
        ou_id=f"pytest-{tenant.tenant_code.lower()}",
        ou_name=f"{tenant.tenant_name}测试组织",
        area=tenant.tenant_name,
    )
    refreshed_user = refreshed["user_info"]
    assert str(refreshed_user.get("tenant_code") or "") == tenant.tenant_code
    return SeededUser(
        user_id=user_id,
        sub=sub,
        username=username,
        nickname=nickname,
        role_key=role_key,
        tenant_code=tenant.tenant_code,
        token=str(refreshed["access_token"]),
    )


@pytest.fixture(scope="session")
def tenant_admin_user(
    anonymous_client: ReleaseApiClient,
    admin_client: ReleaseApiClient,
    role_map: dict[str, int],
    tenant_a: TenantSeed,
) -> SeededUser:
    return _ensure_user(
        anonymous_client=anonymous_client,
        admin_client=admin_client,
        role_map=role_map,
        sub="pytest-admin-pta01",
        username="pytest_admin_pta01",
        nickname="Pytest租户A管理员",
        role_key="admin",
        tenant=tenant_a,
    )


@pytest.fixture(scope="session")
def tenant_admin_user_b(
    anonymous_client: ReleaseApiClient,
    admin_client: ReleaseApiClient,
    role_map: dict[str, int],
    tenant_b: TenantSeed,
) -> SeededUser:
    return _ensure_user(
        anonymous_client=anonymous_client,
        admin_client=admin_client,
        role_map=role_map,
        sub="pytest-admin-ptb01",
        username="pytest_admin_ptb01",
        nickname="Pytest租户B管理员",
        role_key="admin",
        tenant=tenant_b,
    )


@pytest.fixture(scope="session")
def tenant_common_user_a(
    anonymous_client: ReleaseApiClient,
    admin_client: ReleaseApiClient,
    role_map: dict[str, int],
    tenant_a: TenantSeed,
) -> SeededUser:
    return _ensure_user(
        anonymous_client=anonymous_client,
        admin_client=admin_client,
        role_map=role_map,
        sub="pytest-common-pta01",
        username="pytest_common_pta01",
        nickname="Pytest租户A普通用户",
        role_key="common",
        tenant=tenant_a,
    )


@pytest.fixture(scope="session")
def tenant_common_user_b(
    anonymous_client: ReleaseApiClient,
    admin_client: ReleaseApiClient,
    role_map: dict[str, int],
    tenant_b: TenantSeed,
) -> SeededUser:
    return _ensure_user(
        anonymous_client=anonymous_client,
        admin_client=admin_client,
        role_map=role_map,
        sub="pytest-common-ptb01",
        username="pytest_common_ptb01",
        nickname="Pytest租户B普通用户",
        role_key="common",
        tenant=tenant_b,
    )


@pytest.fixture
def tenant_admin_api(release_config: ReleaseTestConfig, tenant_admin_user: SeededUser) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config, token=tenant_admin_user.token)
    yield client
    client.close()


@pytest.fixture
def tenant_admin_api_b(release_config: ReleaseTestConfig, tenant_admin_user_b: SeededUser) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config, token=tenant_admin_user_b.token)
    yield client
    client.close()


@pytest.fixture
def common_api_a(release_config: ReleaseTestConfig, tenant_common_user_a: SeededUser) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config, token=tenant_common_user_a.token)
    yield client
    client.close()


@pytest.fixture
def common_api_b(release_config: ReleaseTestConfig, tenant_common_user_b: SeededUser) -> ReleaseApiClient:
    client = ReleaseApiClient(release_config, token=tenant_common_user_b.token)
    yield client
    client.close()


def _find_module_by_name(admin_client: ReleaseApiClient, module_name: str) -> dict[str, Any] | None:
    response = admin_client.get(f"/api/v3/entry-modules?page=1&page_size=200&name={module_name}".replace("page_size", "page_size"), expected_status=None)
    items = ReleaseApiClient.json_data(response)["items"]
    for item in items:
        if str(item.get("name") or "") == module_name:
            return item
    return None


@pytest.fixture
def release_entry_module(
    admin_client: ReleaseApiClient,
    release_config: ReleaseTestConfig,
    tenant_b: TenantSeed,
) -> dict[str, Any]:
    module_name = release_config.module_name
    existing = _find_module_by_name(admin_client, module_name)
    payload = {
        "name": module_name,
        "description": "pytest release acceptance only",
        "path": release_config.module_path,
        "route_path": release_config.module_path,
        "tenants": [
            {
                "tenant_code": tenant_b.tenant_code,
                "tenant_name": tenant_b.tenant_name,
                "enabled": True,
                "sort_order": 1,
            }
        ],
    }
    created = False
    if existing:
        module_id = int(existing["id"])
        response = admin_client.put(f"/api/v3/entry-modules/{module_id}", json=payload, expected_status=200)
    else:
        response = admin_client.post("/api/v3/entry-modules", json=payload, expected_status=None)
        if response.status_code == 200:
            created = True
        else:
            retried = False
            for attempt in range(1, 4):
                module_name = f"{release_config.module_name}-{attempt}"
                payload["name"] = module_name
                existing = _find_module_by_name(admin_client, module_name)
                if existing:
                    module_id = int(existing["id"])
                    response = admin_client.put(f"/api/v3/entry-modules/{module_id}", json=payload, expected_status=200)
                    retried = True
                    break
                response = admin_client.post("/api/v3/entry-modules", json=payload, expected_status=None)
                if response.status_code == 200:
                    created = True
                    retried = True
                    break
            assert retried, response.text
    module = ReleaseApiClient.json_data(response)
    try:
        yield module
    finally:
        if created:
            admin_client.delete(f"/api/v3/entry-modules/{int(module['id'])}", expected_status=200)


@pytest.fixture(scope="session")
def release_document_type(admin_client: ReleaseApiClient) -> DocumentTypeSeed:
    response = admin_client.get("/api/document-types")
    items = ReleaseApiClient.json_data(response)
    assert isinstance(items, list), items
    for item in items:
        type_id = int(item.get("id") or 0)
        type_code = str(item.get("code") or "").strip()
        type_name = str(item.get("name") or "").strip()
        if type_id > 0 and type_code:
            return DocumentTypeSeed(type_id=type_id, type_code=type_code, type_name=type_name or type_code)
    pytest.skip("当前环境没有可用于发布验收的文档类型")


@pytest.fixture
def make_release_document(
    admin_client: ReleaseApiClient,
    release_document_type: DocumentTypeSeed,
) -> Callable[..., ReleaseDocumentSeed]:
    created_document_ids: list[int] = []

    def _make_document(
        *,
        client: ReleaseApiClient,
        tenant: TenantSeed,
        file_name: str,
        content: bytes | None = None,
        content_type: str = "text/plain",
    ) -> ReleaseDocumentSeed:
        payload_bytes = content or f"pytest release document for {tenant.tenant_code} / {file_name}\n".encode("utf-8")
        response = client.post(
            "/api/upload",
            data={
                "typeId": str(release_document_type.type_id),
                "typeCode": release_document_type.type_code,
                "tenant_code": tenant.tenant_code,
                "region": tenant.tenant_name,
                "fileRole": "primary",
                "autoRun": "false",
                "speed": "normal",
            },
            files={
                "file": (file_name, payload_bytes, content_type),
            },
            expected_status=200,
        )
        data = ReleaseApiClient.json_data(response)
        document_id = int(data["documentId"])
        created_document_ids.append(document_id)
        return ReleaseDocumentSeed(
            document_id=document_id,
            file_id=int(data["fileId"]) if data.get("fileId") is not None else None,
            type_id=release_document_type.type_id,
            type_code=release_document_type.type_code,
            tenant_code=tenant.tenant_code,
            tenant_name=tenant.tenant_name,
            file_name=file_name,
        )

    try:
        yield _make_document
    finally:
        for document_id in reversed(created_document_ids):
            admin_client.delete(f"/api/documents/{document_id}", expected_status=None)


@pytest.fixture
def make_release_dataset(
    admin_client: ReleaseApiClient,
) -> Callable[..., ReleaseDatasetSeed]:
    created_dataset_ids: list[int] = []

    def _make_dataset(
        *,
        tenant: TenantSeed,
        dataset_name: str,
        is_public: bool = False,
        is_default: bool = False,
    ) -> ReleaseDatasetSeed:
        response = admin_client.post(
            "/api/v3/rag/datasets/admin",
            json={
                "tenant_code": tenant.tenant_code,
                "tenant_name": tenant.tenant_name,
                "area": tenant.tenant_name,
                "name": dataset_name,
                "description": f"{dataset_name} - pytest release",
                "is_public": is_public,
                "is_default": is_default,
                "status": 1,
                "sort_order": 0,
            },
            expected_status=200,
        )
        data = ReleaseApiClient.json_data(response)
        dataset_id = int(data["id"])
        created_dataset_ids.append(dataset_id)
        return ReleaseDatasetSeed(
            dataset_id=dataset_id,
            dataset_name=str(data.get("name") or dataset_name),
            tenant_code=str(data.get("tenantCode") or tenant.tenant_code),
            tenant_name=str(data.get("tenantName") or tenant.tenant_name),
            app_id=int(data["appId"]) if data.get("appId") is not None else None,
            app_name=str(data.get("appName") or "") or None,
        )

    try:
        yield _make_dataset
    finally:
        for dataset_id in reversed(created_dataset_ids):
            admin_client.delete(f"/api/v3/rag/datasets/admin/{dataset_id}", expected_status=None)