From f4d0d3f4161f3f74cc6349e52ce3394c502a9a02 Mon Sep 17 00:00:00 2001 From: Erik Gaida Date: Tue, 28 Jul 2026 14:53:45 +0200 Subject: [PATCH 1/3] Added CRUD tests for MBSF APIs --- test/test_mbsf_crud.py | 279 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 279 insertions(+) create mode 100644 test/test_mbsf_crud.py diff --git a/test/test_mbsf_crud.py b/test/test_mbsf_crud.py new file mode 100644 index 0000000..8a9b028 --- /dev/null +++ b/test/test_mbsf_crud.py @@ -0,0 +1,279 @@ +""" +this file is a an end-to-end CRUD test for the MBSF APIs and refers to the Tutorial: +https://hub.5g-mag.com/Getting-Started/pages/5g-multicast-broadcast-services/tutorials/mbsf.html + +Requirements: + pytest + python3-pycurl + +1. Start all needed NFs with: bash ~/rt-mbs-examples/scripts/tmux/mbs-function-tutorial.sh +2. Set the MBSF_BASE_URL to the correct IP address + +Run: + pytest -v -s ~/rt-mbs-examples/test/test_mbsf_crud.py +""" + +from __future__ import annotations + +import copy +import io +import json +import os +from types import SimpleNamespace +from typing import Any +from uuid import uuid4 +import pycurl + + + +MBSF_BASE_URL = "Change this to the IP address of your MBSF instance" + + +USER_SERVICES_URL = f"{MBSF_BASE_URL}/nmbsf-mbs-us/v1/mbs-user-services" +INGEST_SESSIONS_URL = f"{MBSF_BASE_URL}/nmbsf-mbs-ud-ingest/v1/sessions" + +# https://github.com/5G-MAG/Getting-Started/blob/535f72ad57383bb0cbb17528f9c4572ce4756d0a/pages/5g-multicast-broadcast-services/tutorials/mbsf.md?plain=1#L165-L178 +USER_SERVICE_JSON = { + "extServiceIds": ["https://example.broadcaster.com/services/first-service"], + "servType": "MULTICAST", + "servClass": "urn:oma:bcast:oma_bsc:st:1.0", + "servAnnModes": ["VIA_MBS_5", "VIA_MBS_DISTRIBUTION_SESSION", "PASSED_BACK"], + "servNameDescs": [ + { + "servName": "First Service", + "servDescrip": "The first service, operated by Example Broadcaster, is our general entertainment channel", + "language": "eng" + } + ], + "mainServLang": "eng" +} +# ~/rt-mbs-function/tests/json/obj_distr_info.json +STREAMING_INGEST_JSON = { + "mbsUserServId": "", + "mbsDisSessInfos": { + "AP_MBS_SESSION_1": { + "mbsSessionId": { + "ssm": { + "sourceIpAddr": {"ipv4Addr": "127.0.0.5"}, + "destIpAddr": {"ipv4Addr": "232.10.0.5"}, + } + }, + "mbsDistSessState": "INACTIVE", + "maxContBitRate": "10 Mbps", + "distrMethod": "OBJECT", + "objDistrInfo": { + "operatingMode": "STREAMING", + "objAcqMethod": "PULL", + "objAcqIds": ["stream.mpd"], + "objIngUri": ( + "https://livesim2.dashif.org/livesim2/WAVE/" + "vectors/cfhd_sets/12.5_25_50/t1/2022-10-17/" + ), + }, + } + }, +} + +if MBSF_BASE_URL == "Change this to the IP address of your MBSF instance": + raise RuntimeError( + "Set MBSF_BASE_URL to your MBSF URL, for example http://127.0.0.67:7777" + ) + +def _request( + method: str, + url: str, + payload: dict[str, Any] | None = None, +) -> SimpleNamespace: + body = io.BytesIO() + headers: dict[str, str] = {} + + def receive_header(raw_line: bytes) -> int: + line = raw_line.decode("iso-8859-1").strip() + if line.startswith("HTTP/"): + headers.clear() + elif ":" in line: + name, value = line.split(":", 1) + headers[name.strip().lower()] = value.strip() + return len(raw_line) + + curl = pycurl.Curl() + try: + curl.setopt(pycurl.URL, url) + curl.setopt(pycurl.CUSTOMREQUEST, method) + curl.setopt(pycurl.HTTP_VERSION, pycurl.CURL_HTTP_VERSION_2_PRIOR_KNOWLEDGE) + curl.setopt(pycurl.WRITEDATA, body) + curl.setopt(pycurl.HEADERFUNCTION, receive_header) + + if payload is not None: + curl.setopt(pycurl.HTTPHEADER, ["Content-Type: application/json"]) + curl.setopt(pycurl.POSTFIELDS, json.dumps(payload).encode("utf-8")) + + curl.perform() + status = curl.getinfo(pycurl.RESPONSE_CODE) + except pycurl.error as exc: + raise AssertionError( + f"HTTP request failed\nmethod: {method}\nurl: {url}\nerror: {exc}" + ) from exc + finally: + curl.close() + + text = body.getvalue().decode("utf-8", errors="replace") + try: + json_body: Any | None = json.loads(text) if text.strip() else None + except json.JSONDecodeError: + json_body = None + + return SimpleNamespace( + status=status, + headers=headers, + text=text, + json_body=json_body, + ) + + +def test_user_service_crud() -> None: + user_service_url: str | None = None + random_id = uuid4().hex + user_service = copy.deepcopy(USER_SERVICE_JSON) + user_service["extServiceIds"] = [f"urn:uuid:{random_id}"] + + try: + # ========================================================= # + # MBS User Service Operations + # ========================================================= # + # 1. POST: Create the MBS User Service and capture its generated resource ID. + create_service = _request("POST", USER_SERVICES_URL, user_service) + assert create_service.status == 201, create_service.text + assert "location" in create_service.headers, create_service.headers + user_service_id = create_service.headers["location"].rstrip("/").rsplit("/", 1)[-1] + user_service_url = f"{USER_SERVICES_URL}/{user_service_id}" + + assert isinstance(create_service.json_body, dict) + assert create_service.json_body["servType"] == "MULTICAST" + assert create_service.json_body["mainServLang"] == "eng" + + + # 2. PUT: Update and retrieve the MBS User Service. + + updated_user_service = copy.deepcopy(user_service) + updated_user_service["servNameDescs"][0][ + "servDescrip" + ] = "The description has been changed!" + + update_service = _request("PUT", user_service_url, updated_user_service) + assert update_service.status == 200, update_service.text + assert isinstance(update_service.json_body, dict) + assert (update_service.json_body["servNameDescs"][0]["servDescrip"]== "The description has been changed!") + + # 3. GET: Retrieving an MBS User Service + get_service = _request("GET", user_service_url) + assert get_service.status == 200, get_service.text + assert isinstance(get_service.json_body, dict) + assert (get_service.json_body["servNameDescs"][0]["servDescrip"] == "The description has been changed!") + assert get_service.json_body["extServiceIds"] == user_service["extServiceIds"] + + # 4. DELETE: Deleting an MBS User Service + delete_service = _request("DELETE", user_service_url) + assert delete_service.status == 204, delete_service.text + assert _request("GET", user_service_url).status == 404 + user_service_url = None + + finally: + # Remove resources left behind when an assertion fails. + if user_service_url is not None: + try: + _request("DELETE", user_service_url) + except AssertionError: + pass + +def test_ingest_session_crud() -> None: + ingest_user_service_url: str | None = None + ingest_url: str | None = None + random_id = uuid4().hex + + try: + # ========================================================= # + # MBS User Data Ingest Session Operations + # ========================================================= # + # 0. Create a new User Service for the separate ingest-session tutorial. + ingest_user_service = copy.deepcopy(USER_SERVICE_JSON) + ingest_user_service["extServiceIds"] = [f"urn:uuid:{uuid4().hex}"] + create_ingest_service = _request( + "POST", + USER_SERVICES_URL, + ingest_user_service, + ) + assert create_ingest_service.status == 201, create_ingest_service.text + assert "location" in create_ingest_service.headers, create_ingest_service.headers + ingest_user_service_id = create_ingest_service.headers["location"].rstrip( + "/" + ).rsplit("/", 1)[-1] + ingest_user_service_url = f"{USER_SERVICES_URL}/{ingest_user_service_id}" + + # Prepare the MBS User Data Ingest Session payload with unique IP addresses. + ingest_session = copy.deepcopy(STREAMING_INGEST_JSON) + ingest_session["mbsUserServId"] = ingest_user_service_id + ssm = ingest_session["mbsDisSessInfos"]["AP_MBS_SESSION_1"][ + "mbsSessionId" + ]["ssm"] + ssm["sourceIpAddr"]["ipv4Addr"] = ( + f"127.{int(random_id[0:2], 16)}.{int(random_id[2:4], 16)}." + f"{int(random_id[4:6], 16)}" + ) + ssm["destIpAddr"]["ipv4Addr"] = ( + f"232.{int(random_id[6:8], 16)}.{int(random_id[8:10], 16)}." + f"{int(random_id[10:12], 16)}" + ) + + # 1. POST: Create the associated MBS User Data Ingest Session. + create_ingest = _request("POST", INGEST_SESSIONS_URL, ingest_session) + assert create_ingest.status == 201, create_ingest.text + assert "location" in create_ingest.headers, create_ingest.headers + ingest_id = create_ingest.headers["location"].rstrip("/").rsplit("/", 1)[-1] + ingest_url = f"{INGEST_SESSIONS_URL}/{ingest_id}" + assert isinstance(create_ingest.json_body, dict) + assert create_ingest.json_body["mbsUserServId"] == ingest_user_service_id + assert (create_ingest.json_body["mbsDisSessInfos"]["AP_MBS_SESSION_1"]["mbsDistSessState"] == "INACTIVE") + + # 2. PUT: Update an MBS User Data Ingest Session + updated_ingest_session = copy.deepcopy(ingest_session) + updated_ingest_session["mbsDisSessInfos"]["AP_MBS_SESSION_1"][ + "mbsDistSessState" + ] = "ACTIVE" + + update_ingest = _request("PUT", ingest_url, updated_ingest_session) + assert update_ingest.status == 200, update_ingest.text + assert isinstance(update_ingest.json_body, dict) + assert ( + update_ingest.json_body["mbsDisSessInfos"]["AP_MBS_SESSION_1"][ + "mbsDistSessState" + ] + == "ACTIVE" + ) + # 3. GET: Retrieve an MBS User Data Ingest Session + get_ingest = _request("GET", ingest_url) + assert get_ingest.status == 200, get_ingest.text + assert isinstance(get_ingest.json_body, dict) + assert get_ingest.json_body["mbsUserServId"] == ingest_user_service_id + assert (get_ingest.json_body["mbsDisSessInfos"]["AP_MBS_SESSION_1"]["mbsDistSessState"] == "ACTIVE") + + # 4. DELETE: Delete an MBS User Data Ingest Session + delete_ingest = _request("DELETE", ingest_url) + assert delete_ingest.status == 204, delete_ingest.text + assert _request("GET", ingest_url).status == 404 + ingest_url = None + + # Deleting the MBS User Service Session Created in step 0. + delete_service = _request("DELETE", ingest_user_service_url) + assert delete_service.status == 204, delete_service.text + assert _request("GET", ingest_user_service_url).status == 404 + ingest_user_service_url = None + + finally: + # Remove resources left behind when an assertion fails. + for url in (ingest_url, ingest_user_service_url): + if url is not None: + try: + _request("DELETE", url) + except AssertionError: + pass From 293945fffd3a5ede83a69de92e97367351e13cb6 Mon Sep 17 00:00:00 2001 From: Erik Gaida Date: Wed, 29 Jul 2026 10:39:39 +0200 Subject: [PATCH 2/3] Update config and test script for MBSF API integration --- test/config.toml | 5 +++++ test/test_mbsf_crud.py | 26 ++++++++++++++++++-------- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/test/config.toml b/test/config.toml index 4e0294a..e9706ee 100644 --- a/test/config.toml +++ b/test/config.toml @@ -5,3 +5,8 @@ log_level = "DEBUG" address = "smf-mb-smf.5g-mag.org" protocol = "http" port = 80 + +[mbsf] +address = "127.0.0.67" +protocol = "http" +port = 7777 diff --git a/test/test_mbsf_crud.py b/test/test_mbsf_crud.py index 8a9b028..563c90d 100644 --- a/test/test_mbsf_crud.py +++ b/test/test_mbsf_crud.py @@ -1,4 +1,16 @@ """ + + +License: 5G-MAG Public License (v1.0) +Author: Erik Gaida +Copyright: (C) 2026 Fraunhofer FOKUS +For full license terms please see the LICENSE file distributed with this +program. If this file is missing then the license can be retrieved from +https://drive.google.com/file/d/1cinCiA778IErENZ3JN52VFW-1ffHpx7Z/view + + + + this file is a an end-to-end CRUD test for the MBSF APIs and refers to the Tutorial: https://hub.5g-mag.com/Getting-Started/pages/5g-multicast-broadcast-services/tutorials/mbsf.html @@ -7,7 +19,7 @@ python3-pycurl 1. Start all needed NFs with: bash ~/rt-mbs-examples/scripts/tmux/mbs-function-tutorial.sh -2. Set the MBSF_BASE_URL to the correct IP address +2. Set the MBSF address, protocol, and port in config.toml under [mbsf] Run: pytest -v -s ~/rt-mbs-examples/test/test_mbsf_crud.py @@ -18,15 +30,18 @@ import copy import io import json -import os +from pathlib import Path from types import SimpleNamespace from typing import Any +from utils import config from uuid import uuid4 + import pycurl -MBSF_BASE_URL = "Change this to the IP address of your MBSF instance" +CONFIG = config.config_from_file(Path(__file__).with_name("config.toml")) +MBSF_BASE_URL = (f"{CONFIG['mbsf']['protocol']}://{CONFIG['mbsf']['address']}:"f"{CONFIG['mbsf']['port']}") USER_SERVICES_URL = f"{MBSF_BASE_URL}/nmbsf-mbs-us/v1/mbs-user-services" @@ -74,11 +89,6 @@ }, } -if MBSF_BASE_URL == "Change this to the IP address of your MBSF instance": - raise RuntimeError( - "Set MBSF_BASE_URL to your MBSF URL, for example http://127.0.0.67:7777" - ) - def _request( method: str, url: str, From 50f06f871c89c936051d96774cf6e63023da98ed Mon Sep 17 00:00:00 2001 From: Erik Gaida Date: Wed, 29 Jul 2026 11:30:36 +0200 Subject: [PATCH 3/3] switch from pycurl to httpcore for HTTP requests --- test/test_mbsf_crud.py | 56 ++++++++++++++---------------------------- 1 file changed, 19 insertions(+), 37 deletions(-) diff --git a/test/test_mbsf_crud.py b/test/test_mbsf_crud.py index 563c90d..beb958b 100644 --- a/test/test_mbsf_crud.py +++ b/test/test_mbsf_crud.py @@ -16,8 +16,7 @@ Requirements: pytest - python3-pycurl - + httpcore 1. Start all needed NFs with: bash ~/rt-mbs-examples/scripts/tmux/mbs-function-tutorial.sh 2. Set the MBSF address, protocol, and port in config.toml under [mbsf] @@ -28,21 +27,21 @@ from __future__ import annotations import copy -import io import json from pathlib import Path from types import SimpleNamespace from typing import Any -from utils import config from uuid import uuid4 -import pycurl +import httpcore +from utils import config +from utils.http2_prior_knowledge import setup_http2_pool CONFIG = config.config_from_file(Path(__file__).with_name("config.toml")) MBSF_BASE_URL = (f"{CONFIG['mbsf']['protocol']}://{CONFIG['mbsf']['address']}:"f"{CONFIG['mbsf']['port']}") - +HTTP2_POOL = setup_http2_pool() USER_SERVICES_URL = f"{MBSF_BASE_URL}/nmbsf-mbs-us/v1/mbs-user-services" INGEST_SESSIONS_URL = f"{MBSF_BASE_URL}/nmbsf-mbs-ud-ingest/v1/sessions" @@ -89,53 +88,36 @@ }, } +def _decode_headers(raw_headers: list[tuple[bytes, bytes]]) -> dict[str, str]: + return { + name.decode("iso-8859-1").lower(): value.decode("iso-8859-1") + for name, value in raw_headers + } + + def _request( method: str, url: str, payload: dict[str, Any] | None = None, ) -> SimpleNamespace: - body = io.BytesIO() - headers: dict[str, str] = {} - - def receive_header(raw_line: bytes) -> int: - line = raw_line.decode("iso-8859-1").strip() - if line.startswith("HTTP/"): - headers.clear() - elif ":" in line: - name, value = line.split(":", 1) - headers[name.strip().lower()] = value.strip() - return len(raw_line) - - curl = pycurl.Curl() + headers = {"Content-Type": "application/json"} if payload is not None else None + content = json.dumps(payload).encode("utf-8") if payload is not None else None try: - curl.setopt(pycurl.URL, url) - curl.setopt(pycurl.CUSTOMREQUEST, method) - curl.setopt(pycurl.HTTP_VERSION, pycurl.CURL_HTTP_VERSION_2_PRIOR_KNOWLEDGE) - curl.setopt(pycurl.WRITEDATA, body) - curl.setopt(pycurl.HEADERFUNCTION, receive_header) - - if payload is not None: - curl.setopt(pycurl.HTTPHEADER, ["Content-Type: application/json"]) - curl.setopt(pycurl.POSTFIELDS, json.dumps(payload).encode("utf-8")) - - curl.perform() - status = curl.getinfo(pycurl.RESPONSE_CODE) - except pycurl.error as exc: + response = HTTP2_POOL.request(method, url, headers=headers, content=content) + except (httpcore.NetworkError, httpcore.ProtocolError) as exc: raise AssertionError( f"HTTP request failed\nmethod: {method}\nurl: {url}\nerror: {exc}" ) from exc - finally: - curl.close() - text = body.getvalue().decode("utf-8", errors="replace") + text = response.content.decode("utf-8", errors="replace") try: json_body: Any | None = json.loads(text) if text.strip() else None except json.JSONDecodeError: json_body = None return SimpleNamespace( - status=status, - headers=headers, + status=response.status, + headers=_decode_headers(response.headers), text=text, json_body=json_body, )