Skip to content

slapos
slapos
Commit cf27eece authored by Lu Xu's avatar Lu Xu 👀
Browse files
Options

fixup

parent b1b29b98
Hide whitespace changes
Inline Side-by-side
Showing with 458 additions and 87 deletions
software/end-to-end-testing/tests/test_new_lopcomm.py → software/end-to-end-testing/tests/test_master_lopcomm.py
View file @ cf27eece
import json
import time
import slapos.testing.e2e as e2e
from websocket import create_connection
import slapos.testing.e2e as e2e
class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod
......@@ -9,51 +9,38 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
try:
super().setUpClass()
cls.enb_instance_name = time.strftime('e2e-cb003-enb-new-%Y-%B-%d-%H:%M:%S')
cls.ru1_instance_name = time.strftime('e2e-cb003-ru1-new-%Y-%B-%d-%H:%M:%S')
cls.cell1_instance_name = time.strftime('e2e-cb003-cell1-new-%Y-%B-%d-%H:%M:%S')
cls.cn_instance_name = time.strftime('e2e-cb003-core-network-new-%Y-%B-%d-%H:%M:%S')
cls.sim_instance_name = time.strftime('e2e-cb003-sim-new-%Y-%B-%d-%H:%M:%S')
cls.ue_instance_name = time.strftime('e2e-sb005-ue-new-%Y-%B-%d-%H:%M:%S')
cls.ue_cell1_instance_name = time.strftime('e2e-sb005-ue-cell1-new-%Y-%B-%d-%H:%M:%S')
cls.ue_sim_instance_name = time.strftime('e2e-sb005-ue-sim-new-%Y-%B-%d-%H:%M:%S')
# Dynamically named instances for different runs
cls.set_instance_names()
cls.product = "https://lab.nexedi.com/nexedi/slapos/-/raw/master/software/ors-amarisoft/software.cfg"
# Component GUIDs and configurations
cls.comp_enb ="COMP-3920"
cls.comp_cn = "COMP-3920"
cls.comp_ue = "COMP-3756"
cls.dl_earfcn = 300
# Retry configurations
cls.max_retries = 10
cls.retry_delay = 180 # seconds
cls.setup_guids_and_configs()
# Setup instances
cls.setup_instances()
cls.waitUntilGreen(cls.enb_instance_name)
cls.waitUntilGreen(cls.cn_instance_name)
except Exception as e:
cls.logger.error("Error during setup: " + str(e))
# Ensure cleanup
cls.logger.error("Error during setup: %s", str(e))
cls.tearDownClass()
raise
@classmethod
def retry_request(cls, func, *args, **kwargs):
for attempt in range(cls.max_retries):
try:
result = func(*args, **kwargs)
return result
except Exception as e:
cls.logger.error(f"Error on attempt {attempt + 1}: {e}")
if attempt < cls.max_retries - 1:
time.sleep(cls.retry_delay)
return None
def set_instance_names(cls):
current_time = time.strftime('%Y-%B-%d-%H:%M:%S')
ors_branch = 'master'
base_names = ['enb', 'eru1', 'ecell1', 'cn', 'sim', 'ue', 'ucell1', 'usim']
for name in base_names:
setattr(cls, f"{name}_instance_name", f"E2E-{ors_branch}-{name}-{current_time}")
@classmethod
def setup_guids_and_configs(cls):
cls.comp_enb ="COMP-3920"
cls.comp_cn = "COMP-3920"
cls.comp_ue = "COMP-3756"
cls.dl_earfcn = 300
cls.max_retries = 10
cls.retry_delay = 180
@classmethod
def setup_instances(cls):
......@@ -64,17 +51,34 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
cls.setup_websocket_connection()
@classmethod
def wait_for_attribute(cls, attr_name, timeout=600):
def retry_request(cls, func, *args, **kwargs):
for attempt in range(cls.max_retries):
try:
result = func(*args, **kwargs)
if result:
return result
cls.logger.info(f"Attempt {attempt + 1}: Received empty or invalid result, retrying...")
except Exception as e:
cls.logger.error("Error on attempt %d: %s", attempt + 1, e)
if attempt < cls.max_retries - 1:
cls.logger.info("Retrying...")
time.sleep(cls.retry_delay)
cls.logger.warning("All retry attempts failed.")
return None
@classmethod
def wait_for_attribute(cls, attr_name, timeout=3600):
start_time = time.time()
while not hasattr(cls, attr_name) or getattr(cls, attr_name) is None:
if time.time() - start_time > timeout:
raise TimeoutError(f"Timeout waiting for attribute '{attr_name}' to become available.")
time.sleep(1)
time.sleep(cls.retry_delay)
cls.logger.info(f"Attribute '{attr_name}' is now available.")
# Request ENB/RU/Cell
@classmethod
def request_enb(cls, custom_params=None):
cls.logger.info("Request "+ cls.enb_instance_name)
cls.logger.info("Request %s", cls.enb_instance_name)
enb_parameters = {
"enb_id": "0x1A2D0",
"mme_list": {'1': {'mme_addr': cls.cn_ipv6 }},
......@@ -104,7 +108,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod
def request_ru1(cls):
cls.logger.info("Request "+ cls.ru1_instance_name)
cls.logger.info("Request %s", cls.eru1_instance_name)
ru1_parameters = {
'ru_type': 'lopcomm',
'ru_link_type': 'cpri',
......@@ -127,15 +131,15 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_ru1_parameters = json.dumps(ru1_parameters)
cls.retry_request(cls.request, cls.product, cls.ru1_instance_name,
filter_kw={"instance_guid": cls.enb_instance_guid},
partition_parameter_kw={'_': json_ru1_parameters},
shared=True,
software_type='enb')
cls.request(cls.product, cls.eru1_instance_name,
filter_kw={"instance_guid": cls.enb_instance_guid},
partition_parameter_kw={'_': json_ru1_parameters},
shared=True,
software_type='enb')
@classmethod
def request_cell1(cls):
cls.logger.info("Request "+ cls.cell1_instance_name)
cls.logger.info("Request %s", cls.ecell1_instance_name)
cell1_parameters = {
'cell_type': 'lte',
'cell_kind': 'enb',
......@@ -147,21 +151,22 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
'tac': '0x1234',
'ru': {
'ru_type': 'ru_ref',
'ru_ref': cls.ru1_instance_name
'ru_ref': cls.eru1_instance_name
}
}
json_cell1_parameters = json.dumps(cell1_parameters)
cls.retry_request(cls.request, cls.product, cls.cell1_instance_name,
partition_parameter_kw={'_': json_cell1_parameters},
software_type='enb',
filter_kw={"instance_guid": cls.enb_instance_guid},
shared=True, state='started')
cls.request(cls.product, cls.ecell1_instance_name,
partition_parameter_kw={'_': json_cell1_parameters},
software_type='enb',
filter_kw={"instance_guid": cls.enb_instance_guid},
shared=True, state='started')
#Request Core Network/SIM Card
@classmethod
def request_core_network(cls):
cls.logger.info("Request "+ cls.cn_instance_name)
cls.logger.info("Request %s", cls.cn_instance_name)
core_network_parameters = json.dumps({
"core_network_plmn": "50501",
'external_enb_gnb': True,
......@@ -184,46 +189,31 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod
def request_demo_sim_cards(cls):
cls.logger.info("Request "+ cls.sim_instance_name)
cls.logger.info("Request %s", cls.sim_instance_name)
if cls.cn_instance_guid is None:
cls.logger.error("Core network instance GUID not set. Cannot request demo SIM cards.")
return
sim_card_parameters = json.dumps({
sim_card_parameters = {
"sim_algo": "xor",
"imsi": "505010123456789",
"k": "00112233445566778899aabbccddeeff",
"imeisv": "8682430000000101",
"impi": "505010123456789@ims.mnc505.mcc001.3gppnetwork.org",
"impu": ["505010123456789", "tel:0600000000", "tel:600"]
})
}
json_sim_card_parameters = json.dumps(sim_card_parameters)
cls.retry_request(cls.request, cls.product, cls.sim_instance_name,
partition_parameter_kw={'_': sim_card_parameters},
software_type='core-network',
filter_kw={"instance_guid": cls.cn_instance_guid},
shared=True, state='started')
@classmethod
def setup_websocket_connection(cls):
cls.waitUntilGreen(cls.ue_instance_name)
cls.wait_for_attribute('ue_com_addr')
cls.ws_url = f"ws://{cls.ue_com_addr}"
cls.logger.info(f"Websocket URL: {cls.ws_url}")
for attempt in range(cls.max_retries):
try:
cls.ws = create_connection(cls.ws_url)
cls.logger.info("Websocket connection established.")
break
except Exception as e:
cls.logger.error(f"Websocket connection attempt {attempt + 1} failed: {e}")
if attempt < cls.max_retries - 1:
time.sleep(5)
partition_parameter_kw={'_': json_sim_card_parameters},
software_type='core-network',
filter_kw={"instance_guid": cls.cn_instance_guid},
shared=True, state='started')
# Request UE/Cell/SIM
@classmethod
def request_ue(cls):
cls.logger.info("Request "+ cls.ue_instance_name)
cls.logger.info("Request %s", cls.ue_instance_name)
cls.retry_request(cls.request_ue_with_guid)
@classmethod
......@@ -235,6 +225,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
instance_infos = cls.getInstanceInfos(cls.ue_instance_name)
cls.ue_instance_guid = instance_infos.news['instance'][0]['reference']
cls.ue_com_addr = instance_infos.connection_dict.get('com_addr')
cls.rue_addr = instance_infos.connection_dict.get('rue_bind_addr')
cls.request_ue_cell1()
cls.request_ue_sim()
return True
......@@ -242,8 +233,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod
def request_ue_cell1(cls):
cls.logger.info("Request "+ cls.ue_cell1_instance_name)
ue_cell1_parameters = {
cls.logger.info("Request %s", cls.ucell1_instance_name)
ucell1_parameters = {
'cell_type': 'lte',
'cell_kind': 'ue',
'rf_mode': 'fdd',
......@@ -261,34 +252,53 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
}
}
json_ue_cell1_parameters = json.dumps(ue_cell1_parameters)
json_ucell1_parameters = json.dumps(ucell1_parameters)
cls.retry_request(cls.request, cls.product, cls.ue_cell1_instance_name,
partition_parameter_kw={'_': json_ue_cell1_parameters},
cls.retry_request(cls.request, cls.product, cls.ucell1_instance_name,
partition_parameter_kw={'_': json_ucell1_parameters},
software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started')
@classmethod
def request_ue_sim(cls):
cls.logger.info("Request "+ cls.ue_sim_instance_name)
ue_sim_parameters = json.dumps({
cls.logger.info("Request %s", cls.usim_instance_name)
cls.wait_for_attribute('rue_addr')
usim_parameters = {
"ue_type": "lte",
"rue_addr": "2a11:9ac0:d::e41c",
"rue_addr": cls.rue_addr,
"sim_algo": "xor",
"imsi": "505010123456789",
"k": "00112233445566778899aabbccddeeff",
"imeisv": "8682430000000101",
"impi": "505010123456789@ims.mnc505.mcc001.3gppnetwork.org",
"impu": ["505010123456789", "tel:0600000000", "tel:600"]
})
}
cls.retry_request(cls.request, cls.product, cls.ue_sim_instance_name,
partition_parameter_kw={'_': ue_sim_parameters},
json_usim_parameters = json.dumps(usim_parameters)
cls.request(cls.product, cls.usim_instance_name,
partition_parameter_kw={'_': json_usim_parameters},
software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started')
@classmethod
def setup_websocket_connection(cls):
# cls.waitUntilGreen(cls.ue_instance_name)
cls.wait_for_attribute('ue_com_addr')
cls.ws_url = f"ws://{cls.ue_com_addr}"
cls.logger.info(f"Websocket URL: {cls.ws_url}")
for attempt in range(cls.max_retries):
try:
cls.ws = create_connection(cls.ws_url)
cls.logger.info("Websocket connection established.")
break
except Exception as e:
cls.logger.error(f"Websocket connection attempt {attempt + 1} failed: {e}")
if attempt < cls.max_retries - 1:
time.sleep(5)
@classmethod
def tearDownClass(cls):
if hasattr(cls, 'ws') and cls.ws is not None:
......@@ -344,7 +354,7 @@ class BBUTest(WebsocketTestClass):
def test_max_rx_sample_db(self):
custom_params = {"max_rx_sample_db": -99}
BBUTest.request_enb(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name=BBUTest.ru1_instance_name + "-rx-saturated", expected=False)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-rx-saturated", expected=False)
def test_min_rxtx_delay(self):
custom_params = {"min_rxtx_delay": 99}
......
software/end-to-end-testing/tests/test_ors_fix_lopcomm.py 0 → 100644
View file @ cf27eece
import json
import time
from websocket import create_connection
import slapos.testing.e2e as e2e
class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod
def setUpClass(cls):
try:
super().setUpClass()
# Dynamically named instances for different runs
cls.set_instance_names()
cls.product = "https://lab.nexedi.com/lu.xu/slapos/-/raw/ors-fix/software/ors-amarisoft/software.cfg"
cls.setup_guids_and_configs()
# Setup instances
cls.setup_instances()
cls.waitUntilGreen(cls.enb_instance_name)
cls.waitUntilGreen(cls.cn_instance_name)
except Exception as e:
cls.logger.error("Error during setup: %s", str(e))
cls.tearDownClass()
raise
@classmethod
def set_instance_names(cls):
current_time = time.strftime('%Y-%B-%d-%H:%M:%S')
ors_branch = 'master'
base_names = ['enb', 'eru1', 'ecell1', 'cn', 'sim', 'ue', 'ucell1', 'usim']
for name in base_names:
setattr(cls, f"{name}_instance_name", f"E2E-{ors_branch}-{name}-{current_time}")
@classmethod
def setup_guids_and_configs(cls):
cls.comp_enb ="COMP-3920"
cls.comp_cn = "COMP-3920"
cls.comp_ue = "COMP-3756"
cls.dl_earfcn = 300
cls.max_retries = 10
cls.retry_delay = 180
@classmethod
def setup_instances(cls):
cls.request_core_network()
cls.wait_for_attribute('cn_ipv6')
cls.request_enb()
cls.request_ue()
cls.setup_websocket_connection()
@classmethod
def retry_request(cls, func, *args, **kwargs):
for attempt in range(cls.max_retries):
try:
result = func(*args, **kwargs)
if result:
return result
cls.logger.info(f"Attempt {attempt + 1}: Received empty or invalid result, retrying...")
except Exception as e:
cls.logger.error("Error on attempt %d: %s", attempt + 1, e)
if attempt < cls.max_retries - 1:
cls.logger.info("Retrying...")
time.sleep(cls.retry_delay)
cls.logger.warning("All retry attempts failed.")
return None
@classmethod
def wait_for_attribute(cls, attr_name, timeout=3600):
start_time = time.time()
while not hasattr(cls, attr_name) or getattr(cls, attr_name) is None:
if time.time() - start_time > timeout:
raise TimeoutError(f"Timeout waiting for attribute '{attr_name}' to become available.")
time.sleep(cls.retry_delay)
cls.logger.info(f"Attribute '{attr_name}' is now available.")
# Request ENB/RU/Cell
@classmethod
def request_enb(cls, custom_params=None):
cls.logger.info("Request %s", cls.enb_instance_name)
enb_parameters = {
"enb_id": "0x1A2D0",
"mme_list": {'1': {'mme_addr': cls.cn_ipv6 }},
"plmn_list": {"Australia": {"plmn": "50501"}}
}
if custom_params:
enb_parameters.update(custom_params)
json_enb_parameters = json.dumps(enb_parameters)
cls.retry_request(cls.request_enb_with_guid, json_enb_parameters)
@classmethod
def request_enb_with_guid(cls, json_enb_parameters):
enb_instance = cls.request(cls.product, cls.enb_instance_name,
filter_kw={"computer_guid": cls.comp_enb},
partition_parameter_kw={'_': json_enb_parameters},
software_type='enb')
if enb_instance:
instance_infos = cls.getInstanceInfos(cls.enb_instance_name)
cls.enb_instance_guid = instance_infos.news['instance'][0]['reference']
cls.request_ru1()
cls.request_cell1()
return True
return False
@classmethod
def request_ru1(cls):
cls.logger.info("Request %s", cls.eru1_instance_name)
ru1_parameters = {
'ru_type': 'lopcomm',
'ru_link_type': 'cpri',
'cpri_link': {
'sdr_dev': 0,
'sfp_port': 0,
'mult': 16,
'mapping': 'hw',
'rx_delay': 25.11,
'tx_delay': 13.77,
'tx_dbm': 56
},
'n_antenna_dl': 1,
'n_antenna_ul': 1,
'tx_gain': -20,
'rx_gain': -10,
'txrx_active': 'ACTIVE',
}
json_ru1_parameters = json.dumps(ru1_parameters)
cls.request(cls.product, cls.eru1_instance_name,
filter_kw={"instance_guid": cls.enb_instance_guid},
partition_parameter_kw={'_': json_ru1_parameters},
shared=True,
software_type='enb')
@classmethod
def request_cell1(cls):
cls.logger.info("Request %s", cls.ecell1_instance_name)
cell1_parameters = {
'cell_type': 'lte',
'cell_kind': 'enb',
'rf_mode': 'fdd',
'bandwidth': 20,
'dl_earfcn': cls.dl_earfcn,
'pci': 1,
'cell_id': '0x01',
'tac': '0x1234',
'ru': {
'ru_type': 'ru_ref',
'ru_ref': cls.eru1_instance_name
}
}
json_cell1_parameters = json.dumps(cell1_parameters)
cls.request(cls.product, cls.ecell1_instance_name,
partition_parameter_kw={'_': json_cell1_parameters},
software_type='enb',
filter_kw={"instance_guid": cls.enb_instance_guid},
shared=True, state='started')
#Request Core Network/SIM Card
@classmethod
def request_core_network(cls):
cls.logger.info("Request %s", cls.cn_instance_name)
core_network_parameters = json.dumps({
"core_network_plmn": "50501",
'external_enb_gnb': True,
})
cls.retry_request(cls.request_core_network_with_guid, core_network_parameters)
@classmethod
def request_core_network_with_guid(cls, core_network_parameters):
core_network_instance = cls.request(cls.product, cls.cn_instance_name,
filter_kw={"computer_guid": cls.comp_cn},
partition_parameter_kw={'_': core_network_parameters},
software_type='core-network')
if core_network_instance:
instance_infos = cls.getInstanceInfos(cls.cn_instance_name)
cls.cn_instance_guid = instance_infos.news['instance'][0]['reference']
cls.cn_ipv6 = instance_infos.connection_dict.get('core-network-ipv6')
cls.request_demo_sim_cards()
return True
return False
@classmethod
def request_demo_sim_cards(cls):
cls.logger.info("Request %s", cls.sim_instance_name)
if cls.cn_instance_guid is None:
cls.logger.error("Core network instance GUID not set. Cannot request demo SIM cards.")
return
sim_card_parameters = {
"sim_algo": "xor",
"imsi": "505010123456789",
"k": "00112233445566778899aabbccddeeff",
"imeisv": "8682430000000101",
"impi": "505010123456789@ims.mnc505.mcc001.3gppnetwork.org",
"impu": ["505010123456789", "tel:0600000000", "tel:600"]
}
json_sim_card_parameters = json.dumps(sim_card_parameters)
cls.retry_request(cls.request, cls.product, cls.sim_instance_name,
partition_parameter_kw={'_': json_sim_card_parameters},
software_type='core-network',
filter_kw={"instance_guid": cls.cn_instance_guid},
shared=True, state='started')
# Request UE/Cell/SIM
@classmethod
def request_ue(cls):
cls.logger.info("Request %s", cls.ue_instance_name)
cls.retry_request(cls.request_ue_with_guid)
@classmethod
def request_ue_with_guid(cls):
ue_instance = cls.request(cls.product, cls.ue_instance_name,
filter_kw={"computer_guid": cls.comp_ue},
software_type='ue', state='started')
if ue_instance:
instance_infos = cls.getInstanceInfos(cls.ue_instance_name)
cls.ue_instance_guid = instance_infos.news['instance'][0]['reference']
cls.ue_com_addr = instance_infos.connection_dict.get('com_addr')
cls.rue_addr = instance_infos.connection_dict.get('rue_bind_addr')
cls.request_ue_cell1()
cls.request_ue_sim()
return True
return False
@classmethod
def request_ue_cell1(cls):
cls.logger.info("Request %s", cls.ucell1_instance_name)
ucell1_parameters = {
'cell_type': 'lte',
'cell_kind': 'ue',
'rf_mode': 'fdd',
'dl_earfcn': cls.dl_earfcn,
'bandwidth': 20,
'ru': {
'ru_type': 'sdr',
'ru_link_type': 'sdr',
'sdr_dev_list': [0],
'n_antenna_dl': 1,
'n_antenna_ul': 1,
'tx_gain': 60,
'rx_gain': 40,
'txrx_active': 'ACTIVE',
}
}
json_ucell1_parameters = json.dumps(ucell1_parameters)
cls.retry_request(cls.request, cls.product, cls.ucell1_instance_name,
partition_parameter_kw={'_': json_ucell1_parameters},
software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started')
@classmethod
def request_ue_sim(cls):
cls.logger.info("Request %s", cls.usim_instance_name)
cls.wait_for_attribute('rue_addr')
usim_parameters = {
"ue_type": "lte",
"rue_addr": cls.rue_addr,
"sim_algo": "xor",
"imsi": "505010123456789",
"k": "00112233445566778899aabbccddeeff",
"imeisv": "8682430000000101",
"impi": "505010123456789@ims.mnc505.mcc001.3gppnetwork.org",
"impu": ["505010123456789", "tel:0600000000", "tel:600"]
}
json_usim_parameters = json.dumps(usim_parameters)
cls.request(cls.product, cls.usim_instance_name,
partition_parameter_kw={'_': json_usim_parameters},
software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started')
@classmethod
def setup_websocket_connection(cls):
# cls.waitUntilGreen(cls.ue_instance_name)
cls.wait_for_attribute('ue_com_addr')
cls.ws_url = f"ws://{cls.ue_com_addr}"
cls.logger.info(f"Websocket URL: {cls.ws_url}")
for attempt in range(cls.max_retries):
try:
cls.ws = create_connection(cls.ws_url)
cls.logger.info("Websocket connection established.")
break
except Exception as e:
cls.logger.error(f"Websocket connection attempt {attempt + 1} failed: {e}")
if attempt < cls.max_retries - 1:
time.sleep(5)
@classmethod
def tearDownClass(cls):
if hasattr(cls, 'ws') and cls.ws is not None:
cls.ws.close()
super().tearDownClass()
def send(self, msg):
self.ws.send(json.dumps(msg))
def recv(self):
return json.loads(self.ws.recv())
def ue_get(self):
self.send({"message": "ue_get"})
result = self.recv()
if 'message' not in result:
raise ValueError(f"Unexpected response format: {result}")
if 'ue_list' in result:
if not result['ue_list']:
raise ValueError(f"No UE found in response: {result}")
return result['ue_list'][0]
else:
return result
def power_on(self, ue_id):
self.assertFalse(self.ue_get()['power_on'], "UE already powered on")
self.send({"message": "power_on", "ue_id": ue_id})
self.recv()
def power_off(self, ue_id):
self.assertTrue(self.ue_get()['power_on'], "UE already powered off")
self.send({"message": "power_off", "ue_id": ue_id})
self.recv()
class BBUTest(WebsocketTestClass):
def test_ue_has_ip(self):
result = self.recv()
result = self.ue_get()
ue_id = result['ue_id']
try:
self.power_on(ue_id)
time.sleep(5)
result = self.ue_get()
self.assertIn('pdn_list', result, "UE didn't connect")
self.assertIn('ipv4', result['pdn_list'][0], "UE didn't get IPv4")
self.logger.info("UE connected with ip: " + result['pdn_list'][0]['ipv4'])
finally:
self.power_off(ue_id)
def test_max_rx_sample_db(self):
custom_params = {"max_rx_sample_db": -99}
BBUTest.request_enb(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-rx-saturated", expected=False)
def test_min_rxtx_delay(self):
custom_params = {"min_rxtx_delay": 99}
BBUTest.request_enb(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name="check-baseband-latency", expected=False)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7