LatteandClaude Opus 4.8
463ea84360
Implements issue #51. Adds a Fuzzwork SDE downloader/importer and expands the offline lookup layer beyond item types. - sde.py: SDE_URL constant (FAMILIAR_SDE_URL override) and async download_sde() that streams + bz2-decompresses the dump incrementally to sde_path(), resetting the cached connection. - sde.py: system_by_id/system_by_name/station_by_id offline lookups, all degrading to None when the table/file is missing. - universe.py: get_system/get_station prefer the SDE when available, falling back to ESI otherwise (signatures unchanged). - __main__.py: `familiar sde download|status|path` subcommands. - tests: download_sde via injected MockTransport, new lookups + None fallbacks, and SDE-vs-ESI fast-path/fallback for get_system/station. - README: short offline-SDE setup note. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Local SDE store: offline lookups and SDE-preferred / ESI-fallback wiring."""
|
|
|
|
import asyncio
|
|
import bz2
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
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
|
|
|
|
|
|
def _build_sde_file(path: Path) -> None:
|
|
"""Write a tiny on-disk SDE with invTypes, mapSolarSystems and staStations."""
|
|
conn = sqlite3.connect(str(path))
|
|
conn.executescript(
|
|
"CREATE TABLE invTypes ("
|
|
"typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL);"
|
|
"CREATE TABLE mapSolarSystems ("
|
|
"solarSystemID INTEGER PRIMARY KEY, solarSystemName TEXT, regionID INTEGER, "
|
|
"constellationID INTEGER, security REAL);"
|
|
"CREATE TABLE staStations ("
|
|
"stationID INTEGER PRIMARY KEY, stationName TEXT, solarSystemID INTEGER, "
|
|
"corporationID INTEGER);"
|
|
)
|
|
conn.execute("INSERT INTO invTypes VALUES (?, ?, ?, ?)", (34, "Tritanium", 18, 0.01))
|
|
conn.execute(
|
|
"INSERT INTO mapSolarSystems VALUES (?, ?, ?, ?, ?)",
|
|
(30000142, "Jita", 10000002, 20000020, 0.945913),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO staStations VALUES (?, ?, ?, ?)",
|
|
(60003760, "Jita IV - Moon 4 - Caldari Navy Assembly Plant", 30000142, 1000035),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def _make_sde_full():
|
|
"""In-memory SQLite with invTypes, mapSolarSystems and staStations populated."""
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.executescript(
|
|
"CREATE TABLE invTypes ("
|
|
"typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL);"
|
|
"CREATE TABLE mapSolarSystems ("
|
|
"solarSystemID INTEGER PRIMARY KEY, solarSystemName TEXT, regionID INTEGER, "
|
|
"constellationID INTEGER, security REAL);"
|
|
"CREATE TABLE staStations ("
|
|
"stationID INTEGER PRIMARY KEY, stationName TEXT, solarSystemID INTEGER, "
|
|
"corporationID INTEGER);"
|
|
)
|
|
conn.execute("INSERT INTO invTypes VALUES (?, ?, ?, ?)", (34, "Tritanium", 18, 0.01))
|
|
conn.execute(
|
|
"INSERT INTO mapSolarSystems VALUES (?, ?, ?, ?, ?)",
|
|
(30000142, "Jita", 10000002, 20000020, 0.945913),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO staStations VALUES (?, ?, ?, ?)",
|
|
(60003760, "Jita IV - Moon 4 - CNAP", 30000142, 1000035),
|
|
)
|
|
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()
|
|
|
|
|
|
@pytest.fixture
|
|
def use_sde_full():
|
|
"""Inject an in-memory SDE with systems + stations; reset afterwards."""
|
|
conn = _make_sde_full()
|
|
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."
|
|
|
|
|
|
# --- system / station offline lookups -------------------------------------
|
|
|
|
|
|
def test_system_by_id(use_sde_full):
|
|
sysrow = sde.system_by_id(30000142)
|
|
assert sysrow == {
|
|
"system_id": 30000142,
|
|
"name": "Jita",
|
|
"region_id": 10000002,
|
|
"constellation_id": 20000020,
|
|
"security_status": 0.945913,
|
|
}
|
|
|
|
|
|
def test_system_by_name_case_insensitive(use_sde_full):
|
|
assert sde.system_by_name("jita")["system_id"] == 30000142
|
|
assert sde.system_by_name("JITA")["system_id"] == 30000142
|
|
|
|
|
|
def test_system_lookups_missing(use_sde_full):
|
|
assert sde.system_by_id(999999) is None
|
|
assert sde.system_by_name("Nowhere") is None
|
|
|
|
|
|
def test_station_by_id(use_sde_full):
|
|
station = sde.station_by_id(60003760)
|
|
assert station == {
|
|
"station_id": 60003760,
|
|
"name": "Jita IV - Moon 4 - CNAP",
|
|
"system_id": 30000142,
|
|
}
|
|
|
|
|
|
def test_station_by_id_missing(use_sde_full):
|
|
assert sde.station_by_id(999999) is None
|
|
|
|
|
|
def test_system_station_lookups_degrade_without_table(use_sde):
|
|
# `use_sde` has only invTypes — the system/station tables are absent.
|
|
assert sde.system_by_id(30000142) is None
|
|
assert sde.system_by_name("Jita") is None
|
|
assert sde.station_by_id(60003760) is None
|
|
|
|
|
|
def test_system_station_lookups_degrade_without_connection():
|
|
sde.set_connection(None)
|
|
assert sde.system_by_id(30000142) is None
|
|
assert sde.system_by_name("Jita") is None
|
|
assert sde.station_by_id(60003760) is None
|
|
|
|
|
|
# --- universe SDE fast-path / ESI fallback --------------------------------
|
|
|
|
|
|
def test_get_system_uses_sde_without_esi(use_sde_full, 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_system("Jita"))
|
|
assert result["system_id"] == 30000142
|
|
assert result["name"] == "Jita"
|
|
assert result["constellation_id"] == 20000020
|
|
assert result["security_status"] == 0.95 # rounded to 2 dp
|
|
assert result["security_class"] is None
|
|
assert result["planets"] == 0
|
|
assert result["stargates"] == 0
|
|
|
|
|
|
def test_get_system_falls_back_to_esi_when_sde_absent(mock_esi):
|
|
sde.set_connection(None)
|
|
fixtures = {
|
|
"/universe/ids/": {"systems": [{"id": 30000142, "name": "Jita"}]},
|
|
"/universe/systems/": {
|
|
"system_id": 30000142,
|
|
"name": "Jita",
|
|
"constellation_id": 20000020,
|
|
"security_status": 0.9459,
|
|
"security_class": "B",
|
|
"planets": [{"planet_id": 1}],
|
|
"stargates": [1, 2, 3],
|
|
},
|
|
}
|
|
|
|
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_system("Jita"))
|
|
assert result["system_id"] == 30000142
|
|
assert result["security_status"] == 0.95
|
|
assert result["security_class"] == "B"
|
|
assert result["planets"] == 1
|
|
assert result["stargates"] == 3
|
|
|
|
|
|
def test_get_station_uses_sde_for_numeric_id_without_esi(use_sde_full, mock_esi):
|
|
def boom(request):
|
|
raise AssertionError(f"ESI was called: {request.url}")
|
|
|
|
mock_esi(boom)
|
|
result = asyncio.run(universe.get_station(60003760))
|
|
assert result["station_id"] == 60003760
|
|
assert result["system_id"] == 30000142
|
|
assert result["type_id"] is None
|
|
assert result["services"] is None
|
|
|
|
|
|
def test_get_station_falls_back_to_esi_when_sde_absent(mock_esi):
|
|
sde.set_connection(None)
|
|
fixtures = {
|
|
"/universe/stations/": {
|
|
"station_id": 60003760,
|
|
"name": "Jita IV - Moon 4 - CNAP",
|
|
"system_id": 30000142,
|
|
"type_id": 1529,
|
|
"services": ["market", "repair"],
|
|
},
|
|
}
|
|
|
|
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_station(60003760))
|
|
assert result["type_id"] == 1529
|
|
assert result["services"] == ["market", "repair"]
|
|
|
|
|
|
# --- download_sde ----------------------------------------------------------
|
|
|
|
|
|
def test_download_sde_decompresses_and_installs(tmp_path):
|
|
src = tmp_path / "source.sqlite"
|
|
_build_sde_file(src)
|
|
compressed = bz2.compress(src.read_bytes())
|
|
|
|
def handler(request):
|
|
# Serve the bz2 bytes in a couple of chunks to exercise streaming.
|
|
return httpx.Response(200, content=compressed)
|
|
|
|
dest = tmp_path / "installed.sqlite"
|
|
try:
|
|
result = asyncio.run(
|
|
sde.download_sde(
|
|
url="https://example.invalid/sde.sqlite.bz2",
|
|
dest=dest,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
)
|
|
assert result["path"] == str(dest)
|
|
assert dest.is_file()
|
|
assert result["bytes"] == dest.stat().st_size
|
|
|
|
# Point the live store at the freshly written file and query it.
|
|
import os
|
|
|
|
os.environ["FAMILIAR_SDE_PATH"] = str(dest)
|
|
sde.set_connection(None)
|
|
try:
|
|
assert sde.available() is True
|
|
assert sde.type_by_id(34)["name"] == "Tritanium"
|
|
assert sde.system_by_name("Jita")["system_id"] == 30000142
|
|
finally:
|
|
del os.environ["FAMILIAR_SDE_PATH"]
|
|
sde.set_connection(None)
|
|
finally:
|
|
sde.set_connection(None)
|