101 lines
3.3 KiB
Python
101 lines
3.3 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
|
|
|
|
|
|
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
|
|
|
|
|
|
def _request(role: str, chain: dict[str, object] | None = None):
|
|
return SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
business_mapping_store=_MappingStore(),
|
|
roundtable_chain_store=_ChainStore(chain),
|
|
)
|
|
),
|
|
state=SimpleNamespace(user=SimpleNamespace(id="admin-id", system_role=role)),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_administrators_can_update_global_business_mappings() -> None:
|
|
request = _request("user")
|
|
|
|
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_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 == []
|