Skip to content

slapos
slapos
Commit b7091d9f authored by Joanne Hugé's avatar Joanne Hugé
Browse files
Options

end-to-end-testing: update test_ors to work with new master 

We don't create new instances anymore because it is currently
not possible to add instance nodes with slapos python client.
We therefore use existing services but set all necessary
parameters on them before each test.

TODO:
 * start and stop the services on each test
 * add a lock to prevent multiple E2E tests running on the
   same machine
parent 124944f5
Hide whitespace changes
Inline Side-by-side
Showing with 177 additions and 157 deletions
software/end-to-end-testing/tests/test_ors.py
View file @ b7091d9f
import json
import hashlib
import hmac
import time
import slapos.testing.e2e as e2e
from websocket import create_connection
......@@ -9,28 +11,98 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
try:
super().setUpClass()
cls.enb_instance_name = time.strftime('e2e-ors84-enb-%Y-%B-%d-%H:%M:%S')
cls.cn_instance_name = time.strftime('e2e-ors84-core-network-%Y-%B-%d-%H:%M:%S')
cls.sim_instance_name = time.strftime('e2e-ors84-sim-%Y-%B-%d-%H:%M:%S')
cls.ue_instance_name = time.strftime('e2e-simbox005-ue-%Y-%B-%d-%H:%M:%S')
cls.product = cls.product.get('ors-tdd')
cls.ue_product = "/opt/e2e/slapos/software/simpleran/software-fdd-lopcomm.cfg"
cls.logger.info("Setting up class")
# Component GUIDs and configurations
cls.comp_enb = "COMP-4057"
cls.comp_cn = "COMP-4057"
cls.comp_ue = "COMP-3756"
cls.dl_earfcn = 38550
cls.enb_instance_name = 'e2e-ors70-enb-2'
cls.core_network_instance_name = 'e2e-ors70-mme-1737037360'
cls.core_network_sim_instance_name = 'e2e-ors70-sim-card-1737037360'
cls.ue_instance_name = 'e2e-sb005-ue'
cls.ue_cell_instance_name = 'e2e-sb005-ue-cell'
cls.ue_ue_instance_name = 'e2e-sb005-ue-ue'
# Retry configurations
cls.max_retries = 10
cls.retry_delay = 180 # seconds
# Setup instances
cls.setup_instances()
cls.waitUntilGreen(cls.enb_instance_name)
cls.waitUntilGreen(cls.cn_instance_name)
cls.max_retries = 1
cls.retry_delay = 1 # seconds
mnc = '02'
mcc = '001'
plmn = mcc + mnc
mnc = (3 - len(mnc)) * '0' + mnc
cls.parameters = {}
cls.parameters['enb'] = {
'bandwidth': '10 MHz',
'dl_earfcn': 38350,
'plmn_list': {
plmn: {
'plmn': plmn
}
},
'enb_drb_stats_enabled': False,
'xlog_forwarding_enabled': False,
'amarisoft_version': '2024-03-15.1727098076'
}
cls.parameters['core-network#sim'] = {
'sim_algo': 'milenage',
'imsi': f'{plmn}0000000001',
'opc': '000102030405060708090A0B0C0D0E0F',
'amf': '0x9001',
'sqn': '000000000000',
'k': '00112233445566778899AABBCCDDEEFF',
'impu': f'{plmn}0000000001',
'impi': f'{plmn}0000000001@ims.mnc{mnc}.mcc{mcc}.3gppnetwork.org'
}
cls.parameters['core-network'] = {
'core_network_plmn': plmn,
'iperf3': True,
'network_name': 'E2E Testing',
'network_short_name': 'E2E Testing',
'amarisoft_version': '2024-03-15.1727098076'
}
cls.parameters['ue'] = {
'amarisoft_version': '2022-12-16.1733497882'
}
cls.parameters['ue#cell'] = {
'cell_type': 'lte',
'cell_kind': 'ue',
'rf_mode': 'tdd',
'ru': {
'ru_type': 'sdr',
'ru_link_type': 'sdr',
'sdr_dev_list': [
0
],
'n_antenna_dl': 2,
'n_antenna_ul': 2,
'tx_gain': 90,
'rx_gain': 60,
'txrx_active': 'ACTIVE'
},
'dl_earfcn': 38350,
'ul_earfcn': 38350,
'bandwidth': 10
}
cls.parameters['ue#ue'] = {
'ue_type': 'lte',
'imsi': f'{plmn}0000000001',
'k': '00112233445566778899AABBCCDDEEFF',
'sim_algo': 'milenage',
'opc': '000102030405060708090A0B0C0D0E0F',
'amf': '0x9001',
'sqn': '000000000000',
'impu': f'{plmn}0000000001',
'impi': f'{plmn}0000000001@ims.mnc{mnc}.mcc{mcc}.3gppnetwork.org'
}
for ref in cls.parameters:
cls.update_service(ref, 'started', cls.parameters[ref])
cls.logger.info("Waiting 5 minutes")
time.sleep(5 * 60)
cls.logger.info("Waiting until instances are green")
cls.waitUntilGreen(cls.enb_instance_name, timeout=60 * 3)
cls.waitUntilGreen(cls.ue_instance_name)
cls.setup_websocket_connection()
except Exception as e:
cls.logger.error("Error during setup: " + str(e))
......@@ -39,184 +111,132 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
raise
@classmethod
def retry_request(cls, func, *args, **kwargs):
for attempt in range(cls.max_retries):
try:
result = func(*args, **kwargs)
if result:
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
@classmethod
def setup_instances(cls):
cls.request_enb()
cls.request_core_network()
cls.setup_websocket_connection()
@classmethod
def request_enb(cls, custom_params=None):
cls.logger.info("Request "+ cls.enb_instance_name)
enb_parameters = {
"dl_earfcn": cls.dl_earfcn,
"plmn_list": {"Australia": {"plmn": "50501"}}
}
if custom_params:
enb_parameters.update(custom_params)
def setup_websocket_connection(cls):
connection_params = cls.getInstanceInfos(cls.ue_instance_name).connection_dict
cls.waitUntilGreen(cls.ue_instance_name)
cls.ws_host = connection_params.get('websocket-hostname')
cls.ws_port = connection_params.get('websocket-port')
cls.ws_pass = connection_params.get('websocket-password')
cls.ws_url = f'wss://{cls.ws_host}/websocket:{cls.ws_port}'
json_enb_parameters = json.dumps(enb_parameters)
cls.logger.info(f"Websocket URL: {cls.ws_url}")
cls.retry_request(cls.request, cls.product, cls.enb_instance_name,
filter_kw={"computer_guid": cls.comp_enb},
partition_parameter_kw={'_': json_enb_parameters},
software_type='enb')
cls.ws = create_connection(cls.ws_url)
cls.logger.info("Websocket connection established.")
data = json.loads(cls.ws.recv())
res = hmac.new(
'{}:{}:{}'.format(data['type'], cls.ws_pass, data['name']).encode(),
msg=data['challenge'].encode(),
digestmod=hashlib.sha256
).hexdigest()
msg = {'message': 'authenticate', 'res': res}
cls.ws.send(json.dumps(msg))
cls.ws.recv()
cls.logger.info("Websocket authentication established.")
@classmethod
def request_core_network(cls):
core_network_parameters = json.dumps({"core_network_plmn": "50501"})
cls.retry_request(cls.request_core_network_with_guid, core_network_parameters)
def update_service(cls, name, state, parameters=None):
sr_type = name.split('#')[0]
shared = '#' in name
name = name.replace('#', '_').replace('-', '_')
instance_name = getattr(cls, f'{name}_instance_name')
instance_infos = cls.getInstanceInfos(instance_name)
if parameters:
parameters = {'_': json.dumps(parameters)}
else:
parameters = {'_': json.dumps(instance_infos.parameter_dict['_'])}
@classmethod
def request_core_network_with_guid(cls, core_network_parameters):
cls.logger.info("Request "+ cls.cn_instance_name)
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.request_demo_sim_cards()
return True
return False
cls.logger.info(f"Update {instance_name}")
args = [instance_infos.software_url, instance_name,]
kwargs = {
'shared' : shared,
'partition_parameter_kw': parameters,
'software_type' : sr_type,
'state' : state,}
cls.logger.info("args = {}, kwargs = {}".format(repr(args), repr(kwargs)))
@classmethod
def request_demo_sim_cards(cls):
if cls.cn_instance_guid is None:
cls.logger.error("Core network instance GUID not set. Cannot request demo SIM cards.")
return
cls.logger.info("Request "+ cls.sim_instance_name)
sim_card_parameters = json.dumps({
"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.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')
cls.retry_request(cls.request, *args, **kwargs)
@classmethod
def setup_websocket_connection(cls):
ue_instance = cls.retry_request(cls.request_ue)
cls.waitUntilGreen(cls.ue_instance_name)
cls.ue_com_addr = ue_instance.get('com_addr') if ue_instance else None
if not cls.ue_com_addr:
cls.logger.error("Failed to obtain UE com address.")
return
cls.ws_url = f"ws://{cls.ue_com_addr}"
cls.logger.info(f"Websocket URL: {cls.ws_url}")
def retry_request(cls, func, *args, **kwargs):
for attempt in range(cls.max_retries):
try:
cls.ws = create_connection(cls.ws_url)
cls.logger.info("Websocket connection established.")
break
result = func(*args, **kwargs)
if result:
return result
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 request_ue(cls):
cls.logger.info("Request "+ cls.ue_instance_name)
ue_parameters = json.dumps({
"n_antenna_dl": 2,
"n_antenna_ul": 2,
"dl_earfcn": cls.dl_earfcn,
"sim_algo": "xor",
"imsi": "505010123456789",
"k": "00112233445566778899aabbccddeeff",
"imeisv": "8682430000000101",
"impi": "505010123456789@ims.mnc505.mcc001.3gppnetwork.org",
"impu": ["505010123456789", "tel:0600000000", "tel:600"]
})
return cls.retry_request(cls.request, cls.ue_product, cls.ue_instance_name,
filter_kw={"computer_guid": cls.comp_ue},
partition_parameter_kw={'_': ue_parameters},
software_type='ue-lte')
cls.logger.error(f"Error on attempt {attempt + 1}: {e}")
if attempt < cls.max_retries - 1:
time.sleep(cls.retry_delay)
return None
@classmethod
def tearDownClass(cls):
if hasattr(cls, 'ws') and cls.ws is not None:
cls.logger.info("Closing websocket")
cls.ws.close()
super().tearDownClass()
# TODO: uncomment these lines
#cls.update_service('enb', 'stopped')
#cls.update_service('core-network', 'stopped')
#cls.update_service('ue', 'stopped')
# Don't call super().tearDownClass as we don't want to destroy requested instances
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"})
self.send({'message': 'ue_get'})
result = self.recv()
if 'message' not in result:
raise ValueError(f"Unexpected response format: {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}")
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.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.send({'message': 'power_off', 'ue_id': ue_id})
self.recv()
class ORSTest(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}
ORSTest.request_enb(custom_params)
self.waitUntilPromises(ORSTest.enb_instance_name, promise_name="check-rx-saturated", expected=False)
def test_min_rxtx_delay(self):
# Fixed by 9798ef1e, change `expected` to False when released
custom_params = {"min_rxtx_delay": 99}
ORSTest.request_enb(custom_params)
self.waitUntilPromises(ORSTest.enb_instance_name, promise_name="check-baseband-latency", expected=True)
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)
# TODO: uncomment these tests
#def test_max_rx_sample_db(self):
# custom_params = {}
# custom_params.update(self.parameters['enb'])
# custom_params.update({"max_rx_sample_db": -99})
# self.update_service('enb', 'started', custom_params)
# self.waitUntilPromises(ORSTest.enb_instance_name, promise_name="check-rx-saturated", expected=False)
#def test_min_rxtx_delay(self):
# # Fixed by 9798ef1e, change `expected` to False when released
# custom_params = {}
# custom_params.update(self.parameters['enb'])
# custom_params.update({"min_rxtx_delay": 99})
# self.update_service('enb', 'started', custom_params)
# self.waitUntilPromises(ORSTest.enb_instance_name, promise_name="check-baseband-latency", expected=True)
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