A
0be02bd1f4
- Fix FileNotFoundError when database.path has no directory component (os.makedirs(os.path.dirname(path)) with path="subscribers.db" raised). Add ensure_parent_dir helper; use it in app.py and scripts/init_db.py. - Open5GS/free5gc adapters now use profile.apn for the core session/DNN name instead of a hardcoded "internet"; apn is threaded through get_adapter. Tests: +4 (ensure_parent_dir, adapter APN); 82 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
91 lines
3.8 KiB
Python
91 lines
3.8 KiB
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_returns_none(self):
|
|
cfg = {"core": {"type": "none"}}
|
|
assert get_adapter(cfg) is None
|
|
|
|
def test_missing_core_section_returns_none(self):
|
|
assert get_adapter({}) is None
|
|
|
|
def test_open5gs_type_returns_adapter(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)
|
|
|
|
def test_free5gc_type_returns_adapter(self):
|
|
from server.core_adapters.free5gc import Free5GCAdapter
|
|
cfg = {"core": {"type": "free5gc", "mongodb_uri": "mongodb://localhost:27017", "database_name": "free5gc"}}
|
|
with patch("server.core_adapters.open5gs.MongoClient"):
|
|
adapter = get_adapter(cfg)
|
|
assert isinstance(adapter, Free5GCAdapter)
|
|
|
|
def test_apn_threaded_from_profile_config(self):
|
|
cfg = {"core": {"type": "open5gs"}, "profile": {"apn": "ims"}}
|
|
with patch("server.core_adapters.open5gs.MongoClient"):
|
|
adapter = get_adapter(cfg)
|
|
assert adapter.apn == "ims"
|
|
|
|
|
|
@pytest.fixture
|
|
def open5gs_adapter():
|
|
with patch("server.core_adapters.open5gs.MongoClient"):
|
|
adapter = Open5GSAdapter("mongodb://localhost:27017", "open5gs")
|
|
adapter.db = MagicMock()
|
|
yield adapter
|
|
|
|
|
|
class TestOpen5GSAdapter:
|
|
def test_add_subscriber_calls_replace_one(self, open5gs_adapter):
|
|
open5gs_adapter.add_subscriber("302720000000001", "aa" * 16, "bb" * 16)
|
|
open5gs_adapter.db.subscribers.replace_one.assert_called_once()
|
|
|
|
def test_add_subscriber_document_shape(self, open5gs_adapter):
|
|
open5gs_adapter.add_subscriber("302720000000001", "aabbcc", "ddeeff")
|
|
call_args = open5gs_adapter.db.subscribers.replace_one.call_args
|
|
filter_doc = call_args[0][0]
|
|
sub_doc = call_args[0][1]
|
|
|
|
assert filter_doc == {"imsi": "302720000000001"}
|
|
assert sub_doc["imsi"] == "302720000000001"
|
|
assert sub_doc["security"]["k"] == "aabbcc"
|
|
assert sub_doc["security"]["opc"] == "ddeeff"
|
|
assert sub_doc["security"]["amf"] == "8000"
|
|
assert "slice" in sub_doc
|
|
assert len(sub_doc["slice"]) == 1
|
|
assert sub_doc["slice"][0]["sst"] == 1
|
|
assert sub_doc["slice"][0]["session"][0]["name"] == "internet"
|
|
|
|
def test_add_subscriber_uses_configured_apn(self):
|
|
with patch("server.core_adapters.open5gs.MongoClient"):
|
|
adapter = Open5GSAdapter("mongodb://localhost:27017", "open5gs", apn="ims")
|
|
adapter.db = MagicMock()
|
|
adapter.add_subscriber("302720000000001", "aa", "bb")
|
|
sub_doc = adapter.db.subscribers.replace_one.call_args[0][1]
|
|
assert sub_doc["slice"][0]["session"][0]["name"] == "ims"
|
|
|
|
def test_add_subscriber_upsert_true(self, open5gs_adapter):
|
|
open5gs_adapter.add_subscriber("302720000000001", "aa", "bb")
|
|
call_kwargs = open5gs_adapter.db.subscribers.replace_one.call_args[1]
|
|
assert call_kwargs.get("upsert") is True
|
|
|
|
def test_remove_subscriber(self, open5gs_adapter):
|
|
open5gs_adapter.remove_subscriber("302720000000001")
|
|
open5gs_adapter.db.subscribers.delete_one.assert_called_once_with(
|
|
{"imsi": "302720000000001"}
|
|
)
|
|
|
|
def test_list_subscribers(self, open5gs_adapter):
|
|
open5gs_adapter.db.subscribers.find.return_value = [
|
|
{"imsi": "302720000000001"}
|
|
]
|
|
result = open5gs_adapter.list_subscribers()
|
|
assert len(result) == 1
|
|
assert result[0]["imsi"] == "302720000000001"
|