Skip to content

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

fixup

parent 74c31e19
Hide whitespace changes
Inline Side-by-side
Showing with 190 additions and 58 deletions
software/end-to-end-testing/tests/test_master_lopcomm.py
View file @ 2bde133a
import json import json
import time import time
import socket
from datetime import datetime, timedelta
from websocket import create_connection from websocket import create_connection
import slapos.testing.e2e as e2e import slapos.testing.e2e as e2e
class WebsocketTestClass(e2e.EndToEndTestCase): class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
...@@ -29,13 +32,15 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -29,13 +32,15 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
def set_instance_names(cls): def set_instance_names(cls):
current_time = time.strftime('%Y-%B-%d-%H:%M:%S') current_time = time.strftime('%Y-%B-%d-%H:%M:%S')
ors_branch = 'master' ors_branch = 'master'
base_names = ['enb', 'eru1', 'ecell1', 'cn', 'sim', 'ue', 'ucell1', 'usim'] base_names = ['enb', 'eru1', 'ecell1',
'cn', 'sim', 'ue', 'ucell1', 'usim']
for name in base_names: for name in base_names:
setattr(cls, f"{name}_instance_name", f"E2E-{ors_branch}-{name}-{current_time}") setattr(cls, f"{name}_instance_name",
f"E2E-{ors_branch}-{name}-{current_time}")
@classmethod @classmethod
def setup_guids_and_configs(cls): def setup_guids_and_configs(cls):
cls.comp_enb ="COMP-3920" cls.comp_enb = "COMP-3920"
cls.comp_cn = "COMP-3920" cls.comp_cn = "COMP-3920"
cls.comp_ue = "COMP-3756" cls.comp_ue = "COMP-3756"
cls.dl_earfcn = 300 cls.dl_earfcn = 300
...@@ -56,7 +61,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -56,7 +61,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
result = func(*args, **kwargs) result = func(*args, **kwargs)
if result: if result:
return result return result
cls.logger.info(f"Attempt {attempt + 1}: Received empty or invalid result, retrying...") cls.logger.info(
f"Attempt {attempt + 1}: Received empty or invalid result, retrying...")
except Exception as e: except Exception as e:
cls.logger.error("Error on attempt %d: %s", attempt + 1, e) cls.logger.error("Error on attempt %d: %s", attempt + 1, e)
if attempt < cls.max_retries - 1: if attempt < cls.max_retries - 1:
...@@ -70,7 +76,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -70,7 +76,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
start_time = time.time() start_time = time.time()
while not hasattr(cls, attr_name) or getattr(cls, attr_name) is None: while not hasattr(cls, attr_name) or getattr(cls, attr_name) is None:
if time.time() - start_time > timeout: if time.time() - start_time > timeout:
raise TimeoutError(f"Timeout waiting for attribute '{attr_name}' to become available.") raise TimeoutError(
f"Timeout waiting for attribute '{attr_name}' to become available.")
time.sleep(cls.retry_delay) time.sleep(cls.retry_delay)
cls.logger.info(f"Attribute '{attr_name}' is now available.") cls.logger.info(f"Attribute '{attr_name}' is now available.")
...@@ -80,7 +87,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -80,7 +87,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
cls.logger.info("Request %s", cls.enb_instance_name) cls.logger.info("Request %s", cls.enb_instance_name)
enb_parameters = { enb_parameters = {
"enb_id": "0x1A2D0", "enb_id": "0x1A2D0",
"mme_list": {'1': {'mme_addr': cls.cn_ipv6 }}, "mme_list": {'1': {'mme_addr': cls.cn_ipv6}},
"plmn_list": {"Australia": {"plmn": "50501"}} "plmn_list": {"Australia": {"plmn": "50501"}}
} }
...@@ -94,9 +101,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -94,9 +101,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod @classmethod
def request_enb_with_guid(cls, json_enb_parameters): def request_enb_with_guid(cls, json_enb_parameters):
enb_instance = cls.request(cls.product, cls.enb_instance_name, enb_instance = cls.request(cls.product, cls.enb_instance_name,
filter_kw={"computer_guid": cls.comp_enb}, filter_kw={"computer_guid": cls.comp_enb},
partition_parameter_kw={'_': json_enb_parameters}, partition_parameter_kw={
software_type='enb') '_': json_enb_parameters},
software_type='enb')
if enb_instance: if enb_instance:
instance_infos = cls.getInstanceInfos(cls.enb_instance_name) instance_infos = cls.getInstanceInfos(cls.enb_instance_name)
cls.enb_instance_guid = instance_infos.news['instance'][0]['reference'] cls.enb_instance_guid = instance_infos.news['instance'][0]['reference']
...@@ -105,6 +113,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -105,6 +113,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
return True return True
return False return False
@classmethod
def request_ru1(cls, custom_params=None): def request_ru1(cls, custom_params=None):
cls.logger.info("Request %s", cls.eru1_instance_name) cls.logger.info("Request %s", cls.eru1_instance_name)
ru1_parameters = { ru1_parameters = {
...@@ -122,8 +131,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -122,8 +131,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
'mac_addr': '00:0a:00:00:10:20', 'mac_addr': '00:0a:00:00:10:20',
'n_antenna_dl': 1, 'n_antenna_dl': 1,
'n_antenna_ul': 1, 'n_antenna_ul': 1,
'tx_gain': -20, 'tx_gain': -20,
'rx_gain': -10, 'rx_gain': -10,
'txrx_active': 'ACTIVE', 'txrx_active': 'ACTIVE',
} }
if custom_params: if custom_params:
...@@ -131,13 +140,22 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -131,13 +140,22 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_ru1_parameters = json.dumps(ru1_parameters) json_ru1_parameters = json.dumps(ru1_parameters)
for i in range(5): ru1_instance = cls.request(cls.product, cls.eru1_instance_name,
i += 1 filter_kw={
cls.request(cls.product, cls.eru1_instance_name, "instance_guid": cls.enb_instance_guid},
filter_kw={"instance_guid": cls.enb_instance_guid}, partition_parameter_kw={
partition_parameter_kw={'_': json_ru1_parameters}, '_': json_ru1_parameters},
shared=True, software_type='enb') shared=True, software_type='enb', state='started')
for _ in range(5):
cls.logger.info("Request %s", cls.eru1_instance_name)
ru1_instance
if ru1_instance:
instance_infos = cls.getInstanceInfos(cls.eru1_instance_name)
cls.ru1_ipv6 = instance_infos.connection_dict.get('ipv6')
return True
return False
@classmethod @classmethod
def request_cell1(cls): def request_cell1(cls):
...@@ -159,15 +177,14 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -159,15 +177,14 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_cell1_parameters = json.dumps(cell1_parameters) json_cell1_parameters = json.dumps(cell1_parameters)
for i in range(5): for _ in range(5):
i += 1
cls.request(cls.product, cls.ecell1_instance_name, cls.request(cls.product, cls.ecell1_instance_name,
partition_parameter_kw={'_': json_cell1_parameters}, partition_parameter_kw={'_': json_cell1_parameters},
software_type='enb', software_type='enb',
filter_kw={"instance_guid": cls.enb_instance_guid}, filter_kw={"instance_guid": cls.enb_instance_guid},
shared=True, state='started') shared=True, state='started')
#Request Core Network/SIM Card # Request Core Network/SIM Card
@classmethod @classmethod
def request_core_network(cls): def request_core_network(cls):
cls.logger.info("Request %s", cls.cn_instance_name) cls.logger.info("Request %s", cls.cn_instance_name)
...@@ -175,18 +192,22 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -175,18 +192,22 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
"core_network_plmn": "50501", "core_network_plmn": "50501",
'external_enb_gnb': True, 'external_enb_gnb': True,
}) })
cls.retry_request(cls.request_core_network_with_guid, core_network_parameters) cls.retry_request(cls.request_core_network_with_guid,
core_network_parameters)
@classmethod @classmethod
def request_core_network_with_guid(cls, core_network_parameters): def request_core_network_with_guid(cls, core_network_parameters):
core_network_instance = cls.request(cls.product, cls.cn_instance_name, core_network_instance = cls.request(cls.product, cls.cn_instance_name,
filter_kw={"computer_guid": cls.comp_cn}, filter_kw={
partition_parameter_kw={'_': core_network_parameters}, "computer_guid": cls.comp_cn},
partition_parameter_kw={
'_': core_network_parameters},
software_type='core-network') software_type='core-network')
if core_network_instance: if core_network_instance:
instance_infos = cls.getInstanceInfos(cls.cn_instance_name) instance_infos = cls.getInstanceInfos(cls.cn_instance_name)
cls.cn_instance_guid = instance_infos.news['instance'][0]['reference'] cls.cn_instance_guid = instance_infos.news['instance'][0]['reference']
cls.cn_ipv6 = instance_infos.connection_dict.get('core-network-ipv6') cls.cn_ipv6 = instance_infos.connection_dict.get(
'core-network-ipv6')
cls.request_demo_sim_cards() cls.request_demo_sim_cards()
return True return True
return False return False
...@@ -195,7 +216,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -195,7 +216,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
def request_demo_sim_cards(cls): def request_demo_sim_cards(cls):
cls.logger.info("Request %s", cls.sim_instance_name) cls.logger.info("Request %s", cls.sim_instance_name)
if cls.cn_instance_guid is None: if cls.cn_instance_guid is None:
cls.logger.error("Core network instance GUID not set. Cannot request demo SIM cards.") cls.logger.error(
"Core network instance GUID not set. Cannot request demo SIM cards.")
return return
sim_card_parameters = { sim_card_parameters = {
...@@ -209,7 +231,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -209,7 +231,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_sim_card_parameters = json.dumps(sim_card_parameters) json_sim_card_parameters = json.dumps(sim_card_parameters)
cls.retry_request(cls.request, cls.product, cls.sim_instance_name, cls.retry_request(cls.request, cls.product, cls.sim_instance_name,
partition_parameter_kw={'_': json_sim_card_parameters}, partition_parameter_kw={
'_': json_sim_card_parameters},
software_type='core-network', software_type='core-network',
filter_kw={"instance_guid": cls.cn_instance_guid}, filter_kw={"instance_guid": cls.cn_instance_guid},
shared=True, state='started') shared=True, state='started')
...@@ -223,8 +246,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -223,8 +246,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@classmethod @classmethod
def request_ue_with_guid(cls): def request_ue_with_guid(cls):
ue_instance = cls.request(cls.product, cls.ue_instance_name, ue_instance = cls.request(cls.product, cls.ue_instance_name,
filter_kw={"computer_guid": cls.comp_ue}, filter_kw={"computer_guid": cls.comp_ue},
software_type='ue', state='started') software_type='ue', state='started')
if ue_instance: if ue_instance:
instance_infos = cls.getInstanceInfos(cls.ue_instance_name) instance_infos = cls.getInstanceInfos(cls.ue_instance_name)
cls.ue_instance_guid = instance_infos.news['instance'][0]['reference'] cls.ue_instance_guid = instance_infos.news['instance'][0]['reference']
...@@ -259,10 +282,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -259,10 +282,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_ucell1_parameters = json.dumps(ucell1_parameters) json_ucell1_parameters = json.dumps(ucell1_parameters)
cls.retry_request(cls.request, cls.product, cls.ucell1_instance_name, cls.retry_request(cls.request, cls.product, cls.ucell1_instance_name,
partition_parameter_kw={'_': json_ucell1_parameters}, partition_parameter_kw={'_': json_ucell1_parameters},
software_type='ue', software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid}, filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started') shared=True, state='started')
@classmethod @classmethod
def request_ue_sim(cls): def request_ue_sim(cls):
...@@ -281,10 +304,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -281,10 +304,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_usim_parameters = json.dumps(usim_parameters) json_usim_parameters = json.dumps(usim_parameters)
cls.request(cls.product, cls.usim_instance_name, cls.request(cls.product, cls.usim_instance_name,
partition_parameter_kw={'_': json_usim_parameters}, partition_parameter_kw={'_': json_usim_parameters},
software_type='ue', software_type='ue',
filter_kw={"instance_guid": cls.ue_instance_guid}, filter_kw={"instance_guid": cls.ue_instance_guid},
shared=True, state='started') shared=True, state='started')
@classmethod @classmethod
def setup_websocket_connection(cls): def setup_websocket_connection(cls):
...@@ -299,10 +322,24 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -299,10 +322,24 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
cls.logger.info("Websocket connection established.") cls.logger.info("Websocket connection established.")
break break
except Exception as e: except Exception as e:
cls.logger.error(f"Websocket connection attempt {attempt + 1} failed: {e}") cls.logger.error(
f"Websocket connection attempt {attempt + 1} failed: {e}")
if attempt < cls.max_retries - 1: if attempt < cls.max_retries - 1:
time.sleep(5) time.sleep(5)
@classmethod
def send_udp_packet(cls, dst_address, data, dst_port=13200):
sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
try:
sock.sendto(data, (dst_address, dst_port))
cls.logger.info(
f"UDP packet sent successfully to {dst_address}:{dst_port}")
except Exception as e:
cls.logger.error(f"Failed to send UDP packet: {e}")
finally:
sock.close()
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
if hasattr(cls, 'ws') and cls.ws is not None: if hasattr(cls, 'ws') and cls.ws is not None:
...@@ -339,8 +376,17 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -339,8 +376,17 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
self.send({"message": "power_off", "ue_id": ue_id}) self.send({"message": "power_off", "ue_id": ue_id})
self.recv() self.recv()
class BBUTest(WebsocketTestClass): class BBUTest(WebsocketTestClass):
def test_ue_has_ip(self): def test_ue_has_ip(self):
"""
Tests that a User Equipment (UE) can establish a connection and receive an IP address.
This method sets up a websocket connection, requests the status of the UE, and then
activates transmission and reception (txrx) on the Radio Unit (RU). After activating the UE and allowing
some time for network registration, it checks if the UE has successfully connected to the network and
received an IPv4 address from the PDN list. If successful, the IPv4 address is logged.
"""
BBUTest.setup_websocket_connection() BBUTest.setup_websocket_connection()
result = self.recv() result = self.recv()
result = self.ue_get() result = self.ue_get()
...@@ -357,11 +403,19 @@ class BBUTest(WebsocketTestClass): ...@@ -357,11 +403,19 @@ class BBUTest(WebsocketTestClass):
self.logger.info(result) self.logger.info(result)
self.assertIn('pdn_list', result, "UE didn't connect") self.assertIn('pdn_list', result, "UE didn't connect")
self.assertIn('ipv4', result['pdn_list'][0], "UE didn't get IPv4") self.assertIn('ipv4', result['pdn_list'][0], "UE didn't get IPv4")
self.logger.info("UE connected with ip: " + result['pdn_list'][0]['ipv4']) self.logger.info("UE connected with ip: " +
result['pdn_list'][0]['ipv4'])
finally: finally:
self.power_off(ue_id) self.power_off(ue_id)
def test_txrx_inactive(self): def test_txrx_inactive(self):
"""
Verifies that a User Equipment (UE) does not connect when Radio Unit (RU) carriers are inactive.
This test sets up a websocket connection and configures the RU to have inactive transmission and
reception (txrx). After attempting to power on the UE, the method checks that the UE does not establish a
connection, evidenced by the absence of a PDN list in the UE's status.
"""
BBUTest.setup_websocket_connection() BBUTest.setup_websocket_connection()
result = self.recv() result = self.recv()
result = self.ue_get() result = self.ue_get()
...@@ -381,20 +435,111 @@ class BBUTest(WebsocketTestClass): ...@@ -381,20 +435,111 @@ class BBUTest(WebsocketTestClass):
self.power_off(ue_id) self.power_off(ue_id)
def test_max_rx_sample_db(self): def test_max_rx_sample_db(self):
"""
Tests the alarm for saturated RX samples by setting the maximum RX sample dB to an exceptionally low value.
This method modifies the eNodeB configuration to test if the system correctly identifies and handles the condition
where the received signal strength (RX samples) is below a set threshold, potentially triggering a saturation alarm.
"""
custom_params = {"max_rx_sample_db": -999} custom_params = {"max_rx_sample_db": -999}
BBUTest.request_enb(custom_params) BBUTest.request_enb(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name=BBUTest.eru1_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): def test_min_rxtx_delay(self):
"""
Checks the baseband latency by setting a minimum threshold for round-trip delay in transmission and reception.
This method configures the eNodeB to test the baseband's ability to handle specified latencies, ensuring that
the system can maintain synchronization and performance under constrained delay conditions.
"""
custom_params = {"min_rxtx_delay": 99} custom_params = {"min_rxtx_delay": 99}
BBUTest.request_enb(custom_params) BBUTest.request_enb(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name="check-baseband-latency", expected=False) self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name="check-baseband-latency", expected=False)
def test_ru_reset_and_cpri_lock_lost(self): def test_ru_reset_and_cpri_lock_lost(self):
"""
Tests the CPRI link lock loss and recovery by scheduling a reset of the Radio Unit (RU).
This method schedules an RU reset one minute into the future and checks the CPRI link status immediately after
the scheduled reset time, expecting it to be lost. After a delay, it verifies that the CPRI link lock is restored,
demonstrating the system's resilience and recovery capabilities.
"""
current_time = datetime.now() current_time = datetime.now()
future_time = current_time + timedelta(minutes=1) future_time = current_time + timedelta(minutes=1)
reset_crontab_time = f"{future_time.minute} {future_time.hour} * * *" reset_crontab_time = f"{future_time.minute} {future_time.hour} * * *"
custom_params = {"reset_schedule": reset_crontab_time} custom_params = {"reset_schedule": reset_crontab_time}
BBUTest.request_ru1(custom_params) BBUTest.request_ru1(custom_params)
self.waitUntilPromises(BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-cpri-lock", expected=False) self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-cpri-lock", expected=False)
time.sleep(300)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-cpri-lock", expected=True)
def test_vswr(self):
"""
Tests the Voltage Standing Wave Ratio (VSWR) by temporarily setting the antenna threshold to zero and observing
system responses.
This method sends specific data to set the VSWR threshold to zero and checks if the system correctly identifies
and handles the potential VSWR alarm condition. After testing, it resets the VSWR configuration to its original
state and verifies that the system returns to normal operation.
"""
BBUTest.wait_for_attribute('ru1_ipv6')
dst_address = BBUTest.ru1_ipv6
test_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x0A, 0x04, 0x01, 0x00, 0x71, 0x02, 0x4E])
reset_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x0A, 0x04, 0x01, 0x19, 0x8A, 0x02, 0x4E])
BBUTest.send_udp_packet(dst_address, test_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-vswr", expected=False)
BBUTest.send_udp_packet(dst_address, reset_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-vswr", expected=True)
def test_pa_over_current(self):
"""
Tests the system's response to a power amplifier (PA) over-current condition by temporarily setting the current draw to a threshold that triggers an alarm.
This test sends data to the Radio Unit (RU) to simulate an over-current condition on the PA. It checks the system's response to this condition by waiting for the relevant promise to indicate failure (alarm triggered). After the test, it sends a reset command to restore the PA current settings to normal levels and verifies that the system returns to a stable state without alarms.
"""
BBUTest.wait_for_attribute('ru1_ipv6')
dst_address = BBUTest.ru1_ipv6
test_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x04, 0x04, 0x02, 0x00, 0x00, 0x6C, 0x02, 0x4E])
reset_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x04, 0x04, 0x02, 0x20, 0x03, 0x8F, 0x02, 0x4E])
BBUTest.send_udp_packet(dst_address, test_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-pa-current", expected=False)
BBUTest.send_udp_packet(dst_address, reset_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-pa-current", expected=True)
def test_pa_over_output_power(self):
"""
Tests the system's response to a power amplifier (PA) over-output power condition by configuring the output power to exceed normal operational limits.
This method triggers an over-output power condition by sending specific data to the Radio Unit (RU). It then observes if the system properly identifies and reacts to this condition, checking for a corresponding alarm trigger. Following the test, a reset command is issued to bring the output power back within safe operational parameters, ensuring the system's stability is restored.
"""
BBUTest.wait_for_attribute('ru1_ipv6')
dst_address = BBUTest.ru1_ipv6
test_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x06, 0x04, 0x02, 0x00, 0x00, 0x6E, 0x02, 0x4E])
reset_data = bytes(
[0x4E, 0x01, 0x02, 0x01, 0xFF, 0x60, 0xFF, 0x06, 0x04, 0x02, 0x2C, 0x01, 0x9B, 0x02, 0x4E])
BBUTest.send_udp_packet(dst_address, test_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-pa-output-power", expected=False)
BBUTest.send_udp_packet(dst_address, reset_data)
self.waitUntilPromises(
BBUTest.enb_instance_name, promise_name=BBUTest.eru1_instance_name + "-pa-output-power", expected=True)
software/end-to-end-testing/tests/test_ors_fix_lopcomm.py
View file @ 2bde133a
import json import json
import time import time
import socket
from datetime import datetime, timedelta
from websocket import create_connection from websocket import create_connection
import slapos.testing.e2e as e2e import slapos.testing.e2e as e2e
...@@ -92,19 +94,6 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -92,19 +94,6 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
cls.retry_request(cls.request_enb_with_guid, json_enb_parameters) cls.retry_request(cls.request_enb_with_guid, json_enb_parameters)
@classmethod @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
def request_ru1(cls, custom_params=None): def request_ru1(cls, custom_params=None):
cls.logger.info("Request %s", cls.eru1_instance_name) cls.logger.info("Request %s", cls.eru1_instance_name)
ru1_parameters = { ru1_parameters = {
...@@ -131,8 +120,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -131,8 +120,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_ru1_parameters = json.dumps(ru1_parameters) json_ru1_parameters = json.dumps(ru1_parameters)
for i in range(5): for _ in range(5):
i += 1
cls.request(cls.product, cls.eru1_instance_name, cls.request(cls.product, cls.eru1_instance_name,
filter_kw={"instance_guid": cls.enb_instance_guid}, filter_kw={"instance_guid": cls.enb_instance_guid},
partition_parameter_kw={'_': json_ru1_parameters}, partition_parameter_kw={'_': json_ru1_parameters},
...@@ -158,8 +146,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase): ...@@ -158,8 +146,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
json_cell1_parameters = json.dumps(cell1_parameters) json_cell1_parameters = json.dumps(cell1_parameters)
for i in range(5): for _ in range(5):
i += 1
cls.request(cls.product, cls.ecell1_instance_name, cls.request(cls.product, cls.ecell1_instance_name,
partition_parameter_kw={'_': json_cell1_parameters}, partition_parameter_kw={'_': json_cell1_parameters},
software_type='enb', software_type='enb',
......
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