import pytest from beaver_gateway.mcp.types import McpServer from beaver_gateway.mcp.wrap import build_python_tool_server from fastmcp import Client from fastmcp.exceptions import ToolError from beaver_agent.hands.komodo import Komodo @pytest.fixture def server(): tool = Komodo(url="http://127.0.0.1:9", key="k", secret="s") return build_python_tool_server( McpServer.python_tool(name="komodo", tools=[tool.komodo]) ) async def test_only_listed_actions_exist(server): async with Client(server) as client: (tool,) = await client.list_tools() actions = tool.inputSchema["properties"]["action"]["enum"] assert ( "prune" not in actions and "destroy" not in actions and "terminal" not in actions ) assert {"logs", "deploy", "restart", "exec"} <= set(actions) with pytest.raises(ToolError): await client.call_tool( "komodo", {"action": "prune", "stack": "beaver-agent"} ) async def test_exec_denies_infrastructure_before_any_request(server): async with Client(server) as client: result = await client.call_tool( "komodo", {"action": "exec", "container": "dell-periphery", "command": "id"} ) assert "запрещён" in result.content[0].text async def test_missing_parameter_is_an_error(server): async with Client(server) as client: with pytest.raises(ToolError, match="stack"): await client.call_tool("komodo", {"action": "logs"}) def test_clip_keeps_head_and_tail(): k = Komodo(url="u", key="k", secret="s", max_chars=30) out = k._clip("a" * 100) assert out.startswith("a" * 20) and out.endswith("a" * 10) and "обрезано" in out def fleet(answers, calls=None): """Komodo, отвечающий заготовками; calls копит запросы к нему.""" class Fake(Komodo): async def read(self, request, params): if calls is not None: calls.append((request, params)) answer = answers.get(request, []) return answer(params) if callable(answer) else answer return Fake(url="u", key="k", secret="s") def a_stack(name, state="running", stack_id=None): return { "id": stack_id or name, "name": name, "info": {"state": state, "server_name": "dell-home"}, } def an_update(operation="DeployStack", target_id="beaver-agent", ts=1_700_000_000_000): return { "id": f"upd-{ts}", "operation": operation, "start_ts": ts, "success": True, "status": "Complete", "username": "hh", "target": {"type": "Stack", "id": target_id}, } async def test_a_short_page_is_not_the_whole_list(): # Komodo режет List* до default_pagination_limit и молчит об этом. pages = [ [a_stack(f"s{i}") for i in range(500)], [a_stack("zv-import"), a_stack("veris-runner")], ] calls = [] komodo = fleet({"ListStacks": lambda p: pages[p["page"]]}, calls) out = await komodo._stacks() assert [p["page"] for r, p in calls if r == "ListStacks"] == [0, 1] assert "zv-import" in out and "стеков: 502" in out async def test_status_says_when_the_not_running_list_is_cut(): bad = [a_stack(f"b{i}", state="unknown") for i in range(23)] komodo = fleet({"ListStacks": bad, "ListServers": []}) out = await komodo._status() assert "стеков: 23, не running: 23" in out assert "показаны 15 из 23, остальные скрыты лимитом" in out komodo = fleet({"ListStacks": bad[:3], "ListServers": []}) assert "скрыты лимитом" not in await komodo._status() async def test_both_listings_say_that_they_are_limited_by_rights(): komodo = fleet({"ListStacks": [a_stack("beaver-agent")], "ListServers": []}) assert "ограничен правами ключа" in await komodo._status() assert "ограничен правами ключа" in await komodo._stacks() async def test_updates_carry_the_shown_span_and_say_there_is_more(): page = { "updates": [an_update(ts=1_700_000_000_000 - i * 60_000) for i in range(100)] } seen = {"ListStacks": [a_stack("beaver-agent")]} komodo = fleet(seen | {"ListUpdates": page | {"next_page": 1}}) head, second, *lines = (await komodo._updates(15, None, None)).splitlines() assert head == "операций: 15, 11-14 21:59 … 11-14 22:13, раньше в журнале есть ещё" assert second == "фильтра нет: это просто хвост журнала" assert len(lines) == 15 komodo = fleet( seen | {"ListUpdates": {"updates": page["updates"][:3], "next_page": None}} ) assert (await komodo._updates(15, None, None)).startswith( "операций: 3, 11-14 22:11 … 11-14 22:13, это весь журнал" ) async def test_the_filter_goes_to_komodo_as_a_query(): calls = [] komodo = fleet( { "ListStacks": [a_stack("beaver-agent", stack_id="id-1")], "ListUpdates": { "updates": [an_update(target_id="id-1")], "next_page": None, }, }, calls, ) out = await komodo._updates(15, "deploy", "beaver") (params,) = [p for r, p in calls if r == "ListUpdates"] assert params["query"] == { "operation": {"$regex": "deploy", "$options": "i"}, "target.id": {"$in": ["id-1"]}, } assert "фильтрует Komodo: операция ~deploy; ресурс ~beaver (1 шт)" in out # имя ресурса в строке, а не id assert "beaver-agent" in out and "id-1 " not in out async def test_an_unknown_name_is_said_out_loud_instead_of_filtering_by_nothing(): komodo = fleet({"ListStacks": [a_stack("beaver-agent")]}) out = await komodo._updates(15, None, "obsidian") assert out == ( "ресурса ~obsidian ключу руки не видно: либо имя другое, либо он вне прав ключа" ) async def test_a_target_the_key_cannot_see_is_counted_and_shown_by_id(): komodo = fleet( { "ListStacks": [a_stack("beaver-agent", stack_id="id-1")], "ListUpdates": { "updates": [an_update(target_id="id-1"), an_update(target_id="gone")], "next_page": None, }, } ) out = await komodo._updates(15, None, None) assert "записей о невидимых ключу ресурсах: 1 - id вместо имени" in out assert "gone" in out async def test_the_depth_is_clamped_out_loud(): komodo = fleet({"ListUpdates": {"updates": [], "next_page": None}}) assert "запрошено 5000 записей, предел руки 200" in await komodo._updates( 5000, None, None )