diff --git a/EveProfitManager/incursions.py b/EveProfitManager/incursions.py new file mode 100644 index 0000000..1994d3c --- /dev/null +++ b/EveProfitManager/incursions.py @@ -0,0 +1,57 @@ +"""Read where Sansha incursions are active, and their state, from ESI's public incursions endpoint.""" +from dataclasses import dataclass + +STATES = ("established", "mobilizing", "withdrawing") # in the order an incursion is worth running + + +@dataclass(frozen=True) +class Incursion: + constellation: str + staging: str # the system the fleet stages in + state: str + systems: int # infested solar systems + influence: float # 0 to 1; how much of the constellation is still Sansha-held + has_boss: bool + + +def get_incursions(client): + """ + The active incursions, established ones first. Needs no sign-in. + + A constellation or system whose name can't be read shows as its ID instead. + """ + body = client.get("/incursions", auth=False) + if not isinstance(body, list): + raise ValueError(f"Unexpected incursions response: {body!r}") + names = {} # path -> name, so a repeated constellation or system is read once + found = [] + for entry in body: + state = entry.get("state") if isinstance(entry, dict) else None + constellation, staging = _id(entry, "constellation_id"), _id(entry, "staging_solar_system_id") + systems = entry.get("infested_solar_systems") if isinstance(entry, dict) else None + if state not in STATES or constellation is None or staging is None or not isinstance(systems, list): + raise ValueError(f"Unexpected incursion: {entry!r}") + found.append(Incursion( + constellation=_name(client, names, f"/universe/constellations/{constellation}", + f"Constellation {constellation}"), + staging=_name(client, names, f"/universe/systems/{staging}", f"System {staging}"), + state=state, + systems=len(systems), + influence=float(entry.get("influence") or 0), + has_boss=bool(entry.get("has_boss")), + )) + return sorted(found, key=lambda i: STATES.index(i.state)) + + +def _id(entry, key): + value = entry.get(key) if isinstance(entry, dict) else None + return value if isinstance(value, int) and not isinstance(value, bool) else None + + +def _name(client, names, path, fallback): + if path not in names: + try: + names[path] = client.get(path, auth=False)["name"] + except Exception: # a missing name shouldn't hide the incursion + names[path] = fallback + return names[path] diff --git a/EveProfitManager/test_incursions.py b/EveProfitManager/test_incursions.py new file mode 100644 index 0000000..b43fe5a --- /dev/null +++ b/EveProfitManager/test_incursions.py @@ -0,0 +1,70 @@ +import unittest + +from incursions import Incursion, get_incursions + + +class FakeClient: + """Stands in for EsiClient: serves canned bodies by path and records the auth flag.""" + + def __init__(self, bodies): + self.bodies = bodies + self.auth_used = [] + + def get(self, path, auth=True): + self.auth_used.append(auth) + if path not in self.bodies: + raise RuntimeError("not found") + return self.bodies[path] + + +def raw(**over): + body = { + "constellation_id": 20000100, "faction_id": 500019, "has_boss": True, + "infested_solar_systems": [30000001, 30000002, 30000003], "influence": 0.5, + "staging_solar_system_id": 30000001, "state": "established", "type": "Incursion", + } + body.update(over) + return body + + +NAMES = { + "/universe/constellations/20000100": {"name": "Tanoo"}, + "/universe/systems/30000001": {"name": "Jita"}, +} + + +class GetIncursionsTests(unittest.TestCase): + def test_returns_named_incursions_without_signing_in(self): + client = FakeClient({"/incursions": [raw()], **NAMES}) + self.assertEqual( + get_incursions(client), + [Incursion(constellation="Tanoo", staging="Jita", state="established", systems=3, + influence=0.5, has_boss=True)], + ) + self.assertEqual(set(client.auth_used), {False}) + + def test_falls_back_to_ids_when_a_name_cannot_be_read(self): + client = FakeClient({"/incursions": [raw()]}) + incursion = get_incursions(client)[0] + self.assertEqual(incursion.constellation, "Constellation 20000100") + self.assertEqual(incursion.staging, "System 30000001") + + def test_orders_established_before_mobilizing_before_withdrawing(self): + client = FakeClient({"/incursions": [ + raw(state="withdrawing"), raw(state="established"), raw(state="mobilizing"), + ], **NAMES}) + self.assertEqual([i.state for i in get_incursions(client)], + ["established", "mobilizing", "withdrawing"]) + + def test_no_active_incursions_is_an_empty_list(self): + self.assertEqual(get_incursions(FakeClient({"/incursions": []})), []) + + def test_rejects_an_unexpected_response(self): + with self.assertRaises(ValueError): + get_incursions(FakeClient({"/incursions": {"error": "x"}})) + with self.assertRaises(ValueError): + get_incursions(FakeClient({"/incursions": [raw(state="weird")]})) + + +if __name__ == "__main__": + unittest.main()