maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove COTS_UE

Browse Source
This commit is contained in:
Robert Schmidt
2023-02-23 17:38:10 +01:00
parent e1e6413977
commit e7b40cb444
3 changed files with 40 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ci-scripts/cls_cots_ue.py
-138
View File
@@ -1,138 +0,0 @@
# * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
# * contributor license agreements. See the NOTICE file distributed with
# * this work for additional information regarding copyright ownership.
# * The OpenAirInterface Software Alliance licenses this file to You under
# * the OAI Public License, Version 1.1 (the "License"); you may not use this file
# * except in compliance with the License.
# * You may obtain a copy of the License at
# *
# * http://www.openairinterface.org/?page_id=698
# *
# * Unless required by applicable law or agreed to in writing, software
# * distributed under the License is distributed on an "AS IS" BASIS,
# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# * See the License for the specific language governing permissions and
# * limitations under the License.
# *-------------------------------------------------------------------------------
# * For more information about the OpenAirInterface (OAI) Software Alliance:
# * contact@openairinterface.org
# */
#---------------------------------------------------------------------
# Python for CI of OAI-eNB + COTS-UE
#
# Required Python Version
# Python 3.x
#
# Required Python Package
# pexpect
#---------------------------------------------------------------------
#to use isfile
import os
import sys
#to use logging.info()
import logging
#to create a SSH object locally in the methods
import sshconnection
#time.sleep
import time
#to load cots_ue dictionary
import yaml
class CotsUe:
def __init__(self,ADBIPAddr,ADBUserName,ADBPassWord):
self.cots_id = '' #cots id from yaml oppo, s10 etc...
self.ADBIPAddr = ADBIPAddr
self.ADBUserName = ADBUserName
self.ADBPassWord = ADBPassWord
self.cots_run_mode = '' #on of off to toggle airplane mode on/off
self.__cots_cde_dict_file = 'cots_ue_ctl.yaml'
self.__SetAirplaneRetry = 3
#-----------------$
#PUBLIC Methods$
#-----------------$
def Check_Airplane(self):
mySSH = sshconnection.SSHConnection()
mySSH.open(self.ADBIPAddr, self.ADBUserName, self.ADBPassWord)
status=mySSH.cde_check_value('adb shell settings get global airplane_mode_on ', ['0','1'],5)
mySSH.close()
return status
#simply check if the device id exists in the dictionary
#returns true if it exists, false otherwise
def Check_Exists(self,target_id):
#load cots commands dictionary
if (os.path.isfile(self.__cots_cde_dict_file)):
yaml_file=self.__cots_cde_dict_file
elif (os.path.isfile('ci-scripts/'+self.__cots_cde_dict_file)):
yaml_file='ci-scripts/'+self.__cots_cde_dict_file
else:
logging.error("COTS UE dictionary yaml file cannot be found")
sys.exit("COTS UE dictionary yaml file cannot be found")
#load cots commands dictionary
with open(yaml_file,'r') as file:
cots_ue_ctl = yaml.load(file,Loader=yaml.FullLoader)
#check if ue id is in the dictionary
if target_id in cots_ue_ctl:
return True
else:
return False
def Set_Airplane(self, target_id, target_state_str):
#loading cots commands dictionary
if (os.path.isfile(self.__cots_cde_dict_file)):
yaml_file=self.__cots_cde_dict_file
elif (os.path.isfile('ci-scripts/'+self.__cots_cde_dict_file)):
yaml_file='ci-scripts/'+self.__cots_cde_dict_file
else:
logging.error("COTS UE dictionary yaml file cannot be found")
sys.exit("COTS UE dictionary yaml file cannot be found")
#load cots commands dictionary
with open(yaml_file,'r') as file:
cots_ue_ctl = yaml.load(file,Loader=yaml.FullLoader)
#check if ue id is in the dictionary
if target_id in cots_ue_ctl:
mySSH = sshconnection.SSHConnection()
mySSH.open(self.ADBIPAddr, self.ADBUserName, self.ADBPassWord)
logging.info(str(self.ADBIPAddr)+' '+str(self.ADBUserName)+' '+str(self.ADBPassWord))
mySSH.command('adb start-server','\$',5)
mySSH.command('adb devices','\$',5)
logging.info("Toggling COTS UE Airplane mode to : "+target_state_str)
#get current state
current_state = self.Check_Airplane()
if target_state_str.lower()=="on":
target_state=1
else:
target_state=0
if current_state != target_state:
#toggle state
retry = 0
while (current_state!=target_state) and (retry < self.__SetAirplaneRetry):
#loop over the command list from dictionary for the selected ue, to switch to required state
for i in range (0,len(cots_ue_ctl[target_id])):
mySSH.command(cots_ue_ctl[target_id][i], '\$', 5)
time.sleep(1)
current_state = self.Check_Airplane()
retry+=1
#could not toggle despite the retry
if current_state != target_state:
logging.error("ATTENTION : Could not toggle to : "+target_state_str)
logging.error("Current state is : "+ str(current_state))
else:
logging.info("Airplane mode is already "+ target_state_str)
mySSH.command('adb kill-server','\$',5)
mySSH.close()
#ue id is NOT in the dictionary
else:
logging.error("COTS UE Id from XML could not be found in UE YAML dictionary " + self.__cots_cde_dict_file)
sys.exit("COTS UE Id from XML could not be found in UE YAML dictionary " + self.__cots_cde_dict_file)
ci-scripts/cls_oaicitest.py
+31 -30
View File
@@ -292,11 +292,11 @@ class OaiCiTest():
self.ConditionalExit() self.ConditionalExit()
def InitializeUE(self,HTML,RAN,EPC, COTS_UE, CONTAINERS): def InitializeUE(self,HTML,RAN,EPC, CONTAINERS):
ue = cls_module_ue.Module_UE(self.ue_id) ue = cls_module_ue.Module_UE(self.ue_id)
ue.initialize() ue.initialize()
def InitializeOAIUE(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def InitializeOAIUE(self,HTML,RAN,EPC,CONTAINERS):
if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '': if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '':
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
sys.exit('Insufficient Parameter') sys.exit('Insufficient Parameter')
@@ -504,9 +504,9 @@ class OaiCiTest():
HTML.htmlUEFailureMsg='nr-uesoftmodem did NOT synced' HTML.htmlUEFailureMsg='nr-uesoftmodem did NOT synced'
HTML.CreateHtmlTestRow(self.air_interface + ' ' + self.Initialize_OAI_UE_args, 'KO', CONST.OAI_UE_PROCESS_COULD_NOT_SYNC, 'OAI UE') HTML.CreateHtmlTestRow(self.air_interface + ' ' + self.Initialize_OAI_UE_args, 'KO', CONST.OAI_UE_PROCESS_COULD_NOT_SYNC, 'OAI UE')
logging.error('\033[91mInitialize OAI UE Failed! \033[0m') logging.error('\033[91mInitialize OAI UE Failed! \033[0m')
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
def AttachUE(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def AttachUE(self,HTML,RAN,EPC,CONTAINERS):
ue = cls_module_ue.Module_UE(self.ue_id) ue = cls_module_ue.Module_UE(self.ue_id)
connected = ue.attach() connected = ue.attach()
if connected: if connected:
@@ -516,7 +516,7 @@ class OaiCiTest():
HTML.CreateHtmlTestRow('N/A', 'KO', CONST.UE_IP_ADDRESS_ISSUE) HTML.CreateHtmlTestRow('N/A', 'KO', CONST.UE_IP_ADDRESS_ISSUE)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS)
def DetachUE(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def DetachUE(self,HTML,RAN,EPC,CONTAINERS):
ue = cls_module_ue.Module_UE(self.ue_id) ue = cls_module_ue.Module_UE(self.ue_id)
ue.detach() ue.detach()
HTML.CreateHtmlTestRow('NA', 'OK', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow('NA', 'OK', CONST.ALL_PROCESSES_OK)
@@ -695,7 +695,7 @@ class OaiCiTest():
except: except:
os.kill(os.getppid(),signal.SIGUSR1) os.kill(os.getppid(),signal.SIGUSR1)
def CheckStatusUE(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def CheckStatusUE(self,HTML,RAN,EPC,CONTAINERS):
raise Exception("not implemented") raise Exception("not implemented")
check_eNB = True check_eNB = True
check_OAI_UE = False check_OAI_UE = False
@@ -721,7 +721,7 @@ class OaiCiTest():
if (status_queue.empty()): if (status_queue.empty()):
HTML.CreateHtmlTestRow(htmlOptions, 'KO', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow(htmlOptions, 'KO', CONST.ALL_PROCESSES_OK)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
else: else:
check_status = True check_status = True
messages = [] messages = []
@@ -735,7 +735,7 @@ class OaiCiTest():
HTML.CreateHtmlTestRowQueue(htmlOptions, 'OK', messages) HTML.CreateHtmlTestRowQueue(htmlOptions, 'OK', messages)
else: else:
HTML.CreateHtmlTestRowQueue(htmlOptions, 'KO', messages) HTML.CreateHtmlTestRowQueue(htmlOptions, 'KO', messages)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
def ping_iperf_wrong_exit(self, lock, UE_IPAddress, device_id, statusQueue, message): def ping_iperf_wrong_exit(self, lock, UE_IPAddress, device_id, statusQueue, message):
lock.acquire() lock.acquire()
@@ -892,14 +892,14 @@ class OaiCiTest():
message = 'OAI UE ping result\n{qMsg}' message = 'OAI UE ping result\n{qMsg}'
HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', [message]) HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', [message])
def PingNoS1(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def PingNoS1(self,HTML,RAN,EPC,CONTAINERS):
SSH=sshconnection.SSHConnection() SSH=sshconnection.SSHConnection()
check_eNB = True check_eNB = True
check_OAI_UE = True check_OAI_UE = True
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
if (pStatus < 0): if (pStatus < 0):
HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
return return
ping_from_eNB = re.search('oaitun_enb1', str(self.ping_args)) ping_from_eNB = re.search('oaitun_enb1', str(self.ping_args))
if ping_from_eNB is not None: if ping_from_eNB is not None:
@@ -982,10 +982,10 @@ class OaiCiTest():
except: except:
os.kill(os.getppid(),signal.SIGUSR1) os.kill(os.getppid(),signal.SIGUSR1)
def Ping(self,HTML,RAN,EPC,COTS_UE, CONTAINERS): def Ping(self,HTML,RAN,EPC,CONTAINERS):
result = re.search('noS1', str(RAN.Initialize_eNB_args)) result = re.search('noS1', str(RAN.Initialize_eNB_args))
if result is not None: if result is not None:
self.PingNoS1(HTML,RAN,EPC,COTS_UE,CONTAINERS) self.PingNoS1(HTML,RAN,EPC,CONTAINERS)
return return
if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '':
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
@@ -998,7 +998,7 @@ class OaiCiTest():
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
if (pStatus < 0): if (pStatus < 0):
HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
return return
if self.ue_id == "": if self.ue_id == "":
@@ -1009,7 +1009,7 @@ class OaiCiTest():
if not ue.getIP(): if not ue.getIP():
logging.error("no IP addresses returned") logging.error("no IP addresses returned")
HTML.CreateHtmlTestRow(self.ping_args, 'KO', len(self.UEDevices), html_queue) HTML.CreateHtmlTestRow(self.ping_args, 'KO', len(self.UEDevices), html_queue)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
ues.append(ue) ues.append(ue)
logging.debug(ues) logging.debug(ues)
@@ -1028,7 +1028,7 @@ class OaiCiTest():
if (status_queue.empty()): if (status_queue.empty()):
HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.ALL_PROCESSES_OK)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
else: else:
ping_status = True ping_status = True
messages = [] messages = []
@@ -1044,7 +1044,7 @@ class OaiCiTest():
HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', messages) HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', messages)
else: else:
HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', messages) HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', messages)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
def Iperf_ComputeTime(self): def Iperf_ComputeTime(self):
result = re.search('-t (?P<iperf_time>\d+)', str(self.iperf_args)) result = re.search('-t (?P<iperf_time>\d+)', str(self.iperf_args))
@@ -1903,7 +1903,7 @@ class OaiCiTest():
except: except:
os.kill(os.getppid(),signal.SIGUSR1) os.kill(os.getppid(),signal.SIGUSR1)
def IperfNoS1(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def IperfNoS1(self,HTML,RAN,EPC,CONTAINERS):
SSH = sshconnection.SSHConnection() SSH = sshconnection.SSHConnection()
if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '' or self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '': if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '' or self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '':
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
@@ -1913,7 +1913,7 @@ class OaiCiTest():
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
if (pStatus < 0): if (pStatus < 0):
HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus) HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
return return
server_on_enb = re.search('-R', str(self.iperf_args)) server_on_enb = re.search('-R', str(self.iperf_args))
if server_on_enb is not None: if server_on_enb is not None:
@@ -2011,12 +2011,12 @@ class OaiCiTest():
HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', messages) HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', messages)
else: else:
HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', messages) HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', messages)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
def Iperf(self,HTML,RAN,EPC,COTS_UE,CONTAINERS): def Iperf(self,HTML,RAN,EPC,CONTAINERS):
result = re.search('noS1', str(RAN.Initialize_eNB_args)) result = re.search('noS1', str(RAN.Initialize_eNB_args))
if result is not None: if result is not None:
self.IperfNoS1(HTML,RAN,EPC,COTS_UE,CONTAINERS) self.IperfNoS1(HTML,RAN,EPC,CONTAINERS)
return return
if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '':
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
@@ -2029,7 +2029,7 @@ class OaiCiTest():
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
if (pStatus < 0): if (pStatus < 0):
HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus) HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
return return
if self.ue_id == "": if self.ue_id == "":
@@ -2040,7 +2040,7 @@ class OaiCiTest():
if not ue.getIP(): if not ue.getIP():
logging.error("no IP addresses returned") logging.error("no IP addresses returned")
HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE) HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
ues.append(ue) ues.append(ue)
logging.debug(ues) logging.debug(ues)
@@ -2073,7 +2073,7 @@ class OaiCiTest():
if (status_queue.empty()): if (status_queue.empty()):
HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.ALL_PROCESSES_OK)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
else: else:
iperf_status = True iperf_status = True
iperf_noperf = False iperf_noperf = False
@@ -2094,7 +2094,7 @@ class OaiCiTest():
HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', messages) HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', messages)
else: else:
HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', messages) HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', messages)
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
def CheckProcessExist(self, check_eNB, check_OAI_UE,RAN,EPC): def CheckProcessExist(self, check_eNB, check_OAI_UE,RAN,EPC):
multi_jobs = [] multi_jobs = []
@@ -2469,7 +2469,7 @@ class OaiCiTest():
else: else:
HTML.CreateHtmlTestRow('QLog trace is disabled', 'OK', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow('QLog trace is disabled', 'OK', CONST.ALL_PROCESSES_OK)
def TerminateOAIUE(self,HTML,RAN,COTS_UE,EPC,CONTAINERS): def TerminateOAIUE(self,HTML,RAN,EPC,CONTAINERS):
SSH = sshconnection.SSHConnection() SSH = sshconnection.SSHConnection()
SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword) SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
SSH.command('cd ' + self.UESourceCodePath + '/cmake_targets', '\$', 5) SSH.command('cd ' + self.UESourceCodePath + '/cmake_targets', '\$', 5)
@@ -2510,11 +2510,11 @@ class OaiCiTest():
# Not an error then # Not an error then
if (logStatus != CONST.OAI_UE_PROCESS_COULD_NOT_SYNC) or (ueAction != 'Sniffing'): if (logStatus != CONST.OAI_UE_PROCESS_COULD_NOT_SYNC) or (ueAction != 'Sniffing'):
self.Initialize_OAI_UE_args = '' self.Initialize_OAI_UE_args = ''
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
else: else:
if (logStatus == CONST.OAI_UE_PROCESS_COULD_NOT_SYNC): if (logStatus == CONST.OAI_UE_PROCESS_COULD_NOT_SYNC):
self.Initialize_OAI_UE_args = '' self.Initialize_OAI_UE_args = ''
self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.AutoTerminateUEandeNB(HTML,RAN,EPC,CONTAINERS)
else: else:
logging.debug('\u001B[1m' + ueAction + ' Completed \u001B[0m') logging.debug('\u001B[1m' + ueAction + ' Completed \u001B[0m')
HTML.htmlUEFailureMsg='<b>' + ueAction + ' Completed</b>\n' + HTML.htmlUEFailureMsg HTML.htmlUEFailureMsg='<b>' + ueAction + ' Completed</b>\n' + HTML.htmlUEFailureMsg
@@ -2523,14 +2523,15 @@ class OaiCiTest():
else: else:
HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK) HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK)
def AutoTerminateUEandeNB(self,HTML,RAN,COTS_UE,EPC,CONTAINERS): def AutoTerminateUEandeNB(self,HTML,RAN,EPC,CONTAINERS):
# TODO: terminate UE?
if (self.Initialize_OAI_UE_args != ''): if (self.Initialize_OAI_UE_args != ''):
self.testCase_id = 'AUTO-KILL-OAI-UE' self.testCase_id = 'AUTO-KILL-OAI-UE'
HTML.testCase_id = self.testCase_id HTML.testCase_id = self.testCase_id
self.desc = 'Automatic Termination of OAI-UE' self.desc = 'Automatic Termination of OAI-UE'
HTML.desc = self.desc HTML.desc = self.desc
self.ShowTestID() self.ShowTestID()
self.TerminateOAIUE(HTML,RAN,COTS_UE,EPC,CONTAINERS) self.TerminateOAIUE(HTML,RAN,EPC,CONTAINERS)
if (RAN.Initialize_eNB_args != ''): if (RAN.Initialize_eNB_args != ''):
self.testCase_id = 'AUTO-KILL-RAN' self.testCase_id = 'AUTO-KILL-RAN'
HTML.testCase_id = self.testCase_id HTML.testCase_id = self.testCase_id
ci-scripts/main.py
+9 -19
View File
@@ -40,7 +40,6 @@ import constants as CONST
import cls_oaicitest #main class for OAI CI test framework import cls_oaicitest #main class for OAI CI test framework
import cls_physim #class PhySim for physical simulators build and test import cls_physim #class PhySim for physical simulators build and test
import cls_cots_ue #class CotsUe for Airplane mode control
import cls_containerize #class Containerize for all container-based operations on RAN/UE objects import cls_containerize #class Containerize for all container-based operations on RAN/UE objects
import cls_static_code_analysis #class for static code analysis import cls_static_code_analysis #class for static code analysis
import cls_physim1 #class PhySim for physical simulators deploy and run import cls_physim1 #class PhySim for physical simulators deploy and run
@@ -574,15 +573,6 @@ py_param_file_present, py_params, mode = args_parse.ArgsParse(sys.argv,CiTestObj
if py_param_file_present == True: if py_param_file_present == True:
AssignParams(py_params) AssignParams(py_params)
#-----------------------------------------------------------
# COTS UE instanciation
#-----------------------------------------------------------
#COTS_UE instanciation and ADB server init
#ue id and ue mode are retrieved from xml
COTS_UE=cls_cots_ue.CotsUe(CiTestObj.ADBIPAddress, CiTestObj.ADBUserName,CiTestObj.ADBPassword)
#----------------------------------------------------------- #-----------------------------------------------------------
# mode amd XML class (action) analysis # mode amd XML class (action) analysis
#----------------------------------------------------------- #-----------------------------------------------------------
@@ -603,7 +593,7 @@ elif re.match('^TerminateOAIUE$', mode, re.IGNORECASE):
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
sys.exit('Insufficient Parameter') sys.exit('Insufficient Parameter')
signal.signal(signal.SIGUSR1, receive_signal) signal.signal(signal.SIGUSR1, receive_signal)
CiTestObj.TerminateOAIUE(HTML,RAN,COTS_UE,EPC,CONTAINERS) CiTestObj.TerminateOAIUE(HTML,RAN,EPC,CONTAINERS)
elif re.match('^TerminateHSS$', mode, re.IGNORECASE): elif re.match('^TerminateHSS$', mode, re.IGNORECASE):
if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '': if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
HELP.GenericHelp(CONST.Version) HELP.GenericHelp(CONST.Version)
@@ -842,29 +832,29 @@ elif re.match('^TesteNB$', mode, re.IGNORECASE) or re.match('^TestUE$', mode, re
elif action == 'Terminate_eNB': elif action == 'Terminate_eNB':
RAN.TerminateeNB(HTML, EPC) RAN.TerminateeNB(HTML, EPC)
elif action == 'Initialize_UE': elif action == 'Initialize_UE':
CiTestObj.InitializeUE(HTML,RAN, EPC, COTS_UE, CONTAINERS) CiTestObj.InitializeUE(HTML,RAN, EPC, CONTAINERS)
elif action == 'Terminate_UE': elif action == 'Terminate_UE':
CiTestObj.TerminateUE(HTML) CiTestObj.TerminateUE(HTML)
elif action == 'Attach_UE': elif action == 'Attach_UE':
CiTestObj.AttachUE(HTML,RAN,EPC,COTS_UE,CONTAINERS) CiTestObj.AttachUE(HTML,RAN,EPC,CONTAINERS)
elif action == 'Detach_UE': elif action == 'Detach_UE':
CiTestObj.DetachUE(HTML,RAN,EPC,COTS_UE,CONTAINERS) CiTestObj.DetachUE(HTML,RAN,EPC,CONTAINERS)
elif action == 'DataDisable_UE': elif action == 'DataDisable_UE':
CiTestObj.DataDisableUE(HTML) CiTestObj.DataDisableUE(HTML)
elif action == 'DataEnable_UE': elif action == 'DataEnable_UE':
CiTestObj.DataEnableUE(HTML) CiTestObj.DataEnableUE(HTML)
elif action == 'CheckStatusUE': # still used? elif action == 'CheckStatusUE': # still used?
CiTestObj.CheckStatusUE(HTML,RAN,EPC,COTS_UE,CONTAINERS) CiTestObj.CheckStatusUE(HTML,RAN,EPC,CONTAINERS)
elif action == 'Build_OAI_UE': elif action == 'Build_OAI_UE':
CiTestObj.BuildOAIUE(HTML) CiTestObj.BuildOAIUE(HTML)
elif action == 'Initialize_OAI_UE': elif action == 'Initialize_OAI_UE':
CiTestObj.InitializeOAIUE(HTML,RAN,EPC,COTS_UE,CONTAINERS) CiTestObj.InitializeOAIUE(HTML,RAN,EPC,CONTAINERS)
elif action == 'Terminate_OAI_UE': elif action == 'Terminate_OAI_UE':
CiTestObj.TerminateOAIUE(HTML,RAN,COTS_UE,EPC,CONTAINERS) CiTestObj.TerminateOAIUE(HTML,RAN,EPC,CONTAINERS)
elif action == 'Ping': elif action == 'Ping':
CiTestObj.Ping(HTML,RAN,EPC,COTS_UE, CONTAINERS) CiTestObj.Ping(HTML,RAN,EPC,CONTAINERS)
elif action == 'Iperf': elif action == 'Iperf':
CiTestObj.Iperf(HTML,RAN,EPC,COTS_UE, CONTAINERS) CiTestObj.Iperf(HTML,RAN,EPC,CONTAINERS)
elif action == 'Reboot_UE': elif action == 'Reboot_UE':
CiTestObj.RebootUE(HTML,RAN,EPC) CiTestObj.RebootUE(HTML,RAN,EPC)
elif action == 'Initialize_HSS': elif action == 'Initialize_HSS':