Skip to content
Snippets Groups Projects
main.py 156 KiB
Newer Older
				IPAddress = RAN.eNBIPAddress
				UserName = RAN.eNBUserName
				Password = RAN.eNBPassword
			else:
				return -1
		if machine == 'UE':
			if self.UEIPAddress != '' and self.UEUserName != '' and self.UEPassword != '':
				IPAddress = self.UEIPAddress
				UserName = self.UEUserName
				Password = self.UEPassword
		SSH.open(IPAddress, UserName, Password)
		SSH.command('lsb_release -a', '\$', 5)
		result = re.search('Description:\\\\t(?P<os_type>[a-zA-Z0-9\-\_\.\ ]+)', SSH.getBefore())
		if result is not None:
			OsVersion = result.group('os_type')
			logging.debug('OS is: ' + OsVersion)
			SSH.command('hostnamectl', '\$', 5)
			result = re.search('Operating System: (?P<os_type>[a-zA-Z0-9\-\_\.\ ]+)', SSH.getBefore())
			if result is not None:
				OsVersion = result.group('os_type')
				if OsVersion == 'CentOS Linux 7 ':
					SSH.command('cat /etc/redhat-release', '\$', 5)
					result = re.search('CentOS Linux release (?P<os_version>[0-9\.]+)', SSH.getBefore())
					if result is not None:
						OsVersion = OsVersion.replace('7 ', result.group('os_version'))
				logging.debug('OS is: ' + OsVersion)
		result = re.search('uname -r\\\\r\\\\n(?P<kernel_version>[a-zA-Z0-9\-\_\.]+)', SSH.getBefore())
		if result is not None:
			KernelVersion = result.group('kernel_version')
			logging.debug('Kernel Version is: ' + KernelVersion)
		SSH.command('dpkg --list | egrep --color=never libuhd003', '\$', 5)
		result = re.search('libuhd003:amd64 *(?P<uhd_version>[0-9\.]+)', SSH.getBefore())
		if result is not None:
			UhdVersion = result.group('uhd_version')
			logging.debug('UHD Version is: ' + UhdVersion)
			SSH.command('uhd_config_info --version', '\$', 5)
			result = re.search('UHD (?P<uhd_version>[a-zA-Z0-9\.\-]+)', SSH.getBefore())
			if result is not None:
				UhdVersion = result.group('uhd_version')
				logging.debug('UHD Version is: ' + UhdVersion)
		SSH.command('echo ' + Password + ' | sudo -S uhd_find_devices', '\$', 60)
		usrp_boards = re.findall('product: ([0-9A-Za-z]+)\\\\r\\\\n', SSH.getBefore())
		count = 0
		for board in usrp_boards:
			if count == 0:
				UsrpBoard = board
				UsrpBoard += ',' + board
			count += 1
		if count > 0:
			logging.debug('USRP Board(s) : ' + UsrpBoard)
		result = re.search('CPU\(s\): *(?P<nb_cpus>[0-9]+).*Model name: *(?P<model>[a-zA-Z0-9\-\_\.\ \(\)]+).*CPU MHz: *(?P<cpu_mhz>[0-9\.]+)', SSH.getBefore())
		if result is not None:
			CpuNb = result.group('nb_cpus')
			logging.debug('nb_cpus: ' + CpuNb)
			CpuModel = result.group('model')
			logging.debug('model: ' + CpuModel)
			CpuMHz = result.group('cpu_mhz') + ' MHz'
			logging.debug('cpu_mhz: ' + CpuMHz)
Raphael Defosseux's avatar
Raphael Defosseux committed

	def ShowTestID(self):
		logging.debug('\u001B[1m----------------------------------------\u001B[0m')
		logging.debug('\u001B[1mTest ID:' + self.testCase_id + '\u001B[0m')
		logging.debug('\u001B[1m' + self.desc + '\u001B[0m')
		logging.debug('\u001B[1m----------------------------------------\u001B[0m')



#-----------------------------------------------------------
# General Functions
#-----------------------------------------------------------



def CheckClassValidity(xml_class_list,action,id):
	if action not in xml_class_list:
		logging.debug('ERROR: test-case ' + id + ' has unlisted class ' + action + ' ##CHECK xml_class_list.yml')
		resp=False
	else:
		resp=True
	return resp

#assigning parameters to object instance attributes (even if the attributes do not exist !!)
def AssignParams(params_dict):
	for key,value in params_dict.items():
		setattr(CiTestObj, key, value)
		setattr(RAN, key, value)
		setattr(HTML, key, value)
		setattr(ldpc, key, value)



def GetParametersFromXML(action):
	if action == 'Build_eNB':
		RAN.Build_eNB_args=test.findtext('Build_eNB_args')
		forced_workspace_cleanup = test.findtext('forced_workspace_cleanup')
		if (forced_workspace_cleanup is None):
			RAN.Build_eNB_forced_workspace_cleanup=False
		else:
			if re.match('true', forced_workspace_cleanup, re.IGNORECASE):
				RAN.Build_eNB_forced_workspace_cleanup=True
				RAN.Build_eNB_forced_workspace_cleanup=False
		eNB_instance=test.findtext('eNB_instance')
		if (eNB_instance is None):
			RAN.eNB_instance=0
		else:
			RAN.eNB_instance=int(eNB_instance)
		RAN.eNB_serverId=test.findtext('eNB_serverId')
		if (RAN.eNB_serverId is None):
			RAN.eNB_serverId='0'
		xmlBgBuildField = test.findtext('backgroundBuild')
		if (xmlBgBuildField is None):
			RAN.backgroundBuild=False
		else:
			if re.match('true', xmlBgBuildField, re.IGNORECASE):
				RAN.backgroundBuild=True
				RAN.backgroundBuild=False

	if action == 'WaitEndBuild_eNB':
		RAN.Build_eNB_args=test.findtext('Build_eNB_args')
		eNB_instance=test.findtext('eNB_instance')
		if (eNB_instance is None):
			RAN.eNB_instance='0'
		RAN.eNB_serverId=test.findtext('eNB_serverId')
		if (RAN.eNB_serverId is None):
			RAN.eNB_serverId='0'
	if action == 'Initialize_eNB':
		RAN.Initialize_eNB_args=test.findtext('Initialize_eNB_args')
		eNB_instance=test.findtext('eNB_instance')
		if (eNB_instance is None):
			RAN.eNB_instance=0
		else:
			RAN.eNB_instance=int(eNB_instance)
		RAN.eNB_serverId=test.findtext('eNB_serverId')
		if (RAN.eNB_serverId is None):
			RAN.eNB_serverId='0'
			
		#local variable air_interface
		air_interface = test.findtext('air_interface')		
		if (air_interface is None) or (air_interface.lower() not in ['nr','lte','ocp']):
			RAN.air_interface[RAN.eNB_instance] = 'lte-softmodem'
		elif (air_interface.lower() in ['nr','lte']):
			RAN.air_interface[RAN.eNB_instance] = air_interface.lower() +'-softmodem'
		else :
			RAN.air_interface[RAN.eNB_instance] = 'ocp-enb'
	if action == 'Terminate_eNB':
		if (eNB_instance is None):
			RAN.eNB_instance=0
		else:
			RAN.eNB_instance=int(eNB_instance)
		RAN.eNB_serverId=test.findtext('eNB_serverId')
		if (RAN.eNB_serverId is None):
			RAN.eNB_serverId='0'
			
		#local variable air_interface
		air_interface = test.findtext('air_interface')		
		if (air_interface is None) or (air_interface.lower() not in ['nr','lte','ocp']):
			RAN.air_interface[RAN.eNB_instance] = 'lte-softmodem'
		elif (air_interface.lower() in ['nr','lte']):
			RAN.air_interface[RAN.eNB_instance] = air_interface.lower() +'-softmodem'
		else :
			RAN.air_interface[RAN.eNB_instance] = 'ocp-enb'
	if action == 'Attach_UE':
		nbMaxUEtoAttach = test.findtext('nbMaxUEtoAttach')
		if (nbMaxUEtoAttach is None):
			CiTestObj.nbMaxUEtoAttach = -1
			CiTestObj.nbMaxUEtoAttach = int(nbMaxUEtoAttach)
	if action == 'CheckStatusUE':
		expectedNBUE = test.findtext('expectedNbOfConnectedUEs')
		if (expectedNBUE is None):
			CiTestObj.expectedNbOfConnectedUEs = -1
			CiTestObj.expectedNbOfConnectedUEs = int(expectedNBUE)
Boris Djalal's avatar
Boris Djalal committed
	if action == 'Build_OAI_UE':
		CiTestObj.Build_OAI_UE_args = test.findtext('Build_OAI_UE_args')
		CiTestObj.clean_repository = test.findtext('clean_repository')
		if (CiTestObj.clean_repository == 'false'):
			CiTestObj.clean_repository = False
Boris Djalal's avatar
Boris Djalal committed

	if action == 'Initialize_OAI_UE':
		CiTestObj.Initialize_OAI_UE_args = test.findtext('Initialize_OAI_UE_args')
		UE_instance = test.findtext('UE_instance')
		if (UE_instance is None):
			CiTestObj.UE_instance = 0
		else:
			CiTestObj.UE_instance = UE_instance
			
		#local variable air_interface
		air_interface = test.findtext('air_interface')		
		if (air_interface is None) or (air_interface.lower() not in ['nr','lte','ocp']):
		elif (air_interface.lower() in ['nr','lte']):
			CiTestObj.air_interface = air_interface.lower() +'-uesoftmodem'
		else :
			#CiTestObj.air_interface = 'ocp-enb'
			logging.error('OCP UE -- NOT SUPPORTED')
Boris Djalal's avatar
Boris Djalal committed

	if action == 'Terminate_OAI_UE':
		UE_instance=test.findtext('UE_instance')
		if (UE_instance is None):
			CiTestObj.UE_instance = '0'
		else:
			CiTestObj.UE_instance = int(UE_instance)
		
		#local variable air_interface
		air_interface = test.findtext('air_interface')		
		if (air_interface is None) or (air_interface.lower() not in ['nr','lte','ocp']):
			CiTestObj.air_interface = 'lte-uesoftmodem'
		elif (air_interface.lower() in ['nr','lte']):
			CiTestObj.air_interface = air_interface.lower() +'-uesoftmodem'
		else :
			#CiTestObj.air_interface = 'ocp-enb'
			logging.error('OCP UE -- NOT SUPPORTED')
	if action == 'Ping' or action == 'Ping_CatM_module':
		CiTestObj.ping_args = test.findtext('ping_args')
		CiTestObj.ping_packetloss_threshold = test.findtext('ping_packetloss_threshold')
		CiTestObj.iperf_args = test.findtext('iperf_args')
		CiTestObj.iperf_packetloss_threshold = test.findtext('iperf_packetloss_threshold')
		CiTestObj.iperf_profile = test.findtext('iperf_profile')
		if (CiTestObj.iperf_profile is None):
			CiTestObj.iperf_profile = 'balanced'
			if CiTestObj.iperf_profile != 'balanced' and CiTestObj.iperf_profile != 'unbalanced' and CiTestObj.iperf_profile != 'single-ue':
				logging.debug('ERROR: test-case has wrong profile ' + CiTestObj.iperf_profile)
				CiTestObj.iperf_profile = 'balanced'
		CiTestObj.iperf_options = test.findtext('iperf_options')
		if (CiTestObj.iperf_options is None):
			CiTestObj.iperf_options = 'check'
			if CiTestObj.iperf_options != 'check' and CiTestObj.iperf_options != 'sink':
				logging.debug('ERROR: test-case has wrong option ' + CiTestObj.iperf_options)
				CiTestObj.iperf_options = 'check'
	if action == 'IdleSleep':
		string_field = test.findtext('idle_sleep_time_in_sec')
		if (string_field is None):
			CiTestObj.idle_sleep_time = 5
			CiTestObj.idle_sleep_time = int(string_field)
	if action == 'Perform_X2_Handover':
		string_field = test.findtext('x2_ho_options')
		if (string_field is None):
			CiTestObj.x2_ho_options = 'network'
		else:
			if string_field != 'network':
				logging.error('ERROR: test-case has wrong option ' + string_field)
				CiTestObj.x2_ho_options = 'network'
				CiTestObj.x2_ho_options = string_field
	if action == 'Build_PhySim':
		ldpc.buildargs  = test.findtext('physim_build_args')
Remi Hardy's avatar
Remi Hardy committed
		forced_workspace_cleanup = test.findtext('forced_workspace_cleanup')
		if (forced_workspace_cleanup is None):
			ldpc.forced_workspace_cleanup=False
		else:
			if re.match('true', forced_workspace_cleanup, re.IGNORECASE):
				ldpc.forced_workspace_cleanup=True
			else:
				ldpc.forced_workspace_cleanup=False

	if action == 'Run_PhySim':
		ldpc.runargs = test.findtext('physim_run_args')
		
	if action == 'COTS_UE_Airplane':
		COTS_UE.runargs = test.findtext('cots_ue_airplane_args')
#check if given test is in list
#it is in list if one of the strings in 'list' is at the beginning of 'test'
def test_in_list(test, list):
	for check in list:
		check=check.replace('+','')
		if (test.startswith(check)):
			return True
	return False

def receive_signal(signum, frame):
	sys.exit(1)

#-----------------------------------------------------------
#-----------------------------------------------------------

#loading xml action list from yaml
import yaml
Raphael Defosseux's avatar
Raphael Defosseux committed
xml_class_list_file=''
if (os.path.isfile('xml_class_list.yml')):
	xml_class_list_file='xml_class_list.yml'
if (os.path.isfile('ci-scripts/xml_class_list.yml')):
	xml_class_list_file='ci-scripts/xml_class_list.yml'
with open(xml_class_list_file,'r') as file:
    # The FullLoader parameter handles the conversion from YAML
    # scalar values to Python the dictionary format
    xml_class_list = yaml.load(file,Loader=yaml.FullLoader)

 
SSH = sshconnection.SSHConnection()
EPC = epc.EPCManagement()
RAN = ran.RANManagement()
HTML = html.HTMLManagement()
EPC.htmlObj=HTML
RAN.htmlObj=HTML
RAN.epcObj=EPC
ldpc=cls_physim.PhySim()    #create an instance for LDPC test using GPU or CPU build

#-----------------------------------------------------------
# Parsing Command Line Arguments
#-----------------------------------------------------------

py_param_file_present, py_params, mode = args_parse.ArgsParse(sys.argv,CiTestObj,RAN,HTML,EPC,ldpc,HELP)
#-----------------------------------------------------------
# TEMPORARY params management
#-----------------------------------------------------------
#temporary solution for testing:
if py_param_file_present == True:
#print(RAN.__dict__) 
#print(CiTestObj.__dict__) 
#print(HTML.__dict__) 
#print(ldpc.__dict__) 
#-----------------------------------------------------------
# COTS UE instanciation
#-----------------------------------------------------------
#COTS_UE instanciation can only be done here for the moment, due to code architecture
COTS_UE=cls_cots_ue.CotsUe('oppo', CiTestObj.UEIPAddress, CiTestObj.UEUserName,CiTestObj.UEPassword)


#-----------------------------------------------------------
# XML class (action) analysis
#-----------------------------------------------------------
if re.match('^TerminateeNB$', mode, re.IGNORECASE):
	if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	RAN.eNB_serverId='0'
	RAN.eNB_instance='0'
	RAN.eNBSourceCodePath='/tmp/'
elif re.match('^TerminateUE$', mode, re.IGNORECASE):
	if (CiTestObj.ADBIPAddress == '' or CiTestObj.ADBUserName == '' or CiTestObj.ADBPassword == ''):
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	signal.signal(signal.SIGUSR1, receive_signal)
	CiTestObj.TerminateUE()
Boris Djalal's avatar
Boris Djalal committed
elif re.match('^TerminateOAIUE$', mode, re.IGNORECASE):
	if CiTestObj.UEIPAddress == '' or CiTestObj.UEUserName == '' or CiTestObj.UEPassword == '':
		HELP.GenericHelp(CONST.Version)
Boris Djalal's avatar
Boris Djalal committed
		sys.exit('Insufficient Parameter')
	signal.signal(signal.SIGUSR1, receive_signal)
	CiTestObj.TerminateOAIUE()
elif re.match('^TerminateHSS$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.TerminateHSS()
elif re.match('^TerminateMME$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.TerminateMME()
elif re.match('^TerminateSPGW$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath== '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.TerminateSPGW()
elif re.match('^LogCollectBuild$', mode, re.IGNORECASE):
	if (RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '' or RAN.eNBSourceCodePath == '') and (CiTestObj.UEIPAddress == '' or CiTestObj.UEUserName == '' or CiTestObj.UEPassword == '' or CiTestObj.UESourceCodePath == ''):
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	CiTestObj.LogCollectBuild()
elif re.match('^LogCollecteNB$', mode, re.IGNORECASE):
	if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '' or RAN.eNBSourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	RAN.LogCollecteNB()
elif re.match('^LogCollectHSS$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.LogCollectHSS()
elif re.match('^LogCollectMME$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.LogCollectMME()
elif re.match('^LogCollectSPGW$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	EPC.LogCollectSPGW()
elif re.match('^LogCollectPing$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	CiTestObj.LogCollectPing()
elif re.match('^LogCollectIperf$', mode, re.IGNORECASE):
	if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	CiTestObj.LogCollectIperf()
Boris Djalal's avatar
Boris Djalal committed
elif re.match('^LogCollectOAIUE$', mode, re.IGNORECASE):
	if CiTestObj.UEIPAddress == '' or CiTestObj.UEUserName == '' or CiTestObj.UEPassword == '' or CiTestObj.UESourceCodePath == '':
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	CiTestObj.LogCollectOAIUE()
elif re.match('^InitiateHtml$', mode, re.IGNORECASE):
	if (CiTestObj.ADBIPAddress == '' or CiTestObj.ADBUserName == '' or CiTestObj.ADBPassword == ''):
		HELP.GenericHelp(CONST.Version)
		sys.exit('Insufficient Parameter')
	count = 0
	while (count < HTML.nbTestXMLfiles):
		#xml_test_file = cwd + "/" + CiTestObj.testXMLfiles[count]
		xml_test_file = sys.path[0] + "/" + CiTestObj.testXMLfiles[count]
		if (os.path.isfile(xml_test_file)):
			try:
				xmlTree = ET.parse(xml_test_file)
			except:
				print("Error while parsing file: " + xml_test_file)
			xmlRoot = xmlTree.getroot()
			HTML.htmlTabRefs.append(xmlRoot.findtext('htmlTabRef',default='test-tab-' + str(count)))
			HTML.htmlTabNames.append(xmlRoot.findtext('htmlTabName',default='test-tab-' + str(count)))
			HTML.htmlTabIcons.append(xmlRoot.findtext('htmlTabIcon',default='info-sign'))
	if foundCount != HTML.nbTestXMLfiles:
		HTML.nbTestXMLfiles=foundCount
	
	if (CiTestObj.ADBIPAddress != 'none'):
		terminate_ue_flag = False
		CiTestObj.GetAllUEDevices(terminate_ue_flag)
		CiTestObj.GetAllCatMDevices(terminate_ue_flag)
		HTML.SethtmlUEConnected(len(CiTestObj.UEDevices) + len(CiTestObj.CatMDevices))
		HTML.htmlNb_Smartphones=len(CiTestObj.UEDevices)
		HTML.htmlNb_CATM_Modules=len(CiTestObj.CatMDevices)
	HTML.CreateHtmlHeader(CiTestObj.ADBIPAddress)
elif re.match('^FinalizeHtml$', mode, re.IGNORECASE):
	logging.debug('\u001B[1m----------------------------------------\u001B[0m')
	logging.debug('\u001B[1m  Creating HTML footer \u001B[0m')
	logging.debug('\u001B[1m----------------------------------------\u001B[0m')

	CiTestObj.RetrieveSystemVersion('eNB')
	CiTestObj.RetrieveSystemVersion('UE')
	HTML.CreateHtmlFooter(CiTestObj.finalStatus)
elif re.match('^TesteNB$', mode, re.IGNORECASE) or re.match('^TestUE$', mode, re.IGNORECASE):
	if re.match('^TesteNB$', mode, re.IGNORECASE):
		if RAN.eNBIPAddress == '' or RAN.ranRepository == '' or RAN.ranBranch == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '' or RAN.eNBSourceCodePath == '' or EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.Type == '' or EPC.SourceCodePath == '' or CiTestObj.ADBIPAddress == '' or CiTestObj.ADBUserName == '' or CiTestObj.ADBPassword == '':
			HELP.GenericHelp(CONST.Version)
			if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '' or EPC.Type == '':
				HELP.EPCSrvHelp(EPC.IPAddress, EPC.UserName, EPC.Password, EPC.SourceCodePath, EPC.Type)
			if RAN.ranRepository == '':
				HELP.GitSrvHelp(RAN.ranRepository, RAN.ranBranch, RAN.ranCommitID, RAN.ranAllowMerge, RAN.ranTargetBranch)
			if RAN.eNBIPAddress == ''  or RAN.eNBUserName == '' or RAN.eNBPassword == '' or RAN.eNBSourceCodePath == '':
				HELP.eNBSrvHelp(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword, RAN.eNBSourceCodePath)
		if (EPC.IPAddress!= '') and (EPC.IPAddress != 'none'):
			SSH.copyout(EPC.IPAddress, EPC.UserName, EPC.Password, cwd + "/tcp_iperf_stats.awk", "/tmp")
			SSH.copyout(EPC.IPAddress, EPC.UserName, EPC.Password, cwd + "/active_net_interfaces.awk", "/tmp")
		if CiTestObj.UEIPAddress == '' or CiTestObj.ranRepository == '' or CiTestObj.ranBranch == '' or CiTestObj.UEUserName == '' or CiTestObj.UEPassword == '' or CiTestObj.UESourceCodePath == '':
			HELP.GenericHelp(CONST.Version)
	#read test_case_list.xml file
	# if no parameters for XML file, use default value
		xml_test_file = cwd + "/test_case_list.xml"
		xml_test_file = cwd + "/" + CiTestObj.testXMLfiles[0]

	xmlTree = ET.parse(xml_test_file)
	xmlRoot = xmlTree.getroot()

	exclusion_tests=xmlRoot.findtext('TestCaseExclusionList',default='')
	requested_tests=xmlRoot.findtext('TestCaseRequestedList',default='')
	if (HTML.nbTestXMLfiles == 1):
		HTML.htmlTabRefs.append(xmlRoot.findtext('htmlTabRef',default='test-tab-0'))
		HTML.htmlTabNames.append(xmlRoot.findtext('htmlTabName',default='Test-0'))
		repeatCount = xmlRoot.findtext('repeatCount',default='1')
		CiTestObj.repeatCounts.append(int(repeatCount))
	all_tests=xmlRoot.findall('testCase')

	exclusion_tests=exclusion_tests.split()
	requested_tests=requested_tests.split()

	#check that exclusion tests are well formatted
	#(6 digits or less than 6 digits followed by +)
	for test in exclusion_tests:
		if     (not re.match('^[0-9]{6}$', test) and
				not re.match('^[0-9]{1,5}\+$', test)):
			logging.debug('ERROR: exclusion test is invalidly formatted: ' + test)
			sys.exit(1)
		else:
			logging.debug(test)

	#check that requested tests are well formatted
	#(6 digits or less than 6 digits followed by +)
	#be verbose
	for test in requested_tests:
		if     (re.match('^[0-9]{6}$', test) or
				re.match('^[0-9]{1,5}\+$', test)):
			logging.debug('INFO: test group/case requested: ' + test)
		else:
			logging.debug('ERROR: requested test is invalidly formatted: ' + test)
			sys.exit(1)
	if (EPC.IPAddress != '') and (EPC.IPAddress != 'none'):
		CiTestObj.CheckFlexranCtrlInstallation()

	#get the list of tests to be done
	todo_tests=[]
	for test in requested_tests:
		if    (test_in_list(test, exclusion_tests)):
			logging.debug('INFO: test will be skipped: ' + test)
		else:
			#logging.debug('INFO: test will be run: ' + test)
	signal.signal(signal.SIGUSR1, receive_signal)
	if (CiTestObj.ADBIPAddress != 'none'):
		terminate_ue_flag = False
		CiTestObj.GetAllUEDevices(terminate_ue_flag)
		CiTestObj.GetAllCatMDevices(terminate_ue_flag)
	else:
		CiTestObj.UEDevices.append('OAI-UE')
	HTML.SethtmlUEConnected(len(CiTestObj.UEDevices) + len(CiTestObj.CatMDevices))
	HTML.CreateHtmlTabHeader()
	CiTestObj.FailReportCnt = 0
	RAN.prematureExit=True
	HTML.startTime=int(round(time.time() * 1000))
	while CiTestObj.FailReportCnt < CiTestObj.repeatCounts[0] and RAN.prematureExit:
		RAN.prematureExit=False
		# At every iteratin of the retry loop, a separator will be added
		# pass CiTestObj.FailReportCnt as parameter of HTML.CreateHtmlRetrySeparator
		HTML.CreateHtmlRetrySeparator(CiTestObj.FailReportCnt)
		for test_case_id in todo_tests:
			if RAN.prematureExit:
				if RAN.prematureExit:
				id = test.get('id')
				if test_case_id != id:
					continue
				CiTestObj.testCase_id = id
				HTML.testCase_id=CiTestObj.testCase_id
				EPC.testCase_id=CiTestObj.testCase_id
				CiTestObj.desc = test.findtext('desc')
				action = test.findtext('class')
				if (CheckClassValidity(xml_class_list, action, id) == False):
				CiTestObj.ShowTestID()
				GetParametersFromXML(action)
				if action == 'Initialize_UE' or action == 'Attach_UE' or action == 'Detach_UE' or action == 'Ping' or action == 'Iperf' or action == 'Reboot_UE' or action == 'DataDisable_UE' or action == 'DataEnable_UE' or action == 'CheckStatusUE':
					if (CiTestObj.ADBIPAddress != 'none'):
						CiTestObj.GetAllUEDevices(terminate_ue_flag)
				elif action == 'WaitEndBuild_eNB':
				elif action == 'Initialize_eNB':
					RAN.pStatus=CiTestObj.CheckProcessExist(check_eNB, check_OAI_UE)
				elif action == 'Terminate_eNB':
				elif action == 'Initialize_UE':
					CiTestObj.InitializeUE()
				elif action == 'Terminate_UE':
					CiTestObj.TerminateUE()
					CiTestObj.AttachUE()
					CiTestObj.DetachUE()
				elif action == 'DataDisable_UE':
					CiTestObj.DataDisableUE()
				elif action == 'DataEnable_UE':
					CiTestObj.DataEnableUE()
				elif action == 'CheckStatusUE':
					CiTestObj.CheckStatusUE()
				elif action == 'Build_OAI_UE':
					CiTestObj.BuildOAIUE()
				elif action == 'Initialize_OAI_UE':
					CiTestObj.InitializeOAIUE()
				elif action == 'Terminate_OAI_UE':
					CiTestObj.TerminateOAIUE()
				elif action == 'Initialize_CatM_module':
					CiTestObj.InitializeCatM()
				elif action == 'Terminate_CatM_module':
					CiTestObj.TerminateCatM()
				elif action == 'Attach_CatM_module':
					CiTestObj.AttachCatM()
				elif action == 'Detach_CatM_module':
					CiTestObj.TerminateCatM()
				elif action == 'Ping_CatM_module':
					CiTestObj.PingCatM()
					CiTestObj.Iperf()
					CiTestObj.RebootUE()
				elif action == 'Initialize_HSS':
					EPC.InitializeHSS()
				elif action == 'Terminate_HSS':
				elif action == 'Initialize_MME':
					EPC.InitializeMME()
				elif action == 'Terminate_MME':
				elif action == 'Initialize_SPGW':
					EPC.InitializeSPGW()
				elif action == 'Terminate_SPGW':
					EPC.TerminateSPGW()
				elif action == 'Initialize_FlexranCtrl':
					CiTestObj.InitializeFlexranCtrl()
				elif action == 'Terminate_FlexranCtrl':
					CiTestObj.TerminateFlexranCtrl()
					CiTestObj.IdleSleep()
				elif action == 'Perform_X2_Handover':
					CiTestObj.Perform_X2_Handover()
				elif action == 'Build_PhySim':
					HTML=ldpc.Build_PhySim(HTML,CONST)
					if ldpc.exitStatus==1:sys.exit()
				elif action == 'Run_PhySim':
Remi Hardy's avatar
Remi Hardy committed
					HTML=ldpc.Run_PhySim(HTML,CONST,id)
				elif action == 'COTS_UE_Airplane':
					COTS_UE.Set_Airplane(COTS_UE.runargs)
					sys.exit('Invalid class (action) from xml')
		CiTestObj.FailReportCnt += 1
	if CiTestObj.FailReportCnt == CiTestObj.repeatCounts[0] and RAN.prematureExit:
		logging.debug('Testsuite failed ' + str(CiTestObj.FailReportCnt) + ' time(s)')
		logging.info('Testsuite passed after ' + str(CiTestObj.FailReportCnt) + ' time(s)')
elif re.match('^LoadParams$', mode, re.IGNORECASE):
	pass
	HELP.GenericHelp(CONST.Version)
sys.exit(0)