mirror of
https://gitlab.eurecom.fr/oai/openairinterface5g.git
synced 2026-07-13 04:30:28 +00:00
f02c84fcfc
Expand the 25PRB RFSim CI scenario to validate two PDU sessions with PCF-driven QoS rules and deterministic per-flow traffic ports. Update iperf execution to use `-B`/`-p` from test args. Changes: - Update `container_5g_rfsim_u0_25prb.xml` to add dual-session validation steps, routing setup, and multi-flow UDP iperf for UL/DL with explicit bind/port arguments. - Change `cls_oaicitest.py` to parse bind/port from `iperf_args`, and return a clear error when the iperf client exits non-zero. - Update `nrue.uicc.2pdu.conf` to configure session 2 on `openairinterface` with distinct `nssai_sd` values. - Extend `5g_rfsimulator/mini_nonrf_config.yaml` with PCF endpoints, dual slices, dual DNN entries, and PCF policy paths, and disable local PCC rules in SMF. - Update `5g_rfsimulator_u0_25prb/docker-compose.yaml` to add the `oai-pcf` service, mount policy directories, and add ext-dn route for the second UE subnet. - Add PCF policy data files for this scenario, in : `policies/policy_decisions/policy_decision.yaml` `policies/pcc_rules/pcc_rules.yaml` `policies/qos_data/qos_data.yaml` About the ci-scripts: Co-authored-by: Robert Schmidt <robert.schmidt@openairinterface.org> Signed-off-by: Guido Casati <guido.casati@openairinterface.org>
99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
|
|
|
import sys
|
|
import logging
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
stream=sys.stdout,
|
|
format="[%(asctime)s] %(levelname)8s: %(message)s"
|
|
)
|
|
import os
|
|
os.system(f'rm -rf cmake_targets')
|
|
os.system(f'mkdir -p cmake_targets/log')
|
|
import unittest
|
|
import tempfile
|
|
|
|
sys.path.append('./') # to find OAI imports below
|
|
import cls_oai_html
|
|
from cls_ci_helper import TestCaseCtx
|
|
import cls_oaicitest
|
|
import cls_cmd
|
|
|
|
class TestPingIperf(unittest.TestCase):
|
|
def setUp(self):
|
|
self.html = cls_oai_html.HTMLManagement()
|
|
self.html.testCaseId = "000000"
|
|
self.ci = cls_oaicitest.OaiCiTest()
|
|
self.ci.ue_ids = ["test"]
|
|
self.node = "localhost"
|
|
self.ctx = TestCaseCtx.Default(tempfile.mkdtemp())
|
|
def tearDown(self):
|
|
with cls_cmd.LocalCmd() as c:
|
|
c.run(f'rm -rf {self.ctx.logPath}')
|
|
|
|
def test_ping(self):
|
|
self.ci.ping_args = "-c3"
|
|
self.ci.ping_packetloss_threshold = "0"
|
|
self.ci.svr_id = "test"
|
|
infra_file = "tests/config/infra_ping_iperf.yaml"
|
|
# TODO Should need nothing but options and UE(s) to use
|
|
success = self.ci.Ping(self.ctx, self.node, self.html, infra_file=infra_file)
|
|
self.assertTrue(success)
|
|
|
|
def test_iperf(self):
|
|
# note: needs to be five seconds because Iperf() adds -O 3, so if it is
|
|
# too short, the server is terminated before the client loaded
|
|
# everything
|
|
self.ci.iperf_args = "-u -t 5 -b 1M -R"
|
|
self.ci.svr_id = "test"
|
|
self.ci.svr_node = "localhost"
|
|
self.ci.iperf_packetloss_threshold = "0"
|
|
self.ci.iperf_bitrate_threshold = "0"
|
|
self.ci.iperf_profile = "balanced"
|
|
infra_file = "tests/config/infra_ping_iperf.yaml"
|
|
# TODO Should need nothing but options and UE(s) to use
|
|
success = self.ci.Iperf(self.ctx, self.node, self.html, infra_file=infra_file)
|
|
self.assertTrue(success)
|
|
|
|
def test_iperf2_unidir(self):
|
|
self.ci.iperf_args = "-u -t 5 -b 1M"
|
|
self.ci.svr_id = "test"
|
|
self.ci.svr_node = "localhost"
|
|
self.ci.iperf_packetloss_threshold = "0"
|
|
self.ci.iperf_bitrate_threshold = "0"
|
|
self.ci.iperf_profile = "balanced"
|
|
infra_file = "tests/config/infra_ping_iperf.yaml"
|
|
# TODO Should need nothing but options and UE(s) to use
|
|
success = self.ci.Iperf2_Unidir(self.ctx, self.node, self.html, infra_file=infra_file)
|
|
self.assertTrue(success)
|
|
|
|
def test_iperf_highrate(self):
|
|
# note: needs to be five seconds because Iperf() adds -O 3, so if it is
|
|
# too short, the server is terminated before the client loaded
|
|
# everything
|
|
self.ci.iperf_args = "-u -t 5 -b 1000M -R -O 0"
|
|
self.ci.svr_id = "test"
|
|
self.ci.svr_node = "localhost"
|
|
self.ci.iperf_packetloss_threshold = "0"
|
|
self.ci.iperf_bitrate_threshold = "0"
|
|
self.ci.iperf_profile = "balanced"
|
|
infra_file = "tests/config/infra_ping_iperf.yaml"
|
|
# TODO Should need nothing but options and UE(s) to use
|
|
success = self.ci.Iperf(self.ctx, self.node, self.html, infra_file=infra_file)
|
|
self.assertTrue(success)
|
|
|
|
def test_iperf_new_bindport(self):
|
|
self.ci.iperf_args = "-u -t 5 -b 22M -O 0 -p 10000 -B 127.0.0.3"
|
|
self.ci.svr_id = "test"
|
|
self.ci.svr_node = "localhost"
|
|
self.ci.iperf_packetloss_threshold = "0"
|
|
self.ci.iperf_bitrate_threshold = "0"
|
|
self.ci.iperf_profile = "balanced"
|
|
infra_file = "tests/config/infra_ping_iperf.yaml"
|
|
# TODO Should need nothing but options and UE(s) to use
|
|
success = self.ci.Iperf(self.ctx, self.node, self.html, infra_file=infra_file)
|
|
self.assertTrue(success)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|