ria-ran--sim-drop/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_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"