diff --git a/src/familiar/corp.py b/src/familiar/corp.py new file mode 100644 index 0000000..3e7566d --- /dev/null +++ b/src/familiar/corp.py @@ -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) diff --git a/src/familiar/server.py b/src/familiar/server.py index d520c0f..2a4943a 100644 --- a/src/familiar/server.py +++ b/src/familiar/server.py @@ -14,6 +14,7 @@ from fastmcp import FastMCP from . import ( __version__, account, + corp, industry, killmails, market, @@ -71,3 +72,6 @@ killmails.register(mcp) # Phase 4: character profile / social (read-only). profile.register(mcp) + +# Phase 4: corp organization (read-only, needs corp roles). +corp.register(mcp) diff --git a/tests/test_corp.py b/tests/test_corp.py new file mode 100644 index 0000000..669f460 --- /dev/null +++ b/tests/test_corp.py @@ -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