feat(sde): optional local SDE store for offline name/type lookups
CI / lint (pull_request) Successful in 51s
CI / test (pull_request) Successful in 59s
CI / publish (pull_request) Has been skipped
CI / lint (push) Successful in 51s
CI / test (push) Successful in 58s
CI / publish (push) Successful in 50s

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>
This commit is contained in:
LatteandClaude Opus 4.8 committed 2026-06-27 20:44:18 +02:00
1 parent 0dfd6d2350
commit 2f12a03257
5 files changed
+315

No files matched your search

+125
View File
@@ -0,0 +1,125 @@
"""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."