diff --git a/src/familiar/__main__.py b/src/familiar/__main__.py index 78ab902..c4a1009 100644 --- a/src/familiar/__main__.py +++ b/src/familiar/__main__.py @@ -4,9 +4,10 @@ With no arguments it runs the MCP server over stdio (how Claude Desktop / Claude Code / Cowork launch it). The auth subcommands are meant to be run by hand in a terminal: - familiar login # browser-based EVE SSO login (run once) - familiar status # show whether you're logged in - familiar logout # forget stored credentials + familiar login # add a character via EVE SSO (run once per character) + familiar status # list logged-in characters + familiar logout # forget ALL characters + familiar logout NAME # forget one character (by name or id) """ from __future__ import annotations @@ -18,21 +19,32 @@ from . import auth def _login() -> int: bundle = auth.interactive_login() - print(f"Logged in as {bundle.get('character_name')} (id {bundle.get('character_id')}).") + print(f"Added {bundle.get('character_name')} (id {bundle.get('character_id')}).") + chars = auth.list_characters() + print(f"Logged-in characters ({len(chars)}): {', '.join(c['character_name'] for c in chars)}") return 0 def _status() -> int: - info = auth.status() - if info["authenticated"]: - print(f"Logged in as {info.get('character_name')} (id {info.get('character_id')}).") - else: + chars = auth.list_characters() + if not chars: print("Not logged in. Run `familiar login`.") + return 0 + print(f"Logged-in characters ({len(chars)}):") + for c in chars: + print(f" - {c['character_name']} (id {c['character_id']})") return 0 def _logout() -> int: - print("Logged out." if auth.clear_tokens() else "Was not logged in.") + target = sys.argv[2] if len(sys.argv) > 2 else None + removed = auth.remove_character(target) + if not removed: + print("Nothing to remove." if target is None else f"No character matching {target!r}.") + elif target is None: + print("Logged out all characters.") + else: + print(f"Removed {target}.") return 0 diff --git a/src/familiar/account.py b/src/familiar/account.py index aa911b6..d5e8d0a 100644 --- a/src/familiar/account.py +++ b/src/familiar/account.py @@ -1,12 +1,18 @@ -"""Authenticated, read-only tools for the logged-in character. +"""Authenticated, read-only tools for the logged-in character(s). -Each tool resolves a valid access token via :mod:`familiar.auth` (refreshing -transparently) and then calls the character ESI endpoints. If the user has not -logged in, the tools surface a clear "run `familiar login`" error. +Every tool takes an optional ``character`` (name or id): + +* omitted -> data for *all* logged-in characters +* given -> just that one character + +Results are always returned as a uniform ``{"characters": [ ... ]}`` list, one +entry per character, so the AI can handle one or many the same way. Tokens are +resolved (and refreshed) per character via :mod:`familiar.auth`. """ from __future__ import annotations +from collections.abc import Awaitable, Callable from typing import Any from fastmcp import FastMCP @@ -14,101 +20,129 @@ from fastmcp import FastMCP from . import auth from .esi import client +# A per-character fetcher: given (access_token, character_id) -> payload dict. +Fetcher = Callable[[str, int], Awaitable[dict[str, Any]]] -async def _auth() -> tuple[str, int]: - return await auth.get_auth() + +async def _run(character: str | int | None, fetch: Fetcher) -> dict[str, Any]: + chars = auth.list_characters() + if not chars: + raise auth.NotAuthenticated("Not logged in. Run `familiar login` in a terminal first.") + names = {c["character_id"]: c["character_name"] for c in chars} + + if character is None: + targets = [c["character_id"] for c in chars] + else: + targets = [auth.resolve_character(character)] + + results = [] + for cid in targets: + token, _ = await auth.get_auth(cid) + payload = await fetch(token, cid) + results.append({"character_id": cid, "character_name": names.get(cid), **payload}) + return {"characters": results} async def account_status() -> dict[str, Any]: - """Whether familiar is logged in, and as which character (no network call).""" + """List which characters familiar is logged in as (no network call).""" return auth.status() -async def get_wallet() -> dict[str, Any]: - """The character's wallet balance in ISK.""" - token, cid = await _auth() - balance = await client().get(f"/characters/{cid}/wallet/", token=token) - return {"character_id": cid, "balance": balance} +async def get_wallet(character: str | int | None = None) -> dict[str, Any]: + """Wallet balance (ISK) for one character or all of them.""" + + async def fetch(token: str, cid: int) -> dict[str, Any]: + return {"balance": await client().get(f"/characters/{cid}/wallet/", token=token)} + + return await _run(character, fetch) -async def get_skills() -> dict[str, Any]: - """Total skill points, unallocated SP, and how many skills are trained.""" - token, cid = await _auth() - data = await client().get(f"/characters/{cid}/skills/", token=token) - return { - "character_id": cid, - "total_sp": data.get("total_sp"), - "unallocated_sp": data.get("unallocated_sp"), - "skills_trained": len(data.get("skills") or []), - } +async def get_skills(character: str | int | None = None) -> dict[str, Any]: + """Total SP, unallocated SP, and skill count for one character or all.""" - -async def get_skill_queue(limit: int = 10) -> dict[str, Any]: - """The next entries in the skill training queue.""" - token, cid = await _auth() - queue = await client().get(f"/characters/{cid}/skillqueue/", token=token) or [] - upcoming = [ - { - "skill_id": q.get("skill_id"), - "finished_level": q.get("finished_level"), - "position": q.get("queue_position"), - "finish_date": q.get("finish_date"), + async def fetch(token: str, cid: int) -> dict[str, Any]: + data = await client().get(f"/characters/{cid}/skills/", token=token) + return { + "total_sp": data.get("total_sp"), + "unallocated_sp": data.get("unallocated_sp"), + "skills_trained": len(data.get("skills") or []), } - for q in queue[: max(1, limit)] - ] - return {"character_id": cid, "in_queue": len(queue), "upcoming": upcoming} + + return await _run(character, fetch) -async def get_location() -> dict[str, Any]: - """Current system/station, active ship, and online status.""" - token, cid = await _auth() - loc = await client().get(f"/characters/{cid}/location/", token=token) or {} - ship = await client().get(f"/characters/{cid}/ship/", token=token) or {} - online = await client().get(f"/characters/{cid}/online/", token=token) or {} - return { - "character_id": cid, - "solar_system_id": loc.get("solar_system_id"), - "station_id": loc.get("station_id"), - "structure_id": loc.get("structure_id"), - "ship_type_id": ship.get("ship_type_id"), - "ship_name": ship.get("ship_name"), - "online": online.get("online"), - "last_login": online.get("last_login"), - } +async def get_skill_queue(character: str | int | None = None, limit: int = 10) -> dict[str, Any]: + """Upcoming skill-queue entries for one character or all.""" + + async def fetch(token: str, cid: int) -> dict[str, Any]: + queue = await client().get(f"/characters/{cid}/skillqueue/", token=token) or [] + upcoming = [ + { + "skill_id": q.get("skill_id"), + "finished_level": q.get("finished_level"), + "position": q.get("queue_position"), + "finish_date": q.get("finish_date"), + } + for q in queue[: max(1, limit)] + ] + return {"in_queue": len(queue), "upcoming": upcoming} + + return await _run(character, fetch) -async def get_clones() -> dict[str, Any]: - """Home station, jump clones, and currently plugged-in implants.""" - token, cid = await _auth() - clones = await client().get(f"/characters/{cid}/clones/", token=token) or {} - implants = await client().get(f"/characters/{cid}/implants/", token=token) or [] - return { - "character_id": cid, - "home_location": clones.get("home_location"), - "jump_clones": len(clones.get("jump_clones") or []), - "active_implants": implants, - } +async def get_location(character: str | int | None = None) -> dict[str, Any]: + """Current system/station, active ship, and online status for one or all.""" - -async def get_assets(limit: int = 20) -> dict[str, Any]: - """Asset summary: total item stacks and a sample of them. - - The full asset list can be thousands of stacks, so only the first ``limit`` - are returned alongside the total count. - """ - token, cid = await _auth() - assets = await client().get_all_pages(f"/characters/{cid}/assets/", token=token) - sample = [ - { - "type_id": a.get("type_id"), - "quantity": a.get("quantity"), - "location_id": a.get("location_id"), - "location_flag": a.get("location_flag"), - "is_singleton": a.get("is_singleton"), + async def fetch(token: str, cid: int) -> dict[str, Any]: + loc = await client().get(f"/characters/{cid}/location/", token=token) or {} + ship = await client().get(f"/characters/{cid}/ship/", token=token) or {} + online = await client().get(f"/characters/{cid}/online/", token=token) or {} + return { + "solar_system_id": loc.get("solar_system_id"), + "station_id": loc.get("station_id"), + "structure_id": loc.get("structure_id"), + "ship_type_id": ship.get("ship_type_id"), + "ship_name": ship.get("ship_name"), + "online": online.get("online"), + "last_login": online.get("last_login"), } - for a in assets[: max(1, limit)] - ] - return {"character_id": cid, "total_stacks": len(assets), "sample": sample} + + return await _run(character, fetch) + + +async def get_clones(character: str | int | None = None) -> dict[str, Any]: + """Home station, jump clones, and active implants for one character or all.""" + + async def fetch(token: str, cid: int) -> dict[str, Any]: + clones = await client().get(f"/characters/{cid}/clones/", token=token) or {} + implants = await client().get(f"/characters/{cid}/implants/", token=token) or [] + return { + "home_location": clones.get("home_location"), + "jump_clones": len(clones.get("jump_clones") or []), + "active_implants": implants, + } + + return await _run(character, fetch) + + +async def get_assets(character: str | int | None = None, limit: int = 20) -> dict[str, Any]: + """Asset summary (total stacks + a sample) for one character or all.""" + + async def fetch(token: str, cid: int) -> dict[str, Any]: + assets = await client().get_all_pages(f"/characters/{cid}/assets/", token=token) + sample = [ + { + "type_id": a.get("type_id"), + "quantity": a.get("quantity"), + "location_id": a.get("location_id"), + "location_flag": a.get("location_flag"), + "is_singleton": a.get("is_singleton"), + } + for a in assets[: max(1, limit)] + ] + return {"total_stacks": len(assets), "sample": sample} + + return await _run(character, fetch) TOOLS = [ diff --git a/src/familiar/auth.py b/src/familiar/auth.py index 8c7b57f..4d315e8 100644 --- a/src/familiar/auth.py +++ b/src/familiar/auth.py @@ -146,8 +146,10 @@ def _load_key() -> bytes: return key -def save_tokens(store: dict[str, Any]) -> None: - """Encrypt and persist the token bundle.""" +# --------------------------------------------------------------------------- # +# Multi-character store: { "characters": { "": , ... } } +# --------------------------------------------------------------------------- # +def _write_store(store: dict[str, Any]) -> None: blob = Fernet(_load_key()).encrypt(json.dumps(store).encode()) tf = config.token_file() tf.write_bytes(blob) @@ -157,21 +159,70 @@ def save_tokens(store: dict[str, Any]) -> None: pass -def load_tokens() -> dict[str, Any] | None: - """Load and decrypt the token bundle, or None if not logged in.""" +def load_store() -> dict[str, Any]: + """Load the decrypted store, migrating a legacy single-character file. + + Always returns a dict with a ``characters`` mapping (possibly empty). + """ tf = config.token_file() if not tf.exists(): - return None - return json.loads(Fernet(_load_key()).decrypt(tf.read_bytes())) + return {"characters": {}} + data = json.loads(Fernet(_load_key()).decrypt(tf.read_bytes())) + if "characters" in data: + return data + # Legacy format: a single bundle at the top level. Migrate it. + if "refresh_token" in data and "character_id" in data: + return {"characters": {str(data["character_id"]): data}} + return {"characters": {}} -def clear_tokens() -> bool: - """Remove stored credentials. Returns True if something was removed.""" - tf = config.token_file() - if tf.exists(): - tf.unlink() +def _save_character(bundle: dict[str, Any]) -> None: + store = load_store() + store["characters"][str(bundle["character_id"])] = bundle + _write_store(store) + + +def list_characters() -> list[dict[str, Any]]: + """Logged-in characters (id, name, scopes), sorted by name.""" + chars = load_store()["characters"].values() + out = [ + { + "character_id": b.get("character_id"), + "character_name": b.get("character_name"), + "scopes": b.get("scopes"), + } + for b in chars + ] + return sorted(out, key=lambda c: (c.get("character_name") or "").lower()) + + +def remove_character(identifier: str | int | None = None) -> bool: + """Remove one character (by id or name) or, if None, all of them.""" + store = load_store() + chars = store["characters"] + if not chars: + return False + if identifier is None: + store["characters"] = {} + _write_store(store) return True - return False + target = _match_character(identifier, chars) + if target is None: + return False + del chars[target] + _write_store(store) + return True + + +def _match_character(identifier: str | int, chars: dict[str, Any]) -> str | None: + """Return the store key for a character matched by id or (case-insensitive) name.""" + ident = str(identifier).strip() + if ident in chars: + return ident + for key, bundle in chars.items(): + if (bundle.get("character_name") or "").lower() == ident.lower(): + return key + return None def _bundle_from_response(resp: dict[str, Any], *, now: float) -> dict[str, Any]: @@ -186,27 +237,40 @@ def _bundle_from_response(resp: dict[str, Any], *, now: float) -> dict[str, Any] } +def resolve_character(identifier: str | int) -> int: + """Resolve a name-or-id to a logged-in character id, or raise.""" + chars = load_store()["characters"] + key = _match_character(identifier, chars) + if key is None: + known = ", ".join(c["character_name"] or str(c["character_id"]) for c in list_characters()) + raise NotAuthenticated(f"No logged-in character matching {identifier!r}. Known: {known}.") + return int(key) + + async def get_auth( - *, now: float | None = None, transport: httpx.AsyncBaseTransport | None = None + character_id: int, + *, + now: float | None = None, + transport: httpx.AsyncBaseTransport | None = None, ) -> tuple[str, int]: - """Return ``(access_token, character_id)``, refreshing if near expiry. + """Return ``(access_token, character_id)`` for one character, refreshing if needed. - Raises :class:`NotAuthenticated` if the user has not logged in. + Raises :class:`NotAuthenticated` if that character is not logged in. """ - store = load_tokens() - if not store: - raise NotAuthenticated("Not logged in. Run `familiar login` in a terminal first.") + store = load_store() + bundle = store["characters"].get(str(character_id)) + if not bundle: + raise NotAuthenticated(f"Character {character_id} is not logged in. Run `familiar login`.") now = now if now is not None else time.time() - if store.get("access_token") and store.get("expires_at", 0) - now > 60: - return store["access_token"], store["character_id"] + if bundle.get("access_token") and bundle.get("expires_at", 0) - now > 60: + return bundle["access_token"], character_id - resp = await refresh_tokens(store["refresh_token"], transport=transport) - bundle = _bundle_from_response(resp, now=now) - # Preserve the known character even if the new token omits a field. - bundle["character_id"] = store.get("character_id", bundle["character_id"]) - bundle["character_name"] = bundle["character_name"] or store.get("character_name") - save_tokens(bundle) - return bundle["access_token"], bundle["character_id"] + resp = await refresh_tokens(bundle["refresh_token"], transport=transport) + fresh = _bundle_from_response(resp, now=now) + fresh["character_id"] = character_id + fresh["character_name"] = fresh["character_name"] or bundle.get("character_name") + _save_character(fresh) + return fresh["access_token"], character_id _SUCCESS_PAGE = ( @@ -260,18 +324,11 @@ def interactive_login(*, open_browser: bool = True) -> dict[str, Any]: raise AuthError("State mismatch — possible CSRF; aborting login.") resp = asyncio.run(exchange_code(captured["code"], verifier)) bundle = _bundle_from_response(resp, now=time.time()) - save_tokens(bundle) + _save_character(bundle) # adds or updates this character, keeping the others return bundle def status() -> dict[str, Any]: - """Report whether we're logged in and as whom (no network).""" - store = load_tokens() - if not store: - return {"authenticated": False} - return { - "authenticated": True, - "character_id": store.get("character_id"), - "character_name": store.get("character_name"), - "scopes": store.get("scopes"), - } + """Report which characters are logged in (no network call).""" + chars = list_characters() + return {"authenticated": bool(chars), "characters": chars} diff --git a/tests/test_account.py b/tests/test_account.py index 2ac2c85..576ba14 100644 --- a/tests/test_account.py +++ b/tests/test_account.py @@ -1,100 +1,134 @@ -"""Authenticated account tools (auth mocked, ESI mocked).""" +"""Authenticated account tools (auth mocked, ESI mocked), multi-character.""" import asyncio import httpx import pytest +from conftest import route from familiar import account, auth from familiar.server import mcp -from conftest import route + +CHARS = [ + {"character_id": 99, "character_name": "Alpha", "scopes": []}, + {"character_id": 100, "character_name": "Beta", "scopes": []}, +] @pytest.fixture def fake_auth(monkeypatch): - async def _stub(): - return "fake-token", 99 + monkeypatch.setattr(auth, "list_characters", lambda: list(CHARS)) - monkeypatch.setattr(auth, "get_auth", _stub) - return 99 + async def get_auth(cid, **kw): + return f"token-{cid}", cid + + monkeypatch.setattr(auth, "get_auth", get_auth) + + def resolve(identifier): + for c in CHARS: + if ( + str(c["character_id"]) == str(identifier) + or (c["character_name"] or "").lower() == str(identifier).lower() + ): + return c["character_id"] + raise auth.NotAuthenticated("no such character") + + monkeypatch.setattr(auth, "resolve_character", resolve) + return CHARS -def test_get_wallet(mock_esi, fake_auth): - mock_esi(lambda req: httpx.Response(200, json=1234567.89)) +def test_wallet_all_characters(mock_esi, fake_auth): + mock_esi(lambda req: httpx.Response(200, json=1000.0)) result = asyncio.run(account.get_wallet()) - assert result["character_id"] == 99 - assert result["balance"] == 1234567.89 + assert len(result["characters"]) == 2 + assert {c["character_name"] for c in result["characters"]} == {"Alpha", "Beta"} + assert all(c["balance"] == 1000.0 for c in result["characters"]) -def test_get_skills_summarizes(mock_esi, fake_auth): +def test_wallet_single_character_by_name(mock_esi, fake_auth): + mock_esi(lambda req: httpx.Response(200, json=500.0)) + result = asyncio.run(account.get_wallet(character="Alpha")) + assert len(result["characters"]) == 1 + assert result["characters"][0]["character_id"] == 99 + + +def test_unknown_character_raises(mock_esi, fake_auth): + mock_esi(lambda req: httpx.Response(200, json=1.0)) + with pytest.raises(auth.NotAuthenticated): + asyncio.run(account.get_wallet(character="Ghost")) + + +def test_not_logged_in_raises(monkeypatch): + monkeypatch.setattr(auth, "list_characters", lambda: []) + with pytest.raises(auth.NotAuthenticated): + asyncio.run(account.get_wallet()) + + +def test_skills_summary(mock_esi, fake_auth): body = {"total_sp": 5_000_000, "unallocated_sp": 10_000, "skills": [{}, {}, {}]} mock_esi(lambda req: route(req, {"/skills/": body})) - result = asyncio.run(account.get_skills()) - assert result["total_sp"] == 5_000_000 - assert result["skills_trained"] == 3 + result = asyncio.run(account.get_skills(character=99)) + assert result["characters"][0]["total_sp"] == 5_000_000 + assert result["characters"][0]["skills_trained"] == 3 -def test_get_skill_queue_limits(mock_esi, fake_auth): - queue = [ - {"skill_id": i, "finished_level": 5, "queue_position": i, "finish_date": "2026"} - for i in range(5) - ] +def test_skill_queue_limit(mock_esi, fake_auth): + queue = [{"skill_id": i, "finished_level": 5, "queue_position": i} for i in range(5)] mock_esi(lambda req: route(req, {"/skillqueue/": queue})) - result = asyncio.run(account.get_skill_queue(limit=2)) - assert result["in_queue"] == 5 - assert len(result["upcoming"]) == 2 + result = asyncio.run(account.get_skill_queue(character="Beta", limit=2)) + entry = result["characters"][0] + assert entry["in_queue"] == 5 + assert len(entry["upcoming"]) == 2 -def test_get_location_merges_endpoints(mock_esi, fake_auth): +def test_location_merges_endpoints(mock_esi, fake_auth): fixtures = { "/location/": {"solar_system_id": 30000142, "station_id": 60003760}, "/ship/": {"ship_type_id": 670, "ship_name": "My Capsule"}, "/online/": {"online": True, "last_login": "2026-06-27T00:00:00Z"}, } mock_esi(lambda req: route(req, fixtures)) - result = asyncio.run(account.get_location()) - assert result["solar_system_id"] == 30000142 - assert result["ship_name"] == "My Capsule" - assert result["online"] is True + result = asyncio.run(account.get_location(character=99)) + entry = result["characters"][0] + assert entry["solar_system_id"] == 30000142 + assert entry["ship_name"] == "My Capsule" + assert entry["online"] is True -def test_get_clones(mock_esi, fake_auth): +def test_clones(mock_esi, fake_auth): fixtures = { "/clones/": {"home_location": {"location_id": 60003760}, "jump_clones": [{}, {}]}, "/implants/": [10000, 10001], } mock_esi(lambda req: route(req, fixtures)) - result = asyncio.run(account.get_clones()) - assert result["jump_clones"] == 2 - assert result["active_implants"] == [10000, 10001] + result = asyncio.run(account.get_clones(character=99)) + entry = result["characters"][0] + assert entry["jump_clones"] == 2 + assert entry["active_implants"] == [10000, 10001] -def test_get_assets_paginates(mock_esi, fake_auth): - # Two pages of assets; the client should follow X-Pages and concatenate. +def test_assets_paginates(mock_esi, fake_auth): def handler(req): page = req.url.params.get("page", "1") - if page == "1": - return httpx.Response( - 200, - json=[{"type_id": 34, "quantity": 1, "location_id": 1}], - headers={"X-Pages": "2"}, - ) + type_id = 34 if page == "1" else 35 return httpx.Response( 200, - json=[{"type_id": 35, "quantity": 2, "location_id": 1}], + json=[{"type_id": type_id, "quantity": 1, "location_id": 1}], headers={"X-Pages": "2"}, ) mock_esi(handler) - result = asyncio.run(account.get_assets(limit=20)) - assert result["total_stacks"] == 2 - assert {a["type_id"] for a in result["sample"]} == {34, 35} + result = asyncio.run(account.get_assets(character=99, limit=20)) + entry = result["characters"][0] + assert entry["total_stacks"] == 2 + assert {a["type_id"] for a in entry["sample"]} == {34, 35} def test_account_status_delegates(monkeypatch): - monkeypatch.setattr(auth, "status", lambda: {"authenticated": True, "character_id": 1}) + monkeypatch.setattr(auth, "status", lambda: {"authenticated": True, "characters": CHARS}) result = asyncio.run(account.account_status()) assert result["authenticated"] is True + assert len(result["characters"]) == 2 def test_account_tools_registered(): diff --git a/tests/test_auth.py b/tests/test_auth.py index 32c55c8..6e4bbef 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1,4 +1,4 @@ -"""EVE SSO auth: PKCE, JWT claims, token store, exchange/refresh, get_auth.""" +"""EVE SSO auth: PKCE, JWT claims, multi-character store, exchange/refresh.""" import asyncio import base64 @@ -18,6 +18,19 @@ def tmp_store(tmp_path, monkeypatch): return tmp_path +def _char(cid, name, **extra): + return { + "refresh_token": f"r-{cid}", + "access_token": f"a-{cid}", + "expires_at": 10_000, + "character_id": cid, + "character_name": name, + "scopes": [], + **extra, + } + + +# --- pure helpers ---------------------------------------------------------- # def test_generate_pkce_challenge_matches_verifier(): verifier, challenge = auth.generate_pkce() expected = ( @@ -33,11 +46,10 @@ def test_build_authorize_url_has_required_params(): assert "code_challenge=the-challenge" in url assert "code_challenge_method=S256" in url assert "state=the-state" in url - assert "client_id=" in url def test_decode_jwt_and_character_id(): - token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte", "scp": ["a"]}) + token = make_jwt({"sub": "CHARACTER:EVE:2112000001", "name": "Latte"}) claims = auth.decode_jwt_claims(token) assert claims["name"] == "Latte" assert auth.character_id_from_claims(claims) == 2112000001 @@ -53,28 +65,58 @@ def test_decode_malformed_token_raises(): auth.decode_jwt_claims("not-a-jwt") -def test_token_store_round_trip(tmp_store): - bundle = {"refresh_token": "r", "character_id": 7, "character_name": "X"} - assert auth.load_tokens() is None - auth.save_tokens(bundle) - assert auth.load_tokens() == bundle - assert (tmp_store / "token.enc").exists() - assert auth.clear_tokens() is True - assert auth.load_tokens() is None - assert auth.clear_tokens() is False +# --- multi-character store ------------------------------------------------- # +def test_store_starts_empty(tmp_store): + assert auth.load_store() == {"characters": {}} + assert auth.list_characters() == [] -def test_stored_token_is_encrypted_not_plaintext(tmp_store): - auth.save_tokens({"refresh_token": "super-secret-value", "character_id": 1}) +def test_save_and_list_multiple_characters(tmp_store): + auth._save_character(_char(1, "Beta")) + auth._save_character(_char(2, "Alpha")) + chars = auth.list_characters() + assert [c["character_name"] for c in chars] == ["Alpha", "Beta"] # sorted by name + + +def test_legacy_single_bundle_is_migrated(tmp_store): + # Write a pre-multi-character file (single bundle at top level). + auth._write_store({"refresh_token": "r", "character_id": 42, "character_name": "Old"}) + store = auth.load_store() + assert "42" in store["characters"] + assert auth.list_characters()[0]["character_name"] == "Old" + + +def test_remove_character_by_name_and_all(tmp_store): + auth._save_character(_char(1, "Beta")) + auth._save_character(_char(2, "Alpha")) + assert auth.remove_character("Alpha") is True + assert {c["character_id"] for c in auth.list_characters()} == {1} + assert auth.remove_character("nobody") is False + assert auth.remove_character() is True # removes all + assert auth.list_characters() == [] + assert auth.remove_character() is False + + +def test_resolve_character_by_id_and_name(tmp_store): + auth._save_character(_char(7, "Gamma")) + assert auth.resolve_character(7) == 7 + assert auth.resolve_character("gamma") == 7 + with pytest.raises(auth.NotAuthenticated): + auth.resolve_character("ghost") + + +def test_stored_tokens_are_encrypted(tmp_store): + auth._save_character(_char(1, "X", refresh_token="super-secret-value")) raw = (tmp_store / "token.enc").read_bytes() assert b"super-secret-value" not in raw +# --- token requests + get_auth -------------------------------------------- # def _token_transport(payload): return httpx.MockTransport(lambda req: httpx.Response(200, json=payload)) -def test_exchange_code_posts_and_parses(): +def test_exchange_code_parses(): payload = {"access_token": "a", "refresh_token": "r", "expires_in": 1200} out = asyncio.run(auth.exchange_code("code", "verifier", transport=_token_transport(payload))) assert out["refresh_token"] == "r" @@ -86,41 +128,31 @@ def test_token_request_raises_on_error(): asyncio.run(auth.refresh_tokens("r", transport=transport)) -def test_get_auth_without_login_raises(tmp_store): +def test_get_auth_unknown_character_raises(tmp_store): with pytest.raises(auth.NotAuthenticated): - asyncio.run(auth.get_auth()) + asyncio.run(auth.get_auth(123)) def test_get_auth_uses_unexpired_token(tmp_store): - auth.save_tokens( - { - "access_token": "still-good", - "refresh_token": "r", - "character_id": 9, - "expires_at": 10_000, - } - ) - token, cid = asyncio.run(auth.get_auth(now=5_000)) + auth._save_character(_char(9, "Z", access_token="still-good", expires_at=10_000)) + token, cid = asyncio.run(auth.get_auth(9, now=5_000)) assert token == "still-good" assert cid == 9 def test_get_auth_refreshes_when_expired(tmp_store): - auth.save_tokens( - {"access_token": "old", "refresh_token": "r0", "character_id": 9, "expires_at": 1_000} - ) - new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Latte"}) + auth._save_character(_char(9, "Z", refresh_token="r0", expires_at=1_000)) + new_access = make_jwt({"sub": "CHARACTER:EVE:9", "name": "Z"}) payload = {"access_token": new_access, "refresh_token": "r1", "expires_in": 1200} - token, cid = asyncio.run(auth.get_auth(now=2_000, transport=_token_transport(payload))) + token, cid = asyncio.run(auth.get_auth(9, now=2_000, transport=_token_transport(payload))) assert token == new_access assert cid == 9 - # The rotated refresh token is persisted. - assert auth.load_tokens()["refresh_token"] == "r1" + assert auth.load_store()["characters"]["9"]["refresh_token"] == "r1" -def test_status_reflects_store(tmp_store): - assert auth.status() == {"authenticated": False} - auth.save_tokens({"refresh_token": "r", "character_id": 3, "character_name": "Y", "scopes": []}) +def test_status_lists_characters(tmp_store): + assert auth.status() == {"authenticated": False, "characters": []} + auth._save_character(_char(3, "Y")) info = auth.status() assert info["authenticated"] is True - assert info["character_id"] == 3 + assert info["characters"][0]["character_id"] == 3