deerflow-code/offline-backend-20260512/backend/tests/test_business_mapping.py

"""Authorization and shared-chain invariants for global business mappings."""

from __future__ import annotations

from types import SimpleNamespace

import pytest
from fastapi import HTTPException

from app.gateway.routers import business_mapping as business_mapping_router


class _MappingStore:
    def __init__(self) -> None:
        self.calls: list[tuple[str, dict[str, object]]] = []

    async def update_mapping(self, business_code: str, **fields: object) -> dict[str, object]:
        self.calls.append((business_code, fields))
        return {
            "business_code": business_code,
            "label": "岗位多智能体会商",
            "agent_id": None,
            "agent_name": None,
            "chain_id": fields.get("chain_id"),
            "chain_title": fields.get("chain_title"),
            "sort_order": 3,
        }


class _ChainStore:
    def __init__(self, chain: dict[str, object] | None) -> None:
        self.chain = chain

    async def get_chain(self, chain_id: str, user_id: str, *, is_admin: bool) -> dict[str, object] | None:
        assert chain_id == "chain-8bf"
        assert user_id == "admin-id"
        assert is_admin is True
        return self.chain


class _AgentStore:
    def __init__(self, agents: dict[str, dict[str, object]] | None = None) -> None:
        self.agents = agents or {}

    async def get_any(self, agent_id: str) -> dict[str, object] | None:
        return self.agents.get(agent_id)


def _request(
    role: str,
    chain: dict[str, object] | None = None,
    *,
    email: str = "lqq@example.com",
    agents: dict[str, dict[str, object]] | None = None,
):
    return SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(
                business_mapping_store=_MappingStore(),
                roundtable_chain_store=_ChainStore(chain),
                agent_store=_AgentStore(agents),
            )
        ),
        state=SimpleNamespace(user=SimpleNamespace(id="admin-id", email=email, system_role=role)),
    )


@pytest.mark.asyncio
async def test_only_administrators_can_update_non_8bf_business_mappings() -> None:
    request = _request("user")

    with pytest.raises(HTTPException) as error:
        await business_mapping_router.update_mapping(
            request,
            "6BF",
            business_mapping_router.BusinessMappingUpdateRequest(chain_id="chain-8bf"),
        )

    assert error.value.status_code == 403
    assert request.app.state.business_mapping_store.calls == []


@pytest.mark.asyncio
async def test_only_lqq_can_update_the_8bf_business_mapping() -> None:
    request = _request("admin", email="admin@example.com")

    with pytest.raises(HTTPException) as error:
        await business_mapping_router.update_mapping(
            request,
            "8BF",
            business_mapping_router.BusinessMappingUpdateRequest(chain_id="chain-8bf"),
        )

    assert error.value.status_code == 403
    assert request.app.state.business_mapping_store.calls == []


@pytest.mark.asyncio
async def test_lqq_can_update_the_8bf_business_mapping_without_admin_role() -> None:
    request = _request("user", email="lqq@example.com")

    response = await business_mapping_router.update_mapping(
        request,
        "8BF",
        business_mapping_router.BusinessMappingUpdateRequest(),
    )

    assert response.business_code == "8BF"
    assert request.app.state.business_mapping_store.calls == [("8BF", {})]


@pytest.mark.asyncio
async def test_global_mapping_requires_a_public_chain_and_canonicalizes_its_title() -> None:
    request = _request("admin", {"id": "chain-8bf", "title": "岗位协同链", "is_public": True})

    response = await business_mapping_router.update_mapping(
        request,
        "8BF",
        business_mapping_router.BusinessMappingUpdateRequest(
            chain_id="chain-8bf",
            chain_title="客户端伪造的标题",
        ),
    )

    assert response.chain_id == "chain-8bf"
    assert response.chain_title == "岗位协同链"
    assert request.app.state.business_mapping_store.calls == [("8BF", {"chain_id": "chain-8bf", "chain_title": "岗位协同链"})]


@pytest.mark.asyncio
async def test_global_mapping_rejects_a_private_chain() -> None:
    request = _request("admin", {"id": "chain-8bf", "title": "私有链", "is_public": False})

    with pytest.raises(HTTPException) as error:
        await business_mapping_router.update_mapping(
            request,
            "8BF",
            business_mapping_router.BusinessMappingUpdateRequest(chain_id="chain-8bf"),
        )

    assert error.value.status_code == 400
    assert request.app.state.business_mapping_store.calls == []


@pytest.mark.asyncio
async def test_global_mapping_rejects_a_chain_with_an_unpublished_agent_seat() -> None:
    request = _request(
        "admin",
        {
            "id": "chain-8bf",
            "title": "岗位协同链",
            "is_public": True,
            "seats": [
                {"agent_id": "published", "name": "已发布智能体"},
                {"agent_id": "private", "name": "未发布智能体"},
            ],
        },
        agents={
            "published": {"id": "published", "user_id": "owner", "published": True, "is_private": False},
            "private": {"id": "private", "user_id": "owner", "published": False, "is_private": False},
        },
    )

    with pytest.raises(HTTPException) as error:
        await business_mapping_router.update_mapping(
            request,
            "8BF",
            business_mapping_router.BusinessMappingUpdateRequest(chain_id="chain-8bf"),
        )

    assert error.value.status_code == 400
    assert "未发布智能体" in error.value.detail
    assert request.app.state.business_mapping_store.calls == []