ria-ran--sim-drop/tests/test_app.py

import json

import pytest
import yaml

from server.app import create_app


def _write_config(tmp_path, admin_token=""):
    cfg = {
        "network": {"mcc": "302", "mnc": "720", "plmn": "302720",
                    "op_key": "63bfa50ee6523365ff14c1f45f88737d"},
        "wifi": {"ssid": "Emergency-5G", "channel": 6, "ip": "192.168.4.1",
                 "subnet": "192.168.4.0/24", "dhcp_range": "192.168.4.10,192.168.4.250"},
        "server": {"host": "0.0.0.0", "port": 5000, "debug": False,
                   "admin_token": admin_token},
        "database": {"path": str(tmp_path / "subscribers.db")},
        "rate_limiting": {"max_profiles_per_mac": 1, "window_hours": 1},
        "core": {"type": "none", "mongodb_uri": "mongodb://localhost:27017",
                 "database_name": "open5gs"},
        "profile": {"carrier_name": "Ad-Hoc Network", "profile_name": "Emergency 5G",
                    "apn": "internet", "smdp_address": "esim.local"},
    }
    path = tmp_path / "config.yaml"
    path.write_text(yaml.safe_dump(cfg))
    return str(path)


@pytest.fixture
def client(tmp_path):
    app = create_app(_write_config(tmp_path))
    app.config.update(TESTING=True)
    return app.test_client()


@pytest.fixture
def admin_client(tmp_path):
    app = create_app(_write_config(tmp_path, admin_token="s3cret-token"))
    app.config.update(TESTING=True)
    return app.test_client()


class TestRoutes:
    def test_index_redirects_to_activate(self, client):
        r = client.get("/")
        assert r.status_code == 302
        assert "/activate" in r.headers["Location"]

    def test_status_ok(self, client):
        data = client.get("/api/status").get_json()
        assert data["status"] == "ok"
        assert data["core_adapter"] == "none"

    def test_export_disabled_without_admin_token(self, client):
        # No server.admin_token configured -> export is disabled.
        assert client.get("/api/export").status_code == 403

    def test_export_requires_matching_token(self, admin_client):
        assert admin_client.get("/api/export").status_code == 401
        assert admin_client.get(
            "/api/export", headers={"X-Admin-Token": "wrong"}
        ).status_code == 401

    def test_export_with_valid_token(self, admin_client):
        r = admin_client.get("/api/export", headers={"X-Admin-Token": "s3cret-token"})
        assert r.status_code == 200
        data = r.get_json()
        assert data["count"] == 0
        assert data["subscribers"] == []


class TestProvision:
    def test_provision_success(self, client):
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:11:22:33"})
        assert r.status_code == 200
        body = r.get_json()
        assert len(body["imsi"]) == 15
        assert body["profile_url"].startswith("/api/profile/")

    def test_provision_missing_mac(self, client):
        # No body and (in tests) no resolvable ARP entry -> 400
        r = client.post("/api/provision", json={})
        assert r.status_code == 400

    def test_provision_invalid_mac(self, client):
        r = client.post("/api/provision", json={"mac_address": "not-a-mac"})
        assert r.status_code == 400

    def test_oversized_body_rejected(self, client):
        big = '{"mac_address":"' + "a" * (17 * 1024) + '"}'
        r = client.post("/api/provision", data=big,
                        content_type="application/json")
        assert r.status_code == 413

    def test_rate_limited_second_request(self, client):
        client.post("/api/provision", json={"mac_address": "de:ad:be:ef:00:01"})
        r = client.post("/api/provision", json={"mac_address": "de:ad:be:ef:00:01"})
        assert r.status_code == 429

    def test_provision_retries_on_collision(self, client, monkeypatch):
        import server.app as appmod
        real_add = appmod.add_subscriber
        calls = {"n": 0}

        def flaky(conn, imsi, iccid, ki, opc, mac):
            calls["n"] += 1
            if calls["n"] == 1:
                raise ValueError("Duplicate subscriber")
            return real_add(conn, imsi, iccid, ki, opc, mac)

        monkeypatch.setattr(appmod, "add_subscriber", flaky)
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:ab:cd:ef"})
        assert r.status_code == 200
        assert calls["n"] == 2  # failed once, succeeded on retry

    def test_provision_gives_up_after_persistent_collision(self, client, monkeypatch):
        import server.app as appmod

        def always_dup(*a, **k):
            raise ValueError("Duplicate subscriber")

        monkeypatch.setattr(appmod, "add_subscriber", always_dup)
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:ab:cd:11"})
        assert r.status_code == 500


class TestProfileDownload:
    def test_download_with_token(self, client):
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:44:55:66"})
        url = r.get_json()["profile_url"]
        d = client.get(url)
        assert d.status_code == 200
        profile = json.loads(d.data)
        assert profile["type"] == "profile-package-v2"

    def test_unknown_token_is_404(self, client):
        assert client.get("/api/profile/does-not-exist").status_code == 404

    def test_sequential_id_not_downloadable(self, client):
        # The old enumerable /api/profile/<int> scheme must be gone: the row id
        # is not a valid download token.
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:77:88:99"})
        sub_id = r.get_json()["id"]
        assert client.get(f"/api/profile/{sub_id}").status_code == 404