KiloStar/tests/unit/test_api_custom_toolset_auth.py

"""``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。"""
    import kilostar.utils.access as access_mod

    app = FastAPI()
    app.include_router(resource_router)
    app.dependency_overrides[Accessor.get_current_user] = lambda: _fake_user("alice")

    async def _default_authority(uid):
        return UserAuthority.USER

    monkeypatch.setattr(access_mod, "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

    import kilostar.utils.access as access_mod
    monkeypatch.setattr(access_mod, "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()