143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
"""``api/resource.py`` Custom toolset 路由：归属鉴权。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import types
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from kilostar.api.resource import resource_router
|
|
from kilostar.core.postgres_database.model import UserAuthority
|
|
from kilostar.utils.access import Accessor, TokenData
|
|
|
|
|
|
def _fake_user(user_id: str = "alice"):
|
|
return TokenData(user_id=user_id, username=user_id)
|
|
|
|
|
|
@pytest.fixture
|
|
def app_with_user(monkeypatch):
|
|
"""挂上 resource_router；用 dependency_overrides 跳过 JWT，并把 get_authority 默认放成 USER。"""
|
|
app = FastAPI()
|
|
app.include_router(resource_router)
|
|
app.dependency_overrides[Accessor.get_current_user] = lambda: _fake_user("alice")
|
|
|
|
# 默认把权限置为 USER；具体 case 内部可再 monkeypatch 覆盖
|
|
async def _default_authority(uid):
|
|
return UserAuthority.USER
|
|
|
|
monkeypatch.setattr(
|
|
"kilostar.utils.check_user.role_check.get_authority", _default_authority
|
|
)
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_custom_toolset_forbidden_for_non_owner(
|
|
app_with_user, fake_actors
|
|
):
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_custom_toolset = types.SimpleNamespace(
|
|
remote=AsyncMock(
|
|
return_value={"toolset_id": "t1", "owner_id": "bob", "tools": []}
|
|
)
|
|
)
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=app_with_user)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/api/v1/resource/custom-toolset/t1")
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_custom_toolset_allowed_for_owner(app_with_user, fake_actors):
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_custom_toolset = types.SimpleNamespace(
|
|
remote=AsyncMock(
|
|
return_value={"toolset_id": "t1", "owner_id": "alice", "tools": []}
|
|
)
|
|
)
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=app_with_user)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/api/v1/resource/custom-toolset/t1")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["owner_id"] == "alice"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_custom_toolset_allowed_for_admin(
|
|
app_with_user, fake_actors, monkeypatch
|
|
):
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_custom_toolset = types.SimpleNamespace(
|
|
remote=AsyncMock(
|
|
return_value={"toolset_id": "t1", "owner_id": "bob", "tools": []}
|
|
)
|
|
)
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
async def _admin(uid):
|
|
return UserAuthority.SUPER_ADMINISTRATOR
|
|
|
|
monkeypatch.setattr(
|
|
"kilostar.utils.check_user.role_check.get_authority", _admin
|
|
)
|
|
|
|
transport = ASGITransport(app=app_with_user)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/api/v1/resource/custom-toolset/t1")
|
|
|
|
assert resp.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_custom_toolsets_filters_by_owner(app_with_user, fake_actors):
|
|
all_sets = [
|
|
{"toolset_id": "t1", "owner_id": "alice"},
|
|
{"toolset_id": "t2", "owner_id": "bob"},
|
|
]
|
|
gsm = types.SimpleNamespace()
|
|
gsm.list_custom_toolsets = types.SimpleNamespace(
|
|
remote=AsyncMock(return_value=all_sets)
|
|
)
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=app_with_user)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/api/v1/resource/custom-toolset")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert len(body["toolsets"]) == 1
|
|
assert body["toolsets"][0]["toolset_id"] == "t1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_custom_toolset_forbidden_for_non_owner(
|
|
app_with_user, fake_actors
|
|
):
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_custom_toolset = types.SimpleNamespace(
|
|
remote=AsyncMock(
|
|
return_value={"toolset_id": "t1", "owner_id": "bob"}
|
|
)
|
|
)
|
|
delete_mock = AsyncMock(return_value=True)
|
|
gsm.delete_custom_toolset = types.SimpleNamespace(remote=delete_mock)
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=app_with_user)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.delete("/api/v1/resource/custom-toolset/t1")
|
|
|
|
assert resp.status_code == 403
|
|
delete_mock.assert_not_called()
|