LatteandClaude Opus 4.8
2f12a03257
Add src/familiar/sde.py backing type lookups from a local SQLite SDE with graceful fallback to ESI when absent. lookup.resolve_id prefers the SDE for inventory_types; universe.get_item serves SDE-backed fields with no network call. Register a read-only sde_status() MCP tool. Tested against an in-memory SQLite fixture. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
"""Local SDE store: offline lookups and SDE-preferred / ESI-fallback wiring."""
|
|
|
|
import asyncio
|
|
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from familiar import sde, universe
|
|
|
|
|
|
def _make_sde(rows=((34, "Tritanium", 18, 0.01), (35, "Pyerite", 18, 0.01))):
|
|
"""In-memory SQLite seeded with the minimal invTypes schema and rows."""
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute(
|
|
"CREATE TABLE invTypes ("
|
|
"typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL)"
|
|
)
|
|
conn.executemany("INSERT INTO invTypes VALUES (?, ?, ?, ?)", rows)
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
@pytest.fixture
|
|
def use_sde():
|
|
"""Inject an in-memory SDE and reset the cached connection afterwards."""
|
|
conn = _make_sde()
|
|
sde.set_connection(conn)
|
|
yield conn
|
|
sde.set_connection(None)
|
|
conn.close()
|
|
|
|
|
|
def test_available_true_with_table(use_sde):
|
|
assert sde.available() is True
|
|
|
|
|
|
def test_available_false_without_connection():
|
|
sde.set_connection(None)
|
|
# No file on the default/overridden path in the test environment.
|
|
assert sde.available() is False
|
|
|
|
|
|
def test_available_false_when_table_missing():
|
|
conn = sqlite3.connect(":memory:") # connected, but no invTypes table
|
|
sde.set_connection(conn)
|
|
try:
|
|
assert sde.available() is False
|
|
# The graceful-degradation accessors must not crash either.
|
|
assert sde.type_by_id(34) is None
|
|
assert sde.type_by_name("Tritanium") is None
|
|
assert sde.resolve_type_name("Tritanium") is None
|
|
finally:
|
|
sde.set_connection(None)
|
|
conn.close()
|
|
|
|
|
|
def test_type_by_id(use_sde):
|
|
item = sde.type_by_id(34)
|
|
assert item == {"type_id": 34, "name": "Tritanium", "group_id": 18, "volume": 0.01}
|
|
|
|
|
|
def test_type_by_id_missing(use_sde):
|
|
assert sde.type_by_id(999999) is None
|
|
|
|
|
|
def test_type_by_name(use_sde):
|
|
item = sde.type_by_name("Pyerite")
|
|
assert item["type_id"] == 35
|
|
|
|
|
|
def test_resolve_type_name_case_insensitive(use_sde):
|
|
assert sde.resolve_type_name("tritanium") == 34
|
|
assert sde.resolve_type_name("TRITANIUM") == 34
|
|
assert sde.resolve_type_name("Tritanium") == 34
|
|
|
|
|
|
def test_resolve_type_name_unknown(use_sde):
|
|
assert sde.resolve_type_name("Nope") is None
|
|
|
|
|
|
def test_sde_status(use_sde):
|
|
status = sde.sde_status()
|
|
assert status["available"] is True
|
|
assert isinstance(status["path"], str)
|
|
|
|
|
|
def test_get_item_uses_sde_without_esi(use_sde, mock_esi):
|
|
def boom(request): # any ESI call here is a bug
|
|
raise AssertionError(f"ESI was called: {request.url}")
|
|
|
|
mock_esi(boom)
|
|
result = asyncio.run(universe.get_item("Tritanium"))
|
|
assert result["type_id"] == 34
|
|
assert result["name"] == "Tritanium"
|
|
assert result["volume"] == 0.01
|
|
# SDE has no description / packaged_volume.
|
|
assert result["description"] is None
|
|
assert result["packaged_volume"] is None
|
|
|
|
|
|
def test_get_item_falls_back_to_esi_when_sde_absent(mock_esi):
|
|
sde.set_connection(None) # no SDE
|
|
fixtures = {
|
|
"/universe/ids/": {"inventory_types": [{"id": 34, "name": "Tritanium"}]},
|
|
"/universe/types/": {
|
|
"type_id": 34,
|
|
"name": "Tritanium",
|
|
"group_id": 18,
|
|
"volume": 0.01,
|
|
"packaged_volume": 0.01,
|
|
"description": "The most common ore type.",
|
|
},
|
|
}
|
|
|
|
def route(request):
|
|
for needle, body in fixtures.items():
|
|
if needle in request.url.path:
|
|
return httpx.Response(200, json=body)
|
|
return httpx.Response(404, json={"error": "no fixture"})
|
|
|
|
mock_esi(route)
|
|
result = asyncio.run(universe.get_item("Tritanium"))
|
|
assert result["type_id"] == 34
|
|
assert result["description"] == "The most common ore type."
|