ria-ran--sim-drop/plan/10-testing.md

Unit 10 — Testing

Goal

Unit tests covering the database layer, profile generator, and core adapters. Run with pytest.

Files

• tests/test_database.py
• tests/test_profile_generator.py
• tests/test_core_adapters.py

---

tests/test_database.py

import pytest, tempfile, os
from datetime import datetime, timedelta
from unittest.mock import patch
from server.database import init_db, get_connection, add_subscriber, \
    get_subscriber_by_imsi, get_subscriber_by_id, check_rate_limit, \
    list_unsynced, mark_synced, export_all

@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_data(**kwargs):
    defaults = dict(imsi="302720000000001", iccid="8910302720000000001",
                    ki="aa" * 16, opc="bb" * 16, mac_address="aa:bb:cc:dd:ee:ff")
    defaults.update(kwargs)
    return defaults

class TestCRUD:
    def test_add_and_retrieve_by_imsi(self, db_conn):
        d = sub_data()
        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_data()
        sid = add_subscriber(db_conn, **d)
        sub = get_subscriber_by_id(db_conn, sid)
        assert sub.iccid == d["iccid"]

    def test_duplicate_imsi_raises(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        with pytest.raises(ValueError):
            add_subscriber(db_conn, **sub_data())

    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

class TestRateLimiting:
    def test_first_request_allowed(self, db_conn):
        assert check_rate_limit(db_conn, "aa:bb:cc:dd:ee:ff", 1) == True

    def test_second_request_blocked(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        assert check_rate_limit(db_conn, "aa:bb:cc:dd:ee:ff", 1) == False

    def test_different_mac_allowed(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        assert check_rate_limit(db_conn, "11:22:33:44:55:66", 1) == True

    def test_expired_window_allows_again(self, db_conn):
        # Insert a subscriber with a timestamp 2 hours in the past
        add_subscriber(db_conn, **sub_data())
        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) == True

class TestSync:
    def test_list_unsynced(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        assert len(list_unsynced(db_conn)) == 1

    def test_mark_synced(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        mark_synced(db_conn, sub_data()["imsi"])
        assert len(list_unsynced(db_conn)) == 0

class TestExport:
    def test_export_returns_list(self, db_conn):
        add_subscriber(db_conn, **sub_data())
        rows = export_all(db_conn)
        assert len(rows) == 1
        assert "imsi" in rows[0]
        assert "ki" in rows[0]

---

tests/test_profile_generator.py

import pytest
from server.profile_generator import (
    generate_imsi, generate_ki, generate_iccid, derive_opc, create_esim_profile
)

class TestIMSI:
    def test_length(self):
        assert len(generate_imsi("302", "720")) == 15

    def test_prefix(self):
        imsi = generate_imsi("302", "720")
        assert imsi.startswith("302720")

    def test_digits_only(self):
        assert generate_imsi("302", "720").isdigit()

    def test_uniqueness(self):
        imsis = {generate_imsi("302", "720") for _ in range(100)}
        assert len(imsis) > 90  # Very unlikely to get 10+ collisions

class TestKi:
    def test_length(self):
        assert len(generate_ki()) == 16

    def test_randomness(self):
        keys = {generate_ki() for _ in range(50)}
        assert len(keys) == 50

class TestICCID:
    def test_length(self):
        assert len(generate_iccid()) == 19

    def test_digits_only(self):
        assert generate_iccid().isdigit()

    def test_luhn_valid(self):
        iccid = generate_iccid()
        # Verify last digit is valid Luhn check digit
        from server.profile_generator import luhn_checksum
        assert luhn_checksum(iccid[:-1]) == int(iccid[-1])

class TestOPc:
    # Test vector from 3GPP TS 35.208
    KI_HEX = "000102030405060708090a0b0c0d0e0f"
    OP_HEX = "63bfa50ee6523365ff14c1f45f88737d"

    def test_output_length(self):
        ki = bytes.fromhex(self.KI_HEX)
        op = bytes.fromhex(self.OP_HEX)
        opc = derive_opc(ki, op)
        assert len(opc) == 16

    def test_deterministic(self):
        ki = bytes.fromhex(self.KI_HEX)
        op = bytes.fromhex(self.OP_HEX)
        assert derive_opc(ki, op) == derive_opc(ki, op)

    def test_known_vector(self):
        # Pre-computed: AES_Ki(OP) XOR OP
        ki = bytes.fromhex(self.KI_HEX)
        op = bytes.fromhex(self.OP_HEX)
        opc = derive_opc(ki, op)
        # Verify it's not trivially wrong
        assert opc != op
        assert opc != ki

class TestProfile:
    def test_returns_bytes(self):
        ki = generate_ki()
        op = bytes.fromhex("63bfa50ee6523365ff14c1f45f88737d")
        opc = derive_opc(ki, op)
        imsi = generate_imsi("302", "720")
        iccid = generate_iccid()
        profile = create_esim_profile(imsi, ki, opc, iccid, "Test", "Test Carrier")
        assert isinstance(profile, bytes)
        assert len(profile) > 0

---

tests/test_core_adapters.py

import pytest
from unittest.mock import MagicMock, patch
from server.core_adapters import get_adapter
from server.core_adapters.open5gs import Open5GSAdapter

class TestFactory:
    def test_none_type(self):
        cfg = {"core": {"type": "none"}}
        assert get_adapter(cfg) is None

    def test_open5gs_type(self):
        cfg = {"core": {"type": "open5gs", "mongodb_uri": "mongodb://localhost:27017", "database_name": "open5gs"}}
        with patch("server.core_adapters.open5gs.MongoClient"):
            adapter = get_adapter(cfg)
        assert isinstance(adapter, Open5GSAdapter)

class TestOpen5GSAdapter:
    @pytest.fixture
    def adapter(self):
        with patch("server.core_adapters.open5gs.MongoClient") as mock_client:
            a = Open5GSAdapter("mongodb://localhost:27017", "open5gs")
            a.db = MagicMock()
            yield a

    def test_add_subscriber_calls_replace_one(self, adapter):
        adapter.add_subscriber("302720000000001", "aa" * 16, "bb" * 16)
        adapter.db.subscribers.replace_one.assert_called_once()
        call_args = adapter.db.subscribers.replace_one.call_args
        doc = call_args[0][1]
        assert doc["imsi"] == "302720000000001"
        assert "security" in doc
        assert doc["security"]["k"] == "aa" * 16
        assert "slice" in doc
        assert len(doc["slice"]) == 1

    def test_remove_subscriber(self, adapter):
        adapter.remove_subscriber("302720000000001")
        adapter.db.subscribers.delete_one.assert_called_once_with({"imsi": "302720000000001"})

    def test_list_subscribers(self, adapter):
        adapter.db.subscribers.find.return_value = [{"imsi": "302720000000001"}]
        result = adapter.list_subscribers()
        assert len(result) == 1

---

Running Tests

# All tests
pytest tests/ -v

# With coverage
pytest tests/ -v --cov=server --cov-report=term-missing

# Single file
pytest tests/test_profile_generator.py -v

Expected output: all green, ~30 tests.