CAT-M module Attachment Completed in ' + str(attach_cnt+4) + ' seconds' + if (nRSRQ is not None) and (nRSRP is not None): + html_cell += '\n RSRQ = ' + str(-20+(nRSRQ/2)) + ' dB' + html_cell += '\n RSRP = ' + str(-140+nRSRP) + ' dBm' + else: + html_cell += '' + html_queue.put(html_cell) + HTML.CreateHtmlTestRowQueue('N/A', 'OK', 1, html_queue) + else: + logging.error('\u001B[1m CAT-M module Attachment Failed\u001B[0m') + html_cell = '
CAT-M module Attachment Failed' + html_queue.put(html_cell) + HTML.CreateHtmlTestRowQueue('N/A', 'KO', 1, html_queue) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + + def PingCatM(self,HTML,RAN,EPC,COTS_UE): + if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': + HELP.GenericHelp(CONST.Version) + sys.exit('Insufficient Parameter') + check_eNB = True + check_OAI_UE = False + pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) + if (pStatus < 0): + HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + return + try: + statusQueue = SimpleQueue() + lock = Lock() + SSH = sshconnection.SSHConnection() + SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password) + SSH.command('cd ' + EPC.SourceCodePath, '\$', 5) + SSH.command('cd scripts', '\$', 5) + if re.match('OAI', EPC.Type, re.IGNORECASE): + logging.debug('Using the OAI EPC HSS: not implemented yet') + HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) + HTML.CreateHtmlTabFooter(False) + sys.exit(1) + else: + SSH.command('egrep --color=never "Allocated ipv4 addr" /opt/ltebox/var/log/xGwLog.0', '\$', 5) + result = re.search('Allocated ipv4 addr: (?P
CAT-M module\nIP Address : ' + moduleIPAddr + '\n' + qMsg + '' + statusQueue.put(html_cell) + if (packetLossOK): + HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', 1, statusQueue) + else: + HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', 1, statusQueue) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + except: + os.kill(os.getppid(),signal.SIGUSR1) + + def AttachUE_common(self, device_id, statusQueue, lock, idx,COTS_UE): + try: + SSH = sshconnection.SSHConnection() + SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword) + if self.ADBCentralized: + #RH quick add on to integrate cots control defined by yaml + #if device Id exists in yaml dictionary, we execute the new procedure defined in cots_ue class + #otherwise we use the legacy procedure + if COTS_UE.Check_Exists(device_id): + #switch device to Airplane mode OFF (ie Radio ON) + COTS_UE.Set_Airplane(device_id, 'OFF') + elif device_id == '84B7N16418004022': + SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/on"', '\$', 60) + else: + SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/on', '\$', 60) + else: + # airplane mode off // radio on + SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOnCmd[idx], '\$', 60) + time.sleep(2) + max_count = 45 + count = max_count + while count > 0: + if self.ADBCentralized: + SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "dumpsys telephony.registry" | grep -m 1 mDataConnectionState', '\$', 15) + else: + SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "dumpsys telephony.registry"\' | grep -m 1 mDataConnectionState', '\$', 60) + result = re.search('mDataConnectionState.*=(?P
UE (' + device_id + ')\n' + message + ''
+ else:
+ html_cell = 'UE (' + device_id + ')\n' + message + ' in ' + str(count + 2) + ' seconds'
+ html_queue.put(html_cell)
+ if (attach_status):
+ cnt = 0
+ while cnt < len(self.UEDevices):
+ if self.UEDevicesStatus[cnt] == CONST.UE_STATUS_ATTACHING:
+ self.UEDevicesStatus[cnt] = CONST.UE_STATUS_ATTACHED
+ cnt += 1
+ HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
+ result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
+ if result is not None:
+ logging.debug('Waiting 5 seconds to fill up record file')
+ time.sleep(5)
+ else:
+ HTML.CreateHtmlTestRowQueue('N/A', 'KO', len(self.UEDevices), html_queue)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+
+ def DetachUE_common(self, device_id, idx,COTS_UE):
+ try:
+ SSH = sshconnection.SSHConnection()
+ SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
+ if self.ADBCentralized:
+ #RH quick add on to integrate cots control defined by yaml
+ #if device id exists in yaml dictionary, we execute the new procedure defined in cots_ue class
+ #otherwise we use the legacy procedure
+ if COTS_UE.Check_Exists(device_id):
+ #switch device to Airplane mode ON (ie Radio OFF)
+ COTS_UE.Set_Airplane(device_id,'ON')
+ elif device_id == '84B7N16418004022':
+ SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/off"', '\$', 60)
+ else:
+ SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/off', '\$', 60)
+ else:
+ SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOffCmd[idx], '\$', 60)
+ logging.debug('\u001B[1mUE (' + device_id + ') Detach Completed\u001B[0m')
+ SSH.close()
+ except:
+ os.kill(os.getppid(),signal.SIGUSR1)
+
+ def DetachUE(self,HTML,RAN,EPC,COTS_UE):
+ if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
+ HELP.GenericHelp(CONST.Version)
+ sys.exit('Insufficient Parameter')
+ check_eNB = True
+ check_OAI_UE = False
+ pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
+ if (pStatus < 0):
+ HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+ return
+ multi_jobs = []
+ cnt = 0
+ for device_id in self.UEDevices:
+ self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHING
+ p = Process(target = self.DetachUE_common, args = (device_id,cnt,COTS_UE,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ cnt += 1
+ for job in multi_jobs:
+ job.join()
+ HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK)
+ result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
+ if result is not None:
+ logging.debug('Waiting 5 seconds to fill up record file')
+ time.sleep(5)
+ cnt = 0
+ while cnt < len(self.UEDevices):
+ self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHED
+ cnt += 1
+
+ def RebootUE_common(self, device_id):
+ try:
+ SSH = sshconnection.SSHConnection()
+ SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
+ previousmDataConnectionStates = []
+ # Save mDataConnectionState
+ SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell dumpsys telephony.registry | grep mDataConnectionState', '\$', 15)
+ SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell dumpsys telephony.registry | grep mDataConnectionState', '\$', 15)
+ result = re.search('mDataConnectionState.*=(?PUE (' + device_id + ')\n' + message + ''
+ html_queue.put(html_cell)
+ if check_status and passStatus:
+ HTML.CreateHtmlTestRowQueue(htmlOptions, 'OK', len(self.UEDevices), html_queue)
+ else:
+ HTML.CreateHtmlTestRowQueue(htmlOptions, 'KO', len(self.UEDevices), html_queue)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+
+ def GetAllUEIPAddresses(self):
+ SSH = sshconnection.SSHConnection()
+ if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
+ HELP.GenericHelp(CONST.Version)
+ sys.exit('Insufficient Parameter')
+ ue_ip_status = 0
+ self.UEIPAddresses = []
+ if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'):
+ if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '':
+ HELP.GenericHelp(CONST.Version)
+ sys.exit('Insufficient Parameter')
+
+ SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
+ SSH.command('ifconfig oaitun_ue1', '\$', 4)
+ result = re.search('inet addr:(?POAI UE ping result\n' + qMsg + '' + html_queue.put(html_cell) + HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue) + + def PingNoS1(self,HTML,RAN,EPC,COTS_UE): + SSH=sshconnection.SSHConnection() + check_eNB = True + check_OAI_UE = True + pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) + if (pStatus < 0): + HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + return + ping_from_eNB = re.search('oaitun_enb1', str(self.ping_args)) + if ping_from_eNB is not None: + if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '': + HELP.GenericHelp(CONST.Version) + sys.exit('Insufficient Parameter') + else: + if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '': + HELP.GenericHelp(CONST.Version) + sys.exit('Insufficient Parameter') + try: + if ping_from_eNB is not None: + SSH.open(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword) + SSH.command('cd ' + RAN.eNBSourceCodePath + '/cmake_targets/', '\$', 5) + else: + SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword) + SSH.command('cd ' + self.UESourceCodePath + '/cmake_targets/', '\$', 5) + ping_time = re.findall("-c (\d+)",str(self.ping_args)) + ping_status = SSH.command('stdbuf -o0 ping ' + self.ping_args + ' 2>&1 | stdbuf -o0 tee ping_' + self.testCase_id + '.log', '\$', int(ping_time[0])*1.5) + # TIMEOUT CASE + if ping_status < 0: + message = 'Ping with OAI UE crashed due to TIMEOUT!' + logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m') + self.PingNoS1_wrong_exit(message,HTML) + return + result = re.search(', (?P
OAI UE ping result\n' + qMsg + '' + html_queue.put(html_cell) + if packetLossOK: + HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue) + else: + HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue) + + # copying on the EPC server for logCollection + if ping_from_eNB is not None: + copyin_res = SSH.copyin(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword, RAN.eNBSourceCodePath + '/cmake_targets/ping_' + self.testCase_id + '.log', '.') + else: + copyin_res = SSH.copyin(self.UEIPAddress, self.UEUserName, self.UEPassword, self.UESourceCodePath + '/cmake_targets/ping_' + self.testCase_id + '.log', '.') + if (copyin_res == 0): + SSH.copyout(EPC.IPAddress, EPC.UserName, EPC.Password, 'ping_' + self.testCase_id + '.log', EPC.SourceCodePath + '/scripts') + except: + os.kill(os.getppid(),signal.SIGUSR1) + + def Ping(self,HTML,RAN,EPC,COTS_UE): + result = re.search('noS1', str(RAN.Initialize_eNB_args)) + if result is not None: + self.PingNoS1(HTML,RAN,EPC,COTS_UE) + return + if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': + HELP.GenericHelp(CONST.Version) + sys.exit('Insufficient Parameter') + check_eNB = True + if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'): + check_OAI_UE = True + else: + check_OAI_UE = False + pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC) + if (pStatus < 0): + HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + return + ueIpStatus = self.GetAllUEIPAddresses() + if (ueIpStatus < 0): + HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + return + multi_jobs = [] + i = 0 + lock = Lock() + status_queue = SimpleQueue() + for UE_IPAddress in self.UEIPAddresses: + device_id = self.UEDevices[i] + p = Process(target = self.Ping_common, args = (lock,UE_IPAddress,device_id,status_queue,EPC,)) + p.daemon = True + p.start() + multi_jobs.append(p) + i = i + 1 + for job in multi_jobs: + job.join() + + if (status_queue.empty()): + HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.ALL_PROCESSES_OK) + self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE) + else: + ping_status = True + html_queue = SimpleQueue() + while (not status_queue.empty()): + count = status_queue.get() + if (count < 0): + ping_status = False + device_id = status_queue.get() + ip_addr = status_queue.get() + message = status_queue.get() + html_cell = '
UE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
+ html_queue.put(html_cell)
+ if (ping_status):
+ HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue)
+ else:
+ HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+
+ def Iperf_ComputeTime(self):
+ result = re.search('-t (?PUE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
+ html_queue.put(html_cell)
+ if (iperf_noperf and iperf_status):
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'PERF NOT MET', len(self.UEDevices), html_queue)
+ elif (iperf_status):
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', len(self.UEDevices), html_queue)
+ else:
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', len(self.UEDevices), html_queue)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+
+ def Iperf(self,HTML,RAN,EPC,COTS_UE):
+ result = re.search('noS1', str(RAN.Initialize_eNB_args))
+ if result is not None:
+ self.IperfNoS1(HTML,RAN,EPC,COTS_UE)
+ return
+ if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '' or self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
+ HELP.GenericHelp(CONST.Version)
+ sys.exit('Insufficient Parameter')
+ check_eNB = True
+ if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'):
+ check_OAI_UE = True
+ else:
+ check_OAI_UE = False
+ pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE,RAN,EPC)
+ if (pStatus < 0):
+ HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+ return
+ ueIpStatus = self.GetAllUEIPAddresses()
+ if (ueIpStatus < 0):
+ logging.debug('going here')
+ HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+ return
+
+ self.dummyIperfVersion = '2.0.10'
+ #cmd = 'iperf --version'
+ #logging.debug(cmd + '\n')
+ #iperfStdout = subprocess.check_output(cmd, shell=True, universal_newlines=True)
+ #result = re.search('iperf version 2.0.5', str(iperfStdout.strip()))
+ #if result is not None:
+ # dummyIperfVersion = '2.0.5'
+ #result = re.search('iperf version 2.0.10', str(iperfStdout.strip()))
+ #if result is not None:
+ # dummyIperfVersion = '2.0.10'
+
+ multi_jobs = []
+ i = 0
+ ue_num = len(self.UEIPAddresses)
+ lock = Lock()
+ status_queue = SimpleQueue()
+ for UE_IPAddress in self.UEIPAddresses:
+ device_id = self.UEDevices[i]
+ p = Process(target = self.Iperf_common, args = (lock,UE_IPAddress,device_id,i,ue_num,status_queue,EPC,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ i = i + 1
+ for job in multi_jobs:
+ job.join()
+
+ if (status_queue.empty()):
+ HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.ALL_PROCESSES_OK)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+ else:
+ iperf_status = True
+ iperf_noperf = False
+ html_queue = SimpleQueue()
+ while (not status_queue.empty()):
+ count = status_queue.get()
+ if (count < 0):
+ iperf_status = False
+ if (count > 0):
+ iperf_noperf = True
+ device_id = status_queue.get()
+ ip_addr = status_queue.get()
+ message = status_queue.get()
+ html_cell = 'UE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
+ html_queue.put(html_cell)
+ if (iperf_noperf and iperf_status):
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'PERF NOT MET', len(self.UEDevices), html_queue)
+ elif (iperf_status):
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', len(self.UEDevices), html_queue)
+ else:
+ HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', len(self.UEDevices), html_queue)
+ self.AutoTerminateUEandeNB(HTML,RAN,COTS_UE)
+
+ def CheckProcessExist(self, check_eNB, check_OAI_UE,RAN,EPC):
+ multi_jobs = []
+ status_queue = SimpleQueue()
+ # in noS1 config, no need to check status from EPC
+ # in gNB also currently no need to check
+ result = re.search('noS1|band78', str(RAN.Initialize_eNB_args))
+ if result is None:
+ p = Process(target = EPC.CheckHSSProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ p = Process(target = EPC.CheckMMEProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ p = Process(target = EPC.CheckSPGWProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ else:
+ if (check_eNB == False) and (check_OAI_UE == False):
+ return 0
+ if check_eNB:
+ p = Process(target = RAN.CheckeNBProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ if check_OAI_UE:
+ p = Process(target = self.CheckOAIUEProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ for job in multi_jobs:
+ job.join()
+
+ if (status_queue.empty()):
+ return -15
+ else:
+ result = 0
+ while (not status_queue.empty()):
+ status = status_queue.get()
+ if (status < 0):
+ result = status
+ if result == CONST.ENB_PROCESS_FAILED:
+ fileCheck = re.search('enb_', str(RAN.eNBLogFiles[0]))
+ if fileCheck is not None:
+ SSH.copyin(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword, RAN.eNBSourceCodePath + '/cmake_targets/' + RAN.eNBLogFiles[0], '.')
+ logStatus = RAN.AnalyzeLogFile_eNB(RAN.eNBLogFiles[0])
+ if logStatus < 0:
+ result = logStatus
+ RAN.eNBLogFiles[0]=''
+ if RAN.flexranCtrlInstalled and RAN.flexranCtrlStarted:
+ self.TerminateFlexranCtrl()
+ return result
+
+ def CheckOAIUEProcessExist(self, initialize_OAI_UE_flag,HTML,RAN):
+ multi_jobs = []
+ status_queue = SimpleQueue()
+ if initialize_OAI_UE_flag == False:
+ p = Process(target = self.CheckOAIUEProcess, args = (status_queue,))
+ p.daemon = True
+ p.start()
+ multi_jobs.append(p)
+ for job in multi_jobs:
+ job.join()
+
+ if (status_queue.empty()):
+ return -15
+ else:
+ result = 0
+ while (not status_queue.empty()):
+ status = status_queue.get()
+ if (status < 0):
+ result = status
+ if result == CONST.OAI_UE_PROCESS_FAILED:
+ fileCheck = re.search('ue_', str(self.UELogFile))
+ if fileCheck is not None:
+ SSH.copyin(self.UEIPAddress, self.UEUserName, self.UEPassword, self.UESourceCodePath + '/cmake_targets/' + self.UELogFile, '.')
+ logStatus = self.AnalyzeLogFile_UE(self.UELogFile,HTML,RAN)
+ if logStatus < 0:
+ result = logStatus
+ return result
+
+ def CheckOAIUEProcess(self, status_queue):
+ try:
+ SSH = sshconnection.SSHConnection()
+ SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
+ SSH.command('stdbuf -o0 ps -aux | grep --color=never ' + self.air_interface + ' | grep -v grep', '\$', 5)
+ result = re.search(self.air_interface, SSH.getBefore())
+ if result is None:
+ logging.debug('\u001B[1;37;41m OAI UE Process Not Found! \u001B[0m')
+ status_queue.put(CONST.OAI_UE_PROCESS_FAILED)
+ else:
+ status_queue.put(CONST.OAI_UE_PROCESS_OK)
+ SSH.close()
+ except:
+ os.kill(os.getppid(),signal.SIGUSR1)
+
+
+ def AnalyzeLogFile_UE(self, UElogFile,HTML,RAN):
+ if (not os.path.isfile('./' + UElogFile)):
+ return -1
+ ue_log_file = open('./' + UElogFile, 'r')
+ exitSignalReceived = False
+ foundAssertion = False
+ msgAssertion = ''
+ msgLine = 0
+ foundSegFault = False
+ foundRealTimeIssue = False
+ uciStatMsgCount = 0
+ pdcpDataReqFailedCount = 0
+ badDciCount = 0
+ f1aRetransmissionCount = 0
+ fatalErrorCount = 0
+ macBsrTimerExpiredCount = 0
+ rrcConnectionRecfgComplete = 0
+ no_cell_sync_found = False
+ mib_found = False
+ frequency_found = False
+ plmn_found = False
+ nrUEFlag = False
+ nrDecodeMib = 0
+ nrFoundDCI = 0
+ nrCRCOK = 0
+ mbms_messages = 0
+ HTML.htmlUEFailureMsg=''
+ global_status = CONST.ALL_PROCESSES_OK
+ for line in ue_log_file.readlines():
+ result = re.search('nr_synchro_time', str(line))
+ if result is not None:
+ nrUEFlag = True
+ if nrUEFlag:
+ result = re.search('decode mib', str(line))
+ if result is not None:
+ nrDecodeMib += 1
+ result = re.search('found 1 DCIs', str(line))
+ if result is not None:
+ nrFoundDCI += 1
+ result = re.search('CRC OK', str(line))
+ if result is not None:
+ nrCRCOK += 1
+ result = re.search('Exiting OAI softmodem', str(line))
+ if result is not None:
+ exitSignalReceived = True
+ result = re.search('System error|[Ss]egmentation [Ff]ault|======= Backtrace: =========|======= Memory map: ========', str(line))
+ if result is not None and not exitSignalReceived:
+ foundSegFault = True
+ result = re.search('[Cc]ore [dD]ump', str(line))
+ if result is not None and not exitSignalReceived:
+ foundSegFault = True
+ result = re.search('./lte-uesoftmodem', str(line))
+ if result is not None and not exitSignalReceived:
+ foundSegFault = True
+ result = re.search('[Aa]ssertion', str(line))
+ if result is not None and not exitSignalReceived:
+ foundAssertion = True
+ result = re.search('LLL', str(line))
+ if result is not None and not exitSignalReceived:
+ foundRealTimeIssue = True
+ if foundAssertion and (msgLine < 3):
+ msgLine += 1
+ msgAssertion += str(line)
+ result = re.search('uci->stat', str(line))
+ if result is not None and not exitSignalReceived:
+ uciStatMsgCount += 1
+ result = re.search('PDCP data request failed', str(line))
+ if result is not None and not exitSignalReceived:
+ pdcpDataReqFailedCount += 1
+ result = re.search('bad DCI 1', str(line))
+ if result is not None and not exitSignalReceived:
+ badDciCount += 1
+ result = re.search('Format1A Retransmission but TBS are different', str(line))
+ if result is not None and not exitSignalReceived:
+ f1aRetransmissionCount += 1
+ result = re.search('FATAL ERROR', str(line))
+ if result is not None and not exitSignalReceived:
+ fatalErrorCount += 1
+ result = re.search('MAC BSR Triggered ReTxBSR Timer expiry', str(line))
+ if result is not None and not exitSignalReceived:
+ macBsrTimerExpiredCount += 1
+ result = re.search('Generating RRCConnectionReconfigurationComplete', str(line))
+ if result is not None:
+ rrcConnectionRecfgComplete += 1
+ # No cell synchronization found, abandoning
+ result = re.search('No cell synchronization found, abandoning', str(line))
+ if result is not None:
+ no_cell_sync_found = True
+ if RAN.eNBmbmsEnables[0]:
+ result = re.search('TRIED TO PUSH MBMS DATA', str(line))
+ if result is not None:
+ mbms_messages += 1
+ result = re.search("MIB Information => ([a-zA-Z]{1,10}), ([a-zA-Z]{1,10}), NidCell (?P'
+ msg = 'Doing X2 Handover w/ option ' + self.x2_ho_options
+ logging.debug(msg)
+ fullMessage += msg + '\n'
+ if self.x2_ho_options == 'network':
+ if RAN.flexranCtrlInstalled and RAN.flexranCtrlStarted:
+ self.x2ENBBsIds = []
+ self.x2ENBConnectedUEs = []
+ self.x2ENBBsIds.append([])
+ self.x2ENBBsIds.append([])
+ self.x2ENBConnectedUEs.append([])
+ self.x2ENBConnectedUEs.append([])
+ fullMessage += self.X2_Status(0, self.testCase_id + '_pre_ho.json')
+
+ msg = "Activating the X2 Net control on each eNB"
+ logging.debug(msg)
+ fullMessage += msg + '\n'
+ eNB_cnt = self.x2NbENBs
+ cnt = 0
+ while cnt < eNB_cnt:
+ cmd = "curl -XPOST http://" + EPC.IPAddress + ":9999/rrc/x2_ho_net_control/enb/" + str(self.x2ENBBsIds[0][cnt]) + "/1"
+ logging.debug(cmd)
+ fullMessage += cmd + '\n'
+ subprocess.run(cmd, shell=True)
+ cnt += 1
+ # Waiting for the activation to be active
+ time.sleep(10)
+ msg = "Switching UE(s) from eNB to eNB"
+ logging.debug(msg)
+ fullMessage += msg + '\n'
+ cnt = 0
+ while cnt < eNB_cnt:
+ ueIdx = 0
+ while ueIdx < len(self.x2ENBConnectedUEs[0][cnt]):
+ cmd = "curl -XPOST http://" + EPC.IPAddress() + ":9999/rrc/ho/senb/" + str(self.x2ENBBsIds[0][cnt]) + "/ue/" + str(self.x2ENBConnectedUEs[0][cnt][ueIdx]) + "/tenb/" + str(self.x2ENBBsIds[0][eNB_cnt - cnt - 1])
+ logging.debug(cmd)
+ fullMessage += cmd + '\n'
+ subprocess.run(cmd, shell=True)
+ ueIdx += 1
+ cnt += 1
+ time.sleep(10)
+ # check
+ logging.debug("Checking the Status after X2 Handover")
+ fullMessage += self.X2_Status(1, self.testCase_id + '_post_ho.json')
+ cnt = 0
+ x2Status = True
+ while cnt < eNB_cnt:
+ if len(self.x2ENBConnectedUEs[0][cnt]) == len(self.x2ENBConnectedUEs[1][cnt]):
+ x2Status = False
+ cnt += 1
+ if x2Status:
+ msg = "X2 Handover was successful"
+ logging.debug(msg)
+ fullMessage += msg + ''
+ html_queue.put(fullMessage)
+ HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
+ else:
+ msg = "X2 Handover FAILED"
+ logging.error(msg)
+ fullMessage += msg + ''
+ html_queue.put(fullMessage)
+ HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
+ else:
+ HTML.CreateHtmlTestRow('Cannot perform requested X2 Handover', 'KO', CONST.ALL_PROCESSES_OK)
+
+ def LogCollectBuild(self,RAN):
+ SSH = sshconnection.SSHConnection()
+ if (RAN.eNBIPAddress != '' and RAN.eNBUserName != '' and RAN.eNBPassword != ''):
+ IPAddress = RAN.eNBIPAddress
+ UserName = RAN.eNBUserName
+ Password = RAN.eNBPassword
+ SourceCodePath = RAN.eNBSourceCodePath
+ elif (self.UEIPAddress != '' and self.UEUserName != '' and self.UEPassword != ''):
+ IPAddress = self.UEIPAddress
+ UserName = self.UEUserName
+ Password = self.UEPassword
+ SourceCodePath = self.UESourceCodePath
+ else:
+ sys.exit('Insufficient Parameter')
+ SSH.open(IPAddress, UserName, Password)
+ SSH.command('cd ' + SourceCodePath, '\$', 5)
+ SSH.command('cd cmake_targets', '\$', 5)
+ SSH.command('rm -f build.log.zip', '\$', 5)
+ SSH.command('zip build.log.zip build_log_*/*', '\$', 60)
+ SSH.close()
+
+ def LogCollectPing(self,EPC):
+ SSH = sshconnection.SSHConnection()
+ SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password)
+ SSH.command('cd ' + EPC.SourceCodePath, '\$', 5)
+ SSH.command('cd scripts', '\$', 5)
+ SSH.command('rm -f ping.log.zip', '\$', 5)
+ SSH.command('zip ping.log.zip ping*.log', '\$', 60)
+ SSH.command('rm ping*.log', '\$', 5)
+ SSH.close()
+
+ def LogCollectIperf(self,EPC):
+ SSH = sshconnection.SSHConnection()
+ SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password)
+ SSH.command('cd ' + EPC.SourceCodePath, '\$', 5)
+ SSH.command('cd scripts', '\$', 5)
+ SSH.command('rm -f iperf.log.zip', '\$', 5)
+ SSH.command('zip iperf.log.zip iperf*.log', '\$', 60)
+ SSH.command('rm iperf*.log', '\$', 5)
+ SSH.close()
+
+ def LogCollectOAIUE(self):
+ SSH = sshconnection.SSHConnection()
+ SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
+ SSH.command('cd ' + self.UESourceCodePath, '\$', 5)
+ SSH.command('cd cmake_targets', '\$', 5)
+ SSH.command('echo ' + self.UEPassword + ' | sudo -S rm -f ue.log.zip', '\$', 5)
+ SSH.command('echo ' + self.UEPassword + ' | sudo -S zip ue.log.zip ue*.log core* ue_*record.raw ue_*.pcap ue_*txt', '\$', 60)
+ SSH.command('echo ' + self.UEPassword + ' | sudo -S rm ue*.log core* ue_*record.raw ue_*.pcap ue_*txt', '\$', 5)
+ SSH.close()
+
+ def RetrieveSystemVersion(self, machine,HTML,RAN):
+ if RAN.eNBIPAddress == 'none' or self.UEIPAddress == 'none':
+ HTML.OsVersion[0]='Ubuntu 16.04.5 LTS'
+ HTML.KernelVersion[0]='4.15.0-45-generic'
+ HTML.UhdVersion[0]='3.13.0.1-0'
+ HTML.UsrpBoard[0]='B210'
+ HTML.CpuNb[0]='4'
+ HTML.CpuModel[0]='Intel(R) Core(TM) i5-6200U'
+ HTML.CpuMHz[0]='2399.996 MHz'
+ return 0
+ if machine == 'eNB':
+ if RAN.eNBIPAddress != '' and RAN.eNBUserName != '' and RAN.eNBPassword != '':
+ IPAddress = RAN.eNBIPAddress
+ UserName = RAN.eNBUserName
+ Password = RAN.eNBPassword
+ idx = 0
+ else:
+ return -1
+ if machine == 'UE':
+ if self.UEIPAddress != '' and self.UEUserName != '' and self.UEPassword != '':
+ IPAddress = self.UEIPAddress
+ UserName = self.UEUserName
+ Password = self.UEPassword
+ idx = 1
+ else:
+ return -1
+
+ SSH = sshconnection.SSHConnection()
+ SSH.open(IPAddress, UserName, Password)
+ SSH.command('lsb_release -a', '\$', 5)
+ result = re.search('Description:\\\\t(?PCAT-M module Attachment Completed in ' + str(attach_cnt+4) + ' seconds' - if (nRSRQ is not None) and (nRSRP is not None): - html_cell += '\n RSRQ = ' + str(-20+(nRSRQ/2)) + ' dB' - html_cell += '\n RSRP = ' + str(-140+nRSRP) + ' dBm' - else: - html_cell += '' - html_queue.put(html_cell) - HTML.CreateHtmlTestRowQueue('N/A', 'OK', 1, html_queue) - else: - logging.error('\u001B[1m CAT-M module Attachment Failed\u001B[0m') - html_cell = '
CAT-M module Attachment Failed' - html_queue.put(html_cell) - HTML.CreateHtmlTestRowQueue('N/A', 'KO', 1, html_queue) - self.AutoTerminateUEandeNB() - - def PingCatM(self): - if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': - HELP.GenericHelp(CONST.Version) - sys.exit('Insufficient Parameter') - check_eNB = True - check_OAI_UE = False - pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE) - if (pStatus < 0): - HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) - self.AutoTerminateUEandeNB() - return - try: - statusQueue = SimpleQueue() - lock = Lock() - SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password) - SSH.command('cd ' + EPC.SourceCodePath, '\$', 5) - SSH.command('cd scripts', '\$', 5) - if re.match('OAI', EPC.Type, re.IGNORECASE): - logging.debug('Using the OAI EPC HSS: not implemented yet') - HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) - HTML.CreateHtmlTabFooter(False) - sys.exit(1) - else: - SSH.command('egrep --color=never "Allocated ipv4 addr" /opt/ltebox/var/log/xGwLog.0', '\$', 5) - result = re.search('Allocated ipv4 addr: (?P
CAT-M module\nIP Address : ' + moduleIPAddr + '\n' + qMsg + '' - statusQueue.put(html_cell) - if (packetLossOK): - HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', 1, statusQueue) - else: - HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', 1, statusQueue) - self.AutoTerminateUEandeNB() - except: - os.kill(os.getppid(),signal.SIGUSR1) - - def AttachUE_common(self, device_id, statusQueue, lock, idx): - try: - SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword) - if self.ADBCentralized: - #RH quick add on to integrate cots control defined by yaml - #if device Id exists in yaml dictionary, we execute the new procedure defined in cots_ue class - #otherwise we use the legacy procedure - if COTS_UE.Check_Exists(device_id): - #switch device to Airplane mode OFF (ie Radio ON) - COTS_UE.Set_Airplane(device_id, 'OFF') - elif device_id == '84B7N16418004022': - SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/on"', '\$', 60) - else: - SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/on', '\$', 60) - else: - # airplane mode off // radio on - SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOnCmd[idx], '\$', 60) - time.sleep(2) - max_count = 45 - count = max_count - while count > 0: - if self.ADBCentralized: - SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "dumpsys telephony.registry" | grep -m 1 mDataConnectionState', '\$', 15) - else: - SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "dumpsys telephony.registry"\' | grep -m 1 mDataConnectionState', '\$', 60) - result = re.search('mDataConnectionState.*=(?P
UE (' + device_id + ')\n' + message + ''
- else:
- html_cell = 'UE (' + device_id + ')\n' + message + ' in ' + str(count + 2) + ' seconds'
- html_queue.put(html_cell)
- if (attach_status):
- cnt = 0
- while cnt < len(self.UEDevices):
- if self.UEDevicesStatus[cnt] == CONST.UE_STATUS_ATTACHING:
- self.UEDevicesStatus[cnt] = CONST.UE_STATUS_ATTACHED
- cnt += 1
- HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
- result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
- if result is not None:
- logging.debug('Waiting 5 seconds to fill up record file')
- time.sleep(5)
- else:
- HTML.CreateHtmlTestRowQueue('N/A', 'KO', len(self.UEDevices), html_queue)
- self.AutoTerminateUEandeNB()
-
- def DetachUE_common(self, device_id, idx):
- try:
- SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
- if self.ADBCentralized:
- #RH quick add on to integrate cots control defined by yaml
- #if device id exists in yaml dictionary, we execute the new procedure defined in cots_ue class
- #otherwise we use the legacy procedure
- if COTS_UE.Check_Exists(device_id):
- #switch device to Airplane mode ON (ie Radio OFF)
- COTS_UE.Set_Airplane(device_id,'ON')
- elif device_id == '84B7N16418004022':
- SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/off"', '\$', 60)
- else:
- SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/off', '\$', 60)
- else:
- SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOffCmd[idx], '\$', 60)
- logging.debug('\u001B[1mUE (' + device_id + ') Detach Completed\u001B[0m')
- SSH.close()
- except:
- os.kill(os.getppid(),signal.SIGUSR1)
-
- def DetachUE(self):
- if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
- HELP.GenericHelp(CONST.Version)
- sys.exit('Insufficient Parameter')
- check_eNB = True
- check_OAI_UE = False
- pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
- if (pStatus < 0):
- HTML.CreateHtmlTestRow('N/A', 'KO', pStatus)
- self.AutoTerminateUEandeNB()
- return
- multi_jobs = []
- cnt = 0
- for device_id in self.UEDevices:
- self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHING
- p = Process(target = self.DetachUE_common, args = (device_id,cnt,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- cnt += 1
- for job in multi_jobs:
- job.join()
- HTML.CreateHtmlTestRow('N/A', 'OK', CONST.ALL_PROCESSES_OK)
- result = re.search('T_stdout', str(RAN.Initialize_eNB_args))
- if result is not None:
- logging.debug('Waiting 5 seconds to fill up record file')
- time.sleep(5)
- cnt = 0
- while cnt < len(self.UEDevices):
- self.UEDevicesStatus[cnt] = CONST.UE_STATUS_DETACHED
- cnt += 1
-
- def RebootUE_common(self, device_id):
- try:
- SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
- previousmDataConnectionStates = []
- # Save mDataConnectionState
- SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell dumpsys telephony.registry | grep mDataConnectionState', '\$', 15)
- SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell dumpsys telephony.registry | grep mDataConnectionState', '\$', 15)
- result = re.search('mDataConnectionState.*=(?PUE (' + device_id + ')\n' + message + ''
- html_queue.put(html_cell)
- if check_status and passStatus:
- HTML.CreateHtmlTestRowQueue(htmlOptions, 'OK', len(self.UEDevices), html_queue)
- else:
- HTML.CreateHtmlTestRowQueue(htmlOptions, 'KO', len(self.UEDevices), html_queue)
- self.AutoTerminateUEandeNB()
-
- def GetAllUEIPAddresses(self):
- if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
- HELP.GenericHelp(CONST.Version)
- sys.exit('Insufficient Parameter')
- ue_ip_status = 0
- self.UEIPAddresses = []
- if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'):
- if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '':
- HELP.GenericHelp(CONST.Version)
- sys.exit('Insufficient Parameter')
- SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
- SSH.command('ifconfig oaitun_ue1', '\$', 4)
- result = re.search('inet addr:(?POAI UE ping result\n' + qMsg + '' - html_queue.put(html_cell) - HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue) - - def PingNoS1(self): - check_eNB = True - check_OAI_UE = True - pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE) - if (pStatus < 0): - HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) - self.AutoTerminateUEandeNB() - return - ping_from_eNB = re.search('oaitun_enb1', str(self.ping_args)) - if ping_from_eNB is not None: - if RAN.eNBIPAddress == '' or RAN.eNBUserName == '' or RAN.eNBPassword == '': - HELP.GenericHelp(CONST.Version) - sys.exit('Insufficient Parameter') - else: - if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '': - HELP.GenericHelp(CONST.Version) - sys.exit('Insufficient Parameter') - try: - if ping_from_eNB is not None: - SSH.open(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword) - SSH.command('cd ' + RAN.eNBSourceCodePath + '/cmake_targets/', '\$', 5) - else: - SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword) - SSH.command('cd ' + self.UESourceCodePath + '/cmake_targets/', '\$', 5) - ping_time = re.findall("-c (\d+)",str(self.ping_args)) - ping_status = SSH.command('stdbuf -o0 ping ' + self.ping_args + ' 2>&1 | stdbuf -o0 tee ping_' + self.testCase_id + '.log', '\$', int(ping_time[0])*1.5) - # TIMEOUT CASE - if ping_status < 0: - message = 'Ping with OAI UE crashed due to TIMEOUT!' - logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m') - self.PingNoS1_wrong_exit(message) - return - result = re.search(', (?P
OAI UE ping result\n' + qMsg + '' - html_queue.put(html_cell) - if packetLossOK: - HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue) - else: - HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue) - - # copying on the EPC server for logCollection - if ping_from_eNB is not None: - copyin_res = SSH.copyin(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword, RAN.eNBSourceCodePath + '/cmake_targets/ping_' + self.testCase_id + '.log', '.') - else: - copyin_res = SSH.copyin(self.UEIPAddress, self.UEUserName, self.UEPassword, self.UESourceCodePath + '/cmake_targets/ping_' + self.testCase_id + '.log', '.') - if (copyin_res == 0): - SSH.copyout(EPC.IPAddress, EPC.UserName, EPC.Password, 'ping_' + self.testCase_id + '.log', EPC.SourceCodePath + '/scripts') - except: - os.kill(os.getppid(),signal.SIGUSR1) - - def Ping(self): - result = re.search('noS1', str(RAN.Initialize_eNB_args)) - if result is not None: - self.PingNoS1() - return - if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '': - HELP.GenericHelp(CONST.Version) - sys.exit('Insufficient Parameter') - check_eNB = True - if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'): - check_OAI_UE = True - else: - check_OAI_UE = False - pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE) - if (pStatus < 0): - HTML.CreateHtmlTestRow(self.ping_args, 'KO', pStatus) - self.AutoTerminateUEandeNB() - return - ueIpStatus = self.GetAllUEIPAddresses() - if (ueIpStatus < 0): - HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE) - self.AutoTerminateUEandeNB() - return - multi_jobs = [] - i = 0 - lock = Lock() - status_queue = SimpleQueue() - for UE_IPAddress in self.UEIPAddresses: - device_id = self.UEDevices[i] - p = Process(target = self.Ping_common, args = (lock,UE_IPAddress,device_id,status_queue,)) - p.daemon = True - p.start() - multi_jobs.append(p) - i = i + 1 - for job in multi_jobs: - job.join() - - if (status_queue.empty()): - HTML.CreateHtmlTestRow(self.ping_args, 'KO', CONST.ALL_PROCESSES_OK) - self.AutoTerminateUEandeNB() - else: - ping_status = True - html_queue = SimpleQueue() - while (not status_queue.empty()): - count = status_queue.get() - if (count < 0): - ping_status = False - device_id = status_queue.get() - ip_addr = status_queue.get() - message = status_queue.get() - html_cell = '
UE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
- html_queue.put(html_cell)
- if (ping_status):
- HTML.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue)
- else:
- HTML.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue)
- self.AutoTerminateUEandeNB()
-
- def Iperf_ComputeTime(self):
- result = re.search('-t (?PUE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
- html_queue.put(html_cell)
- if (iperf_noperf and iperf_status):
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'PERF NOT MET', len(self.UEDevices), html_queue)
- elif (iperf_status):
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', len(self.UEDevices), html_queue)
- else:
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', len(self.UEDevices), html_queue)
- self.AutoTerminateUEandeNB()
-
- def Iperf(self):
- result = re.search('noS1', str(RAN.Initialize_eNB_args))
- if result is not None:
- self.IperfNoS1()
- return
- if EPC.IPAddress == '' or EPC.UserName == '' or EPC.Password == '' or EPC.SourceCodePath == '' or self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
- HELP.GenericHelp(CONST.Version)
- sys.exit('Insufficient Parameter')
- check_eNB = True
- if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'):
- check_OAI_UE = True
- else:
- check_OAI_UE = False
- pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
- if (pStatus < 0):
- HTML.CreateHtmlTestRow(self.iperf_args, 'KO', pStatus)
- self.AutoTerminateUEandeNB()
- return
- ueIpStatus = self.GetAllUEIPAddresses()
- if (ueIpStatus < 0):
- logging.debug('going here')
- HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.UE_IP_ADDRESS_ISSUE)
- self.AutoTerminateUEandeNB()
- return
-
- self.dummyIperfVersion = '2.0.10'
- #cmd = 'iperf --version'
- #logging.debug(cmd + '\n')
- #iperfStdout = subprocess.check_output(cmd, shell=True, universal_newlines=True)
- #result = re.search('iperf version 2.0.5', str(iperfStdout.strip()))
- #if result is not None:
- # dummyIperfVersion = '2.0.5'
- #result = re.search('iperf version 2.0.10', str(iperfStdout.strip()))
- #if result is not None:
- # dummyIperfVersion = '2.0.10'
-
- multi_jobs = []
- i = 0
- ue_num = len(self.UEIPAddresses)
- lock = Lock()
- status_queue = SimpleQueue()
- for UE_IPAddress in self.UEIPAddresses:
- device_id = self.UEDevices[i]
- p = Process(target = self.Iperf_common, args = (lock,UE_IPAddress,device_id,i,ue_num,status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- i = i + 1
- for job in multi_jobs:
- job.join()
-
- if (status_queue.empty()):
- HTML.CreateHtmlTestRow(self.iperf_args, 'KO', CONST.ALL_PROCESSES_OK)
- self.AutoTerminateUEandeNB()
- else:
- iperf_status = True
- iperf_noperf = False
- html_queue = SimpleQueue()
- while (not status_queue.empty()):
- count = status_queue.get()
- if (count < 0):
- iperf_status = False
- if (count > 0):
- iperf_noperf = True
- device_id = status_queue.get()
- ip_addr = status_queue.get()
- message = status_queue.get()
- html_cell = 'UE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + ''
- html_queue.put(html_cell)
- if (iperf_noperf and iperf_status):
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'PERF NOT MET', len(self.UEDevices), html_queue)
- elif (iperf_status):
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'OK', len(self.UEDevices), html_queue)
- else:
- HTML.CreateHtmlTestRowQueue(self.iperf_args, 'KO', len(self.UEDevices), html_queue)
- self.AutoTerminateUEandeNB()
-
- def CheckProcessExist(self, check_eNB, check_OAI_UE):
- multi_jobs = []
- status_queue = SimpleQueue()
- # in noS1 config, no need to check status from EPC
- # in gNB also currently no need to check
- result = re.search('noS1|band78', str(RAN.Initialize_eNB_args))
- if result is None:
- p = Process(target = EPC.CheckHSSProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- p = Process(target = EPC.CheckMMEProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- p = Process(target = EPC.CheckSPGWProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- else:
- if (check_eNB == False) and (check_OAI_UE == False):
- return 0
- if check_eNB:
- p = Process(target = RAN.CheckeNBProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- if check_OAI_UE:
- p = Process(target = self.CheckOAIUEProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- for job in multi_jobs:
- job.join()
-
- if (status_queue.empty()):
- return -15
- else:
- result = 0
- while (not status_queue.empty()):
- status = status_queue.get()
- if (status < 0):
- result = status
- if result == CONST.ENB_PROCESS_FAILED:
- fileCheck = re.search('enb_', str(RAN.eNBLogFiles[0]))
- if fileCheck is not None:
- SSH.copyin(RAN.eNBIPAddress, RAN.eNBUserName, RAN.eNBPassword, RAN.eNBSourceCodePath + '/cmake_targets/' + RAN.eNBLogFiles[0], '.')
- logStatus = RAN.AnalyzeLogFile_eNB(RAN.eNBLogFiles[0])
- if logStatus < 0:
- result = logStatus
- RAN.eNBLogFiles[0]=''
- if RAN.flexranCtrlInstalled and RAN.flexranCtrlStarted:
- self.TerminateFlexranCtrl()
- return result
-
- def CheckOAIUEProcessExist(self, initialize_OAI_UE_flag):
- multi_jobs = []
- status_queue = SimpleQueue()
- if initialize_OAI_UE_flag == False:
- p = Process(target = self.CheckOAIUEProcess, args = (status_queue,))
- p.daemon = True
- p.start()
- multi_jobs.append(p)
- for job in multi_jobs:
- job.join()
-
- if (status_queue.empty()):
- return -15
- else:
- result = 0
- while (not status_queue.empty()):
- status = status_queue.get()
- if (status < 0):
- result = status
- if result == CONST.OAI_UE_PROCESS_FAILED:
- fileCheck = re.search('ue_', str(self.UELogFile))
- if fileCheck is not None:
- SSH.copyin(self.UEIPAddress, self.UEUserName, self.UEPassword, self.UESourceCodePath + '/cmake_targets/' + self.UELogFile, '.')
- logStatus = self.AnalyzeLogFile_UE(self.UELogFile)
- if logStatus < 0:
- result = logStatus
- return result
-
- def CheckOAIUEProcess(self, status_queue):
- try:
- SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
- SSH.command('stdbuf -o0 ps -aux | grep --color=never ' + self.air_interface + ' | grep -v grep', '\$', 5)
- result = re.search(self.air_interface, SSH.getBefore())
- if result is None:
- logging.debug('\u001B[1;37;41m OAI UE Process Not Found! \u001B[0m')
- status_queue.put(CONST.OAI_UE_PROCESS_FAILED)
- else:
- status_queue.put(CONST.OAI_UE_PROCESS_OK)
- SSH.close()
- except:
- os.kill(os.getppid(),signal.SIGUSR1)
-
-
- def AnalyzeLogFile_UE(self, UElogFile):
- if (not os.path.isfile('./' + UElogFile)):
- return -1
- ue_log_file = open('./' + UElogFile, 'r')
- exitSignalReceived = False
- foundAssertion = False
- msgAssertion = ''
- msgLine = 0
- foundSegFault = False
- foundRealTimeIssue = False
- uciStatMsgCount = 0
- pdcpDataReqFailedCount = 0
- badDciCount = 0
- f1aRetransmissionCount = 0
- fatalErrorCount = 0
- macBsrTimerExpiredCount = 0
- rrcConnectionRecfgComplete = 0
- no_cell_sync_found = False
- mib_found = False
- frequency_found = False
- plmn_found = False
- nrUEFlag = False
- nrDecodeMib = 0
- nrFoundDCI = 0
- nrCRCOK = 0
- mbms_messages = 0
- HTML.htmlUEFailureMsg=''
- global_status = CONST.ALL_PROCESSES_OK
- for line in ue_log_file.readlines():
- result = re.search('nr_synchro_time', str(line))
- if result is not None:
- nrUEFlag = True
- if nrUEFlag:
- result = re.search('decode mib', str(line))
- if result is not None:
- nrDecodeMib += 1
- result = re.search('found 1 DCIs', str(line))
- if result is not None:
- nrFoundDCI += 1
- result = re.search('CRC OK', str(line))
- if result is not None:
- nrCRCOK += 1
- result = re.search('Exiting OAI softmodem', str(line))
- if result is not None:
- exitSignalReceived = True
- result = re.search('System error|[Ss]egmentation [Ff]ault|======= Backtrace: =========|======= Memory map: ========', str(line))
- if result is not None and not exitSignalReceived:
- foundSegFault = True
- result = re.search('[Cc]ore [dD]ump', str(line))
- if result is not None and not exitSignalReceived:
- foundSegFault = True
- result = re.search('./lte-uesoftmodem', str(line))
- if result is not None and not exitSignalReceived:
- foundSegFault = True
- result = re.search('[Aa]ssertion', str(line))
- if result is not None and not exitSignalReceived:
- foundAssertion = True
- result = re.search('LLL', str(line))
- if result is not None and not exitSignalReceived:
- foundRealTimeIssue = True
- if foundAssertion and (msgLine < 3):
- msgLine += 1
- msgAssertion += str(line)
- result = re.search('uci->stat', str(line))
- if result is not None and not exitSignalReceived:
- uciStatMsgCount += 1
- result = re.search('PDCP data request failed', str(line))
- if result is not None and not exitSignalReceived:
- pdcpDataReqFailedCount += 1
- result = re.search('bad DCI 1', str(line))
- if result is not None and not exitSignalReceived:
- badDciCount += 1
- result = re.search('Format1A Retransmission but TBS are different', str(line))
- if result is not None and not exitSignalReceived:
- f1aRetransmissionCount += 1
- result = re.search('FATAL ERROR', str(line))
- if result is not None and not exitSignalReceived:
- fatalErrorCount += 1
- result = re.search('MAC BSR Triggered ReTxBSR Timer expiry', str(line))
- if result is not None and not exitSignalReceived:
- macBsrTimerExpiredCount += 1
- result = re.search('Generating RRCConnectionReconfigurationComplete', str(line))
- if result is not None:
- rrcConnectionRecfgComplete += 1
- # No cell synchronization found, abandoning
- result = re.search('No cell synchronization found, abandoning', str(line))
- if result is not None:
- no_cell_sync_found = True
- if RAN.eNBmbmsEnables[0]:
- result = re.search('TRIED TO PUSH MBMS DATA', str(line))
- if result is not None:
- mbms_messages += 1
- result = re.search("MIB Information => ([a-zA-Z]{1,10}), ([a-zA-Z]{1,10}), NidCell (?P'
- msg = 'Doing X2 Handover w/ option ' + self.x2_ho_options
- logging.debug(msg)
- fullMessage += msg + '\n'
- if self.x2_ho_options == 'network':
- if RAN.flexranCtrlInstalled and RAN.flexranCtrlStarted:
- self.x2ENBBsIds = []
- self.x2ENBConnectedUEs = []
- self.x2ENBBsIds.append([])
- self.x2ENBBsIds.append([])
- self.x2ENBConnectedUEs.append([])
- self.x2ENBConnectedUEs.append([])
- fullMessage += self.X2_Status(0, self.testCase_id + '_pre_ho.json')
-
- msg = "Activating the X2 Net control on each eNB"
- logging.debug(msg)
- fullMessage += msg + '\n'
- eNB_cnt = self.x2NbENBs
- cnt = 0
- while cnt < eNB_cnt:
- cmd = "curl -XPOST http://" + EPC.IPAddress + ":9999/rrc/x2_ho_net_control/enb/" + str(self.x2ENBBsIds[0][cnt]) + "/1"
- logging.debug(cmd)
- fullMessage += cmd + '\n'
- subprocess.run(cmd, shell=True)
- cnt += 1
- # Waiting for the activation to be active
- time.sleep(10)
- msg = "Switching UE(s) from eNB to eNB"
- logging.debug(msg)
- fullMessage += msg + '\n'
- cnt = 0
- while cnt < eNB_cnt:
- ueIdx = 0
- while ueIdx < len(self.x2ENBConnectedUEs[0][cnt]):
- cmd = "curl -XPOST http://" + EPC.IPAddress() + ":9999/rrc/ho/senb/" + str(self.x2ENBBsIds[0][cnt]) + "/ue/" + str(self.x2ENBConnectedUEs[0][cnt][ueIdx]) + "/tenb/" + str(self.x2ENBBsIds[0][eNB_cnt - cnt - 1])
- logging.debug(cmd)
- fullMessage += cmd + '\n'
- subprocess.run(cmd, shell=True)
- ueIdx += 1
- cnt += 1
- time.sleep(10)
- # check
- logging.debug("Checking the Status after X2 Handover")
- fullMessage += self.X2_Status(1, self.testCase_id + '_post_ho.json')
- cnt = 0
- x2Status = True
- while cnt < eNB_cnt:
- if len(self.x2ENBConnectedUEs[0][cnt]) == len(self.x2ENBConnectedUEs[1][cnt]):
- x2Status = False
- cnt += 1
- if x2Status:
- msg = "X2 Handover was successful"
- logging.debug(msg)
- fullMessage += msg + ''
- html_queue.put(fullMessage)
- HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
- else:
- msg = "X2 Handover FAILED"
- logging.error(msg)
- fullMessage += msg + ''
- html_queue.put(fullMessage)
- HTML.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
- else:
- HTML.CreateHtmlTestRow('Cannot perform requested X2 Handover', 'KO', CONST.ALL_PROCESSES_OK)
-
- def LogCollectBuild(self):
- if (RAN.eNBIPAddress != '' and RAN.eNBUserName != '' and RAN.eNBPassword != ''):
- IPAddress = RAN.eNBIPAddress
- UserName = RAN.eNBUserName
- Password = RAN.eNBPassword
- SourceCodePath = RAN.eNBSourceCodePath
- elif (self.UEIPAddress != '' and self.UEUserName != '' and self.UEPassword != ''):
- IPAddress = self.UEIPAddress
- UserName = self.UEUserName
- Password = self.UEPassword
- SourceCodePath = self.UESourceCodePath
- else:
- sys.exit('Insufficient Parameter')
- SSH.open(IPAddress, UserName, Password)
- SSH.command('cd ' + SourceCodePath, '\$', 5)
- SSH.command('cd cmake_targets', '\$', 5)
- SSH.command('rm -f build.log.zip', '\$', 5)
- SSH.command('zip build.log.zip build_log_*/*', '\$', 60)
- SSH.close()
- def LogCollectPing(self):
- SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password)
- SSH.command('cd ' + EPC.SourceCodePath, '\$', 5)
- SSH.command('cd scripts', '\$', 5)
- SSH.command('rm -f ping.log.zip', '\$', 5)
- SSH.command('zip ping.log.zip ping*.log', '\$', 60)
- SSH.command('rm ping*.log', '\$', 5)
- SSH.close()
-
- def LogCollectIperf(self):
- SSH.open(EPC.IPAddress, EPC.UserName, EPC.Password)
- SSH.command('cd ' + EPC.SourceCodePath, '\$', 5)
- SSH.command('cd scripts', '\$', 5)
- SSH.command('rm -f iperf.log.zip', '\$', 5)
- SSH.command('zip iperf.log.zip iperf*.log', '\$', 60)
- SSH.command('rm iperf*.log', '\$', 5)
- SSH.close()
-
- def LogCollectOAIUE(self):
- SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
- SSH.command('cd ' + self.UESourceCodePath, '\$', 5)
- SSH.command('cd cmake_targets', '\$', 5)
- SSH.command('echo ' + self.UEPassword + ' | sudo -S rm -f ue.log.zip', '\$', 5)
- SSH.command('echo ' + self.UEPassword + ' | sudo -S zip ue.log.zip ue*.log core* ue_*record.raw ue_*.pcap ue_*txt', '\$', 60)
- SSH.command('echo ' + self.UEPassword + ' | sudo -S rm ue*.log core* ue_*record.raw ue_*.pcap ue_*txt', '\$', 5)
- SSH.close()
-
- def RetrieveSystemVersion(self, machine):
- if RAN.eNBIPAddress == 'none' or self.UEIPAddress == 'none':
- HTML.OsVersion[0]='Ubuntu 16.04.5 LTS'
- HTML.KernelVersion[0]='4.15.0-45-generic'
- HTML.UhdVersion[0]='3.13.0.1-0'
- HTML.UsrpBoard[0]='B210'
- HTML.CpuNb[0]='4'
- HTML.CpuModel[0]='Intel(R) Core(TM) i5-6200U'
- HTML.CpuMHz[0]='2399.996 MHz'
- return 0
- if machine == 'eNB':
- if RAN.eNBIPAddress != '' and RAN.eNBUserName != '' and RAN.eNBPassword != '':
- IPAddress = RAN.eNBIPAddress
- UserName = RAN.eNBUserName
- Password = RAN.eNBPassword
- idx = 0
- else:
- return -1
- if machine == 'UE':
- if self.UEIPAddress != '' and self.UEUserName != '' and self.UEPassword != '':
- IPAddress = self.UEIPAddress
- UserName = self.UEUserName
- Password = self.UEPassword
- idx = 1
- else:
- return -1
-
- SSH.open(IPAddress, UserName, Password)
- SSH.command('lsb_release -a', '\$', 5)
- result = re.search('Description:\\\\t(?P" >> ./test_simulator_results.html - if [ $GNB_RECEIVED -gt 0 ] - then - echo "- gNB --> RA received" >> ./test_simulator_results.html - else - echo "- gNB RA NOT RECEIVED" >> ./test_simulator_results.html - fi - if [ $GNB_CONNECTED -gt 0 ] - then - echo "- gNB --> 5G connected" >> ./test_simulator_results.html - else - echo "- gNB NOT 5G CONNECTED" >> ./test_simulator_results.html - fi - if [ $UE_RA_PROC_OK -gt 0 ] - then - echo "- NR UE --> RA procedure succeded" >> ./test_simulator_results.html - else - echo "- NR UE RA procedure failed" >> ./test_simulator_results.html - fi - echo "
" >> ./test_simulator_results.html +# if [ $GNB_RECEIVED -gt 0 ] +# then +# echo "- gNB --> RA received" >> ./test_simulator_results.html +# else +# echo "- gNB RA NOT RECEIVED" >> ./test_simulator_results.html +# fi +# if [ $GNB_CONNECTED -gt 0 ] +# then +# echo "- gNB --> 5G connected" >> ./test_simulator_results.html +# else +# echo "- gNB NOT 5G CONNECTED" >> ./test_simulator_results.html +# fi +# if [ $UE_RA_PROC_OK -gt 0 ] +# then +# echo "- NR UE --> RA procedure succeded" >> ./test_simulator_results.html +# else +# echo "- NR UE RA procedure failed" >> ./test_simulator_results.html +# fi +# echo "