adding module UE detach, to be tested

ci-scripts/cls_module_ue.py

@@ -78,6 +78,7 @@ class Module_UE:
 			logging.debug(self.Process + " process NOT found")
 			return False 
 
+	#Wakeup/Detach can probably be improved with encapsulation of the command such def Command(self, command)
 	#this method wakes up the module by calling the specified python script 
 	def WakeUp(self):
 		mySSH = sshconnection.SSHConnection()
@@ -87,6 +88,15 @@ class Module_UE:
 		logging.debug("Module wake-up")
 		mySSH.close()
 
+	#this method detaches the module by calling the specified python script 
+	def Detach(self):
+		mySSH = sshconnection.SSHConnection()
+		mySSH.open(self.HostIPAddress, self.HostUsername, self.HostPassword)
+		mySSH.command('echo ' + self.HostPassword + ' | sudo -S python3 ' + self.DetachScript + ' ','\$',5)
+		time.sleep(5)
+		logging.debug("Module detach")
+		mySSH.close()
+
 	#this method retrieves the Module IP address (not the Host IP address) 
 	def GetModuleIPAddress(self):
 		HOST=self.HostIPAddress

ci-scripts/cls_oaicitest.py

@@ -392,7 +392,7 @@ class OaiCiTest():
 				Module_UE.GetModuleIPAddress()
 				HTML.CreateHtmlTestRow(Module_UE.UEIPAddress, 'OK', CONST.ALL_PROCESSES_OK)	
 				self.UEIPAddresses.append(Module_UE.UEIPAddress)
-		logging.debug(self.UEIPAddresses)
+		logging.debug('UEs IP addresses : '+ self.UEIPAddresses)
 
 	def InitializeOAIUE(self,HTML,RAN,EPC,COTS_UE):
 		if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '':
@@ -957,31 +957,32 @@ class OaiCiTest():
 		except:
 			os.kill(os.getppid(),signal.SIGUSR1)
 
-	def AttachUE(self,HTML,RAN,EPC,COTS_UE):
+	def AttachUE(self,HTML,RAN,EPC,COTS_UE,InfraUE):
-		if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
+		if self.ue_id=='':#no ID specified, then it is a COTS controlled by ADB
-			HELP.GenericHelp(CONST.Version)
+			if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
-			sys.exit('Insufficient Parameter')
+				HELP.GenericHelp(CONST.Version)
-		check_eNB = True
+				sys.exit('Insufficient Parameter')
-		check_OAI_UE = False
+			check_eNB = True
-		pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
+			check_OAI_UE = False
-		if (pStatus < 0):
+			pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
-			HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
+			if (pStatus < 0):
-			self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC)
+				HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
-			return
+				self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC)
-		multi_jobs = []
+				return
-		status_queue = SimpleQueue()
+			multi_jobs = []
-		lock = Lock()
+			status_queue = SimpleQueue()
-		nb_ue_to_connect = 0
+			lock = Lock()
-		for device_id in self.UEDevices:
+			nb_ue_to_connect = 0
-			if (self.nbMaxUEtoAttach == -1) or (nb_ue_to_connect < self.nbMaxUEtoAttach):
+			for device_id in self.UEDevices:
-				self.UEDevicesStatus[nb_ue_to_connect] = CONST.UE_STATUS_ATTACHING
+				if (self.nbMaxUEtoAttach == -1) or (nb_ue_to_connect < self.nbMaxUEtoAttach):
-				p = Process(target = self.AttachUE_common, args = (device_id, status_queue, lock,nb_ue_to_connect,COTS_UE,))
+					self.UEDevicesStatus[nb_ue_to_connect] = CONST.UE_STATUS_ATTACHING
-				p.daemon = True
+					p = Process(target = self.AttachUE_common, args = (device_id, status_queue, lock,nb_ue_to_connect,COTS_UE,))
-				p.start()
+					p.daemon = True
-				multi_jobs.append(p)
+					p.start()
-			nb_ue_to_connect = nb_ue_to_connect + 1
+					multi_jobs.append(p)
-		for job in multi_jobs:
+				nb_ue_to_connect = nb_ue_to_connect + 1
-			job.join()
+			for job in multi_jobs:
+				job.join()
 
 		if (status_queue.empty()):
 			HTML.CreateHtmlTestRow('N/A', 'KO', CONST.ALL_PROCESSES_OK)
@@ -1039,36 +1040,44 @@ class OaiCiTest():
 			os.kill(os.getppid(),signal.SIGUSR1)
 
 	def DetachUE(self,HTML,RAN,EPC,COTS_UE):
-		if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
+		if self.ue_id=='':#no ID specified, then it is a COTS controlled by ADB
-			HELP.GenericHelp(CONST.Version)
+			if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
-			sys.exit('Insufficient Parameter')
+				HELP.GenericHelp(CONST.Version)
-		check_eNB = True
+				sys.exit('Insufficient Parameter')
-		check_OAI_UE = False
+			check_eNB = True
-		pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
+			check_OAI_UE = False
-		if (pStatus < 0):
+			pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
-			HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
+			if (pStatus < 0):
-			self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC)
+				HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
-			return
+				self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE,EPC)
-		multi_jobs = []
+				return
-		cnt = 0
+			multi_jobs = []
-		for device_id in self.UEDevices:
+			cnt = 0
-			self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHING
+			for device_id in self.UEDevices:
-			p = Process(target = self.DetachUE_common, args = (device_id,cnt,COTS_UE,))
+				self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHING
-			p.daemon = True
+				p = Process(target = self.DetachUE_common, args = (device_id,cnt,COTS_UE,))
-			p.start()
+				p.daemon = True
-			multi_jobs.append(p)
+				p.start()
-			cnt += 1
+				multi_jobs.append(p)
-		for job in multi_jobs:
+				cnt += 1
-			job.join()
+			for job in multi_jobs:
-		HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK)
+				job.join()
-		result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
+			HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK)
-		if result is not None:
+			result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
-			logging.debug('Waiting 5 seconds to fill up record file')
+			if result is not None:
-			time.sleep(5)
+				logging.debug('Waiting 5 seconds to fill up record file')
-		cnt = 0
+				time.sleep(5)
-		while cnt < len(self.UEDevices):
+			cnt = 0
-			self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHED
+			while cnt < len(self.UEDevices):
-			cnt += 1
+				self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHED
+				cnt += 1
+		else:#if an ID is specified, it is a module from the yaml infrastructure file
+			Module_UE = cls_module_ue.Module_UE(InfraUE.ci_ue_infra[self.ue_id])
+			is_module=Module_UE.CheckIsModule()
+			if is_module:
+				Module_UE.Detach()
+				Module_UE.GetModuleIPAddress()
+				HTML.CreateHtmlTestRow(Module_UE.UEIPAddress, 'OK', CONST.ALL_PROCESSES_OK)			
 
 	def RebootUE_common(self, device_id):
 		try:

ci-scripts/main.py

@@ -200,6 +200,12 @@ def GetParametersFromXML(action):
 		else:
 			CiTestObj.ue_id = ue_id
 
+	elif action == 'Detach_UE':
+		ue_id = test.findtext('id')
+		if (ue_id is None):
+			CiTestObj.ue_id = ""
+		else:
+			CiTestObj.ue_id = ue_id
 
 	elif action == 'Attach_UE':
 		nbMaxUEtoAttach = test.findtext('nbMaxUEtoAttach')