"""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 from deerflow.persistence.business_mapping.sql import DEFAULT_BUSINESS_ROWS class _MappingStore: def __init__(self) -> None: self.calls: list[tuple[str, dict[str, object]]] = [] self.create_calls: list[dict[str, str]] = [] 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, } async def create_custom_mapping( self, *, business_code: str, mapping_type: str, ) -> dict[str, object] | None: self.create_calls.append( {"business_code": business_code, "mapping_type": mapping_type} ) return { "business_code": business_code, "label": business_code, "mapping_type": mapping_type, "is_custom": True, "agent_id": None, "agent_name": None, "chain_id": None, "chain_title": None, "sort_order": 99, } 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)), ) def test_seed_includes_independent_3q_slots() -> None: codes = [row["business_code"] for row in DEFAULT_BUSINESS_ROWS] assert codes[:6] == ["3Q", "6BF", "7BF", "8BF", "AFX", "b5-1"] assert codes[6:13] == ["3q-0", "3q-1", "3q-2", "3q-3", "3q-4", "3q-5", "3q-6"] assert DEFAULT_BUSINESS_ROWS[codes.index("b5-1")]["label"] == "任务分析" assert "6-1" in codes and "7-4" in codes assert codes.index("3q-0") < codes.index("3q-1") assert codes.index("3q-6") < codes.index("6-1") custom_codes = {row["business_code"] for row in DEFAULT_BUSINESS_ROWS if row.get("is_custom")} assert custom_codes == {"3q-1", "3q-2", "3q-3", "3q-4", "3q-5", "3q-6"} for code in custom_codes: seed = DEFAULT_BUSINESS_ROWS[codes.index(code)] assert seed["mapping_type"] == "自定义类型" assert DEFAULT_BUSINESS_ROWS[codes.index("3q-0")].get("is_custom") is not True @pytest.mark.asyncio async def test_admin_can_create_a_custom_mapping() -> None: request = _request("admin") response = await business_mapping_router.create_mapping( request, business_mapping_router.BusinessMappingCreateRequest( name="区域研判", ), ) assert response.business_code == "区域研判" assert response.label == "区域研判" assert response.mapping_type == "自定义类型" assert response.is_custom is True assert request.app.state.business_mapping_store.create_calls == [ {"business_code": "区域研判", "mapping_type": "自定义类型"} ] @pytest.mark.asyncio async def test_non_admin_cannot_create_a_custom_mapping() -> None: request = _request("user") with pytest.raises(HTTPException) as error: await business_mapping_router.create_mapping( request, business_mapping_router.BusinessMappingCreateRequest( name="区域研判", ), ) assert error.value.status_code == 403 assert request.app.state.business_mapping_store.create_calls == [] @pytest.mark.asyncio async def test_custom_mapping_name_cannot_consume_the_a_mode_suffix() -> None: request = _request("admin") with pytest.raises(HTTPException) as error: await business_mapping_router.create_mapping( request, business_mapping_router.BusinessMappingCreateRequest( name="区域研判-A", ), ) assert error.value.status_code == 422 assert request.app.state.business_mapping_store.create_calls == [] @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 == []