265 lines
8.9 KiB
Python
265 lines
8.9 KiB
Python
"""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 == []
|