A
8672b0de6b
Full working closed-loop eSIM provisioning stack (spec in claude.md / plan/): Flask API, SQLite subscriber store, credential/profile generator, Open5GS and free5gc core adapters, captive portal, operational scripts, systemd units, network configs, and a 44-test suite. Profile generation currently returns a mock JSON profile; real SGP.22 generation via pySim is the planned Phase-2 upgrade. Fixes two bugs found while getting the suite green: - generate_imsi hardcoded a 10-digit MSIN, producing 16-digit IMSIs for a 3-digit MNC; MSIN length is now 15 - len(mcc) - len(mnc). - check_rate_limit compared timestamps as strings across mismatched formats (SQLite CURRENT_TIMESTAMP vs Python isoformat); both sides now normalized via SQLite datetime() so the window check is chronologically correct. Add .gitignore for venv, caches, runtime config.yaml, and databases. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
import pytest
|
|
from datetime import datetime, timedelta
|
|
|
|
from server.database import (
|
|
add_subscriber,
|
|
check_rate_limit,
|
|
export_all,
|
|
get_connection,
|
|
get_subscriber_by_id,
|
|
get_subscriber_by_imsi,
|
|
init_db,
|
|
list_unsynced,
|
|
mark_synced,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def db_conn(tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
init_db(db_path)
|
|
conn = get_connection(db_path)
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
def _sub(**overrides):
|
|
defaults = dict(
|
|
imsi="302720000000001",
|
|
iccid="8910302720000000001",
|
|
ki="aa" * 16,
|
|
opc="bb" * 16,
|
|
mac_address="aa:bb:cc:dd:ee:ff",
|
|
)
|
|
defaults.update(overrides)
|
|
return defaults
|
|
|
|
|
|
class TestCRUD:
|
|
def test_add_and_retrieve_by_imsi(self, db_conn):
|
|
d = _sub()
|
|
sid = add_subscriber(db_conn, **d)
|
|
sub = get_subscriber_by_imsi(db_conn, d["imsi"])
|
|
assert sub is not None
|
|
assert sub.imsi == d["imsi"]
|
|
assert sub.id == sid
|
|
|
|
def test_add_and_retrieve_by_id(self, db_conn):
|
|
d = _sub()
|
|
sid = add_subscriber(db_conn, **d)
|
|
sub = get_subscriber_by_id(db_conn, sid)
|
|
assert sub is not None
|
|
assert sub.iccid == d["iccid"]
|
|
|
|
def test_duplicate_imsi_raises(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
with pytest.raises(ValueError, match="Duplicate"):
|
|
add_subscriber(db_conn, **_sub())
|
|
|
|
def test_duplicate_iccid_raises(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
with pytest.raises(ValueError):
|
|
add_subscriber(db_conn, **_sub(imsi="302720000000002"))
|
|
|
|
def test_not_found_returns_none(self, db_conn):
|
|
assert get_subscriber_by_imsi(db_conn, "999999999999999") is None
|
|
assert get_subscriber_by_id(db_conn, 99999) is None
|
|
|
|
def test_synced_to_core_defaults_false(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
sub = get_subscriber_by_imsi(db_conn, _sub()["imsi"])
|
|
assert sub.synced_to_core is False
|
|
|
|
|
|
class TestRateLimiting:
|
|
def test_first_request_allowed(self, db_conn):
|
|
assert check_rate_limit(db_conn, "aa:bb:cc:dd:ee:ff", 1) is True
|
|
|
|
def test_second_request_blocked(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
assert check_rate_limit(db_conn, "aa:bb:cc:dd:ee:ff", 1) is False
|
|
|
|
def test_different_mac_allowed(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
assert check_rate_limit(db_conn, "11:22:33:44:55:66", 1) is True
|
|
|
|
def test_expired_window_allows_again(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
past = (datetime.utcnow() - timedelta(hours=2)).isoformat()
|
|
db_conn.execute("UPDATE subscribers SET created_at = ?", (past,))
|
|
db_conn.commit()
|
|
assert check_rate_limit(db_conn, "aa:bb:cc:dd:ee:ff", 1) is True
|
|
|
|
|
|
class TestSync:
|
|
def test_list_unsynced_after_insert(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
assert len(list_unsynced(db_conn)) == 1
|
|
|
|
def test_mark_synced_removes_from_unsynced(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
mark_synced(db_conn, _sub()["imsi"])
|
|
assert len(list_unsynced(db_conn)) == 0
|
|
|
|
def test_synced_flag_set(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
mark_synced(db_conn, _sub()["imsi"])
|
|
sub = get_subscriber_by_imsi(db_conn, _sub()["imsi"])
|
|
assert sub.synced_to_core is True
|
|
|
|
|
|
class TestExport:
|
|
def test_export_returns_list_with_correct_keys(self, db_conn):
|
|
add_subscriber(db_conn, **_sub())
|
|
rows = export_all(db_conn)
|
|
assert len(rows) == 1
|
|
for key in ("imsi", "iccid", "ki", "opc", "mac_address", "created_at", "synced_to_core"):
|
|
assert key in rows[0]
|
|
|
|
def test_export_empty_db(self, db_conn):
|
|
assert export_all(db_conn) == []
|