Phase 4: corp organization reads (#32)
Add read-only corp tools: members, member tracking, divisions, titles, medals, standings (corp-level, need corp roles), plus the character's own corp roles. Closes #32. 95 tests, coverage 91%. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
3 files changed
+181
No files matched your search
@@ -0,0 +1,109 @@
|
|||||||
|
"""Read-only corporation organization tools.
|
||||||
|
|
||||||
|
Membership, member tracking, divisions, titles, medals, standings (corp-level,
|
||||||
|
need the relevant corp role) plus the character's own corp roles. Keyed off each
|
||||||
|
logged-in character's corporation.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastmcp import FastMCP
|
||||||
|
|
||||||
|
from .charutil import run_per_character, run_per_corp, summarize
|
||||||
|
from .esi import client
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_members(character: str | int | None = None, limit: int = 50) -> dict[str, Any]:
|
||||||
|
"""Corp member character ids (needs a corp role) for one character's corp or all."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
members = await client().get(f"/corporations/{corp_id}/members/", token=token) or []
|
||||||
|
return summarize(members, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_member_tracking(
|
||||||
|
character: str | int | None = None, limit: int = 50
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Member tracking (last login, location, ship) — needs the Director role."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
tracking = await client().get(f"/corporations/{corp_id}/membertracking/", token=token) or []
|
||||||
|
return summarize(tracking, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_divisions(character: str | int | None = None) -> dict[str, Any]:
|
||||||
|
"""Corp hangar/wallet division names (needs the Director role)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
data = await client().get(f"/corporations/{corp_id}/divisions/", token=token) or {}
|
||||||
|
return {"hangar": data.get("hangar"), "wallet": data.get("wallet")}
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_titles(character: str | int | None = None, limit: int = 50) -> dict[str, Any]:
|
||||||
|
"""Corp titles and their granted roles (needs a corp role)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
titles = await client().get(f"/corporations/{corp_id}/titles/", token=token) or []
|
||||||
|
return summarize(titles, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_medals(character: str | int | None = None, limit: int = 50) -> dict[str, Any]:
|
||||||
|
"""Corp medals (needs a corp role)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
medals = await client().get_all_pages(f"/corporations/{corp_id}/medals/", token=token)
|
||||||
|
return summarize(medals, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_standings(character: str | int | None = None, limit: int = 50) -> dict[str, Any]:
|
||||||
|
"""Corp standings toward other entities (needs a corp role)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
standings = await client().get_all_pages(f"/corporations/{corp_id}/standings/", token=token)
|
||||||
|
return summarize(standings, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_my_corp_roles(character: str | int | None = None) -> dict[str, Any]:
|
||||||
|
"""The character's own corp roles (character endpoint, no special role needed)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, cid: int) -> dict[str, Any]:
|
||||||
|
data = await client().get(f"/characters/{cid}/roles/", token=token) or {}
|
||||||
|
return {
|
||||||
|
"roles": data.get("roles"),
|
||||||
|
"roles_at_hq": data.get("roles_at_hq"),
|
||||||
|
"roles_at_base": data.get("roles_at_base"),
|
||||||
|
"roles_at_other": data.get("roles_at_other"),
|
||||||
|
}
|
||||||
|
|
||||||
|
return await run_per_character(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
TOOLS = [
|
||||||
|
get_corp_members,
|
||||||
|
get_corp_member_tracking,
|
||||||
|
get_corp_divisions,
|
||||||
|
get_corp_titles,
|
||||||
|
get_corp_medals,
|
||||||
|
get_corp_standings,
|
||||||
|
get_my_corp_roles,
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def register(mcp: FastMCP) -> None:
|
||||||
|
"""Register the corp organization tools onto the given server."""
|
||||||
|
for fn in TOOLS:
|
||||||
|
mcp.tool(fn)
|
||||||
@@ -14,6 +14,7 @@ from fastmcp import FastMCP
|
|||||||
from . import (
|
from . import (
|
||||||
__version__,
|
__version__,
|
||||||
account,
|
account,
|
||||||
|
corp,
|
||||||
industry,
|
industry,
|
||||||
killmails,
|
killmails,
|
||||||
market,
|
market,
|
||||||
@@ -71,3 +72,6 @@ killmails.register(mcp)
|
|||||||
|
|
||||||
# Phase 4: character profile / social (read-only).
|
# Phase 4: character profile / social (read-only).
|
||||||
profile.register(mcp)
|
profile.register(mcp)
|
||||||
|
|
||||||
|
# Phase 4: corp organization (read-only, needs corp roles).
|
||||||
|
corp.register(mcp)
|
||||||
@@ -0,0 +1,68 @@
|
|||||||
|
"""Corp organization tools (auth & ESI mocked)."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from conftest import route
|
||||||
|
from familiar import auth, corp
|
||||||
|
from familiar.server import mcp
|
||||||
|
|
||||||
|
CHARS = [{"character_id": 99, "character_name": "Alpha", "scopes": []}]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def fake_auth(monkeypatch):
|
||||||
|
monkeypatch.setattr(auth, "list_characters", lambda: list(CHARS))
|
||||||
|
|
||||||
|
async def get_auth(cid, **kw):
|
||||||
|
return f"token-{cid}", cid
|
||||||
|
|
||||||
|
monkeypatch.setattr(auth, "get_auth", get_auth)
|
||||||
|
monkeypatch.setattr(auth, "resolve_character", lambda i: 99)
|
||||||
|
return CHARS
|
||||||
|
|
||||||
|
|
||||||
|
def test_corp_members_resolves_corp(mock_esi, fake_auth):
|
||||||
|
fixtures = {"/members/": [1, 2, 3], "/characters/": {"corporation_id": 2000}}
|
||||||
|
mock_esi(lambda r: route(r, fixtures))
|
||||||
|
res = asyncio.run(corp.get_corp_members())
|
||||||
|
entry = res["characters"][0]
|
||||||
|
assert entry["corporation_id"] == 2000
|
||||||
|
assert entry["count"] == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_corp_divisions(mock_esi, fake_auth):
|
||||||
|
fixtures = {
|
||||||
|
"/divisions/": {"hangar": [{"division": 1, "name": "Ore"}], "wallet": []},
|
||||||
|
"/characters/": {"corporation_id": 2000},
|
||||||
|
}
|
||||||
|
mock_esi(lambda r: route(r, fixtures))
|
||||||
|
res = asyncio.run(corp.get_corp_divisions())
|
||||||
|
assert res["characters"][0]["hangar"][0]["name"] == "Ore"
|
||||||
|
|
||||||
|
|
||||||
|
def test_corp_medals_paginated(mock_esi, fake_auth):
|
||||||
|
fixtures = {"/medals/": [{"medal_id": 1}], "/characters/": {"corporation_id": 2000}}
|
||||||
|
mock_esi(lambda r: route(r, fixtures))
|
||||||
|
res = asyncio.run(corp.get_corp_medals())
|
||||||
|
assert res["characters"][0]["count"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_my_corp_roles(mock_esi, fake_auth):
|
||||||
|
mock_esi(lambda r: route(r, {"/roles/": {"roles": ["Director", "Accountant"]}}))
|
||||||
|
res = asyncio.run(corp.get_my_corp_roles())
|
||||||
|
assert "Director" in res["characters"][0]["roles"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_corp_tools_registered():
|
||||||
|
tools = {t.name for t in asyncio.run(mcp.list_tools())}
|
||||||
|
assert {
|
||||||
|
"get_corp_members",
|
||||||
|
"get_corp_member_tracking",
|
||||||
|
"get_corp_divisions",
|
||||||
|
"get_corp_titles",
|
||||||
|
"get_corp_medals",
|
||||||
|
"get_corp_standings",
|
||||||
|
"get_my_corp_roles",
|
||||||
|
} <= tools
|
||||||
Reference in new issue
Block a user