77 lines
3.1 KiB
Python
77 lines
3.1 KiB
Python
|
A
|
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)
|
||
|
|
|
||
|
|
|
||
|
|
@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_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"
|