# 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` ```python 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` ```python 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` ```python 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 ```bash # 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.