Ryan Malloy
1d36729bed
Core modules: - tle.py: NORAD decoding (Alpha-5 + Super-5, matching get_el.c), 3LE/2LE parsing, TLERecord dataclass with epoch-based dedup - config.py: TOML config + env var overlay (XDG-compliant paths) - cache.py: File-based cache with staleness checking - catalog.py: Multi-source merge with MergeStats tracking - regime.py: LEO/MEO/GEO/HEO classification by mean motion Source downloaders (httpx): - celestrak.py: Active catalog + supplemental GP groups - satnogs.py: JSON API with 3LE conversion - spacetrack.py: POST auth flow, bulk GP download Output formatters: - sql.py: pg_orrery-compatible INSERT generation (E'' strings) - tle_file.py: Standard 3LE text output - json_out.py: JSON with orbital metadata and regime CLI (Click + Rich): - download: Cache TLEs from all sources - build: Merge + output SQL/3LE/JSON (pipes to psql) - load: Direct DB load via psycopg (optional [pg] extra) - info: Cache stats and configuration display 58 tests covering NORAD decoding (all 4 encoding cases), parsing, merge/dedup, SQL escaping, regime classification.
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""Tests for output formatters (SQL, 3LE, JSON)."""
|
|
|
|
import json
|
|
|
|
from pg_orrery_catalog.output.json_out import generate_json
|
|
from pg_orrery_catalog.output.sql import escape_sql_string, generate_sql
|
|
from pg_orrery_catalog.output.tle_file import generate_3le
|
|
from pg_orrery_catalog.tle import TLERecord
|
|
|
|
|
|
def _make_record(norad_id: int, name: str = "", epoch: float = 24001.0) -> TLERecord:
|
|
line1 = f"1 {norad_id:05d}U 98067A {epoch:014.8f} .00000000 00000-0 00000-0 0 9990"
|
|
line2 = f"2 {norad_id:05d} 51.6400 100.0000 0007417 30.0000 330.1234 15.49456789999990"
|
|
return TLERecord(
|
|
line1=line1, line2=line2, name=name or f"SAT-{norad_id}",
|
|
norad_id=norad_id, epoch=epoch,
|
|
)
|
|
|
|
|
|
class TestSQLEscaping:
|
|
def test_single_quote(self):
|
|
assert escape_sql_string("it's") == "it''s"
|
|
|
|
def test_backslash(self):
|
|
assert escape_sql_string("a\\b") == "a\\\\b"
|
|
|
|
def test_both(self):
|
|
assert escape_sql_string("it's a\\b") == "it''s a\\\\b"
|
|
|
|
def test_clean(self):
|
|
assert escape_sql_string("HELLO") == "HELLO"
|
|
|
|
|
|
class TestSQLGeneration:
|
|
def test_basic_output(self):
|
|
records = {25544: _make_record(25544, "ISS (ZARYA)")}
|
|
sql = generate_sql(records, table="test_table")
|
|
assert "DROP TABLE IF EXISTS test_table;" in sql
|
|
assert "CREATE TABLE IF NOT EXISTS test_table" in sql
|
|
assert "INSERT INTO test_table" in sql
|
|
assert "ISS (ZARYA)" in sql
|
|
assert "E'" in sql # escape string syntax
|
|
assert "\\n" in sql # newline between TLE lines
|
|
|
|
def test_no_drop(self):
|
|
records = {1: _make_record(1)}
|
|
sql = generate_sql(records, table="t", drop_existing=False)
|
|
assert "DROP TABLE" not in sql
|
|
|
|
def test_sorted_by_norad(self):
|
|
records = {
|
|
99999: _make_record(99999),
|
|
1: _make_record(1),
|
|
25544: _make_record(25544),
|
|
}
|
|
sql = generate_sql(records)
|
|
lines = [line for line in sql.split("\n") if line.startswith("INSERT")]
|
|
assert len(lines) == 3
|
|
# First INSERT should be NORAD 1 (lowest)
|
|
assert "SAT-1" in lines[0]
|
|
|
|
def test_name_escaping(self):
|
|
records = {1: _make_record(1, "O'BRIEN SAT")}
|
|
sql = generate_sql(records)
|
|
assert "O''BRIEN SAT" in sql
|
|
|
|
def test_default_name(self):
|
|
"""When TLERecord has no name, SQL output uses 'NORAD {id}'."""
|
|
rec = _make_record(1)
|
|
rec.name = "" # clear the name to test fallback
|
|
records = {1: rec}
|
|
sql = generate_sql(records)
|
|
assert "NORAD 1" in sql
|
|
|
|
|
|
class TestTLEFileGeneration:
|
|
def test_3le_format(self):
|
|
records = {25544: _make_record(25544, "ISS (ZARYA)")}
|
|
text = generate_3le(records)
|
|
lines = text.strip().split("\n")
|
|
assert len(lines) == 3
|
|
assert lines[0].startswith("0 ")
|
|
assert lines[1].startswith("1 ")
|
|
assert lines[2].startswith("2 ")
|
|
|
|
def test_sorted_output(self):
|
|
records = {
|
|
2: _make_record(2, "B"),
|
|
1: _make_record(1, "A"),
|
|
}
|
|
text = generate_3le(records)
|
|
lines = text.strip().split("\n")
|
|
assert "A" in lines[0]
|
|
|
|
def test_name_prefix(self):
|
|
records = {1: _make_record(1, "0 ALREADY PREFIXED")}
|
|
text = generate_3le(records)
|
|
# Should not double-prefix
|
|
assert "0 0 ALREADY PREFIXED" not in text
|
|
assert "0 ALREADY PREFIXED" in text
|
|
|
|
|
|
class TestJSONGeneration:
|
|
def test_valid_json(self):
|
|
records = {25544: _make_record(25544, "ISS")}
|
|
text = generate_json(records)
|
|
data = json.loads(text)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 1
|
|
assert data[0]["norad_id"] == 25544
|
|
assert data[0]["name"] == "ISS"
|
|
|
|
def test_fields_present(self):
|
|
records = {1: _make_record(1)}
|
|
data = json.loads(generate_json(records))
|
|
entry = data[0]
|
|
for field in ("norad_id", "name", "line1", "line2", "epoch", "mean_motion",
|
|
"inclination", "eccentricity", "regime", "source"):
|
|
assert field in entry
|
|
|
|
def test_regime_classification(self):
|
|
records = {1: _make_record(1)} # mean_motion ~15.49 → LEO
|
|
data = json.loads(generate_json(records))
|
|
assert data[0]["regime"] == "LEO"
|