Newer
Older
if autoAttachState is not None:
if autoAttachState == 0:
Gabriele Perrone
committed
SSH.command('AT^AUTOATT=1', 'OK', 5)
logging.debug('\u001B[1m Auto-Attach enabled\u001B[0m')
else:
logging.debug('\u001B[1;37;41m Could not check Auto-Attach! \u001B[0m')
# Force closure of picocom but device might still be locked
Gabriele Perrone
committed
SSH.close()
self.picocom_closure = False
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
self.checkDevTTYisUnlocked()
def TerminateCatM(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
self.picocom_closure = True
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
# dummy call to start a sudo session. The picocom command does NOT handle well the `sudo -S`
Gabriele Perrone
committed
SSH.command('echo ' + self.ADBPassword + ' | sudo -S ls', '\$', 10)
SSH.command('sudo picocom --baud 921600 --flow n --databits 8 /dev/ttyUSB0', 'Terminal ready', 10)
time.sleep(1)
# Calling twice AT to clear all buffers
Gabriele Perrone
committed
SSH.command('AT', 'OK|ERROR', 5)
SSH.command('AT', 'OK', 5)
Gabriele Perrone
committed
SSH.command('AT+CFUN=0', 'OK', 5)
logging.debug('\u001B[1m Cellular Functionality disabled\u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.picocom_closure = False
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
self.checkDevTTYisUnlocked()
def AttachCatM(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
self.picocom_closure = True
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
# dummy call to start a sudo session. The picocom command does NOT handle well the `sudo -S`
Gabriele Perrone
committed
SSH.command('echo ' + self.ADBPassword + ' | sudo -S ls', '\$', 10)
SSH.command('sudo picocom --baud 921600 --flow n --databits 8 /dev/ttyUSB0', 'Terminal ready', 10)
time.sleep(1)
# Calling twice AT to clear all buffers
Gabriele Perrone
committed
SSH.command('AT', 'OK|ERROR', 5)
SSH.command('AT', 'OK', 5)
Gabriele Perrone
committed
SSH.command('AT+CFUN=1', 'SIMSTORE,READY', 5)
logging.debug('\u001B[1m Cellular Functionality enabled\u001B[0m')
time.sleep(4)
# We should check if we register
count = 0
attach_cnt = 0
attach_status = False
while count < 5:
Gabriele Perrone
committed
SSH.command('AT+CEREG?', 'OK', 5)
result = re.search('CEREG: 2,(?P<state>[0-9\-]+),', SSH.getBefore())
if result is not None:
mDataConnectionState = int(result.group('state'))
if mDataConnectionState is not None:
if mDataConnectionState == 1:
count = 10
attach_status = True
result = re.search('CEREG: 2,1,"(?P<networky>[0-9A-Z]+)","(?P<networkz>[0-9A-Z]+)"', SSH.getBefore())
if result is not None:
networky = result.group('networky')
networkz = result.group('networkz')
logging.debug('\u001B[1m CAT-M module attached to eNB (' + str(networky) + '/' + str(networkz) + ')\u001B[0m')
else:
logging.debug('\u001B[1m CAT-M module attached to eNB\u001B[0m')
else:
logging.debug('+CEREG: 2,' + str(mDataConnectionState))
attach_cnt = attach_cnt + 1
logging.debug(SSH.getBefore())
count = count + 1
time.sleep(1)
Gabriele Perrone
committed
SSH.command('AT+CESQ', 'OK', 5)
result = re.search('CESQ: 99,99,255,255,(?P<rsrq>[0-9]+),(?P<rsrp>[0-9]+)', SSH.getBefore())
if result is not None:
nRSRQ = int(result.group('rsrq'))
nRSRP = int(result.group('rsrp'))
if (nRSRQ is not None) and (nRSRP is not None):
logging.debug(' RSRQ = ' + str(-20+(nRSRQ/2)) + ' dB')
logging.debug(' RSRP = ' + str(-140+nRSRP) + ' dBm')
Gabriele Perrone
committed
SSH.close()
self.picocom_closure = False
self.checkDevTTYisUnlocked()
html_cell = '<pre style="background-color:white">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</pre>'
else:
html_cell += '</pre>'
html_queue.put(html_cell)
self.CreateHtmlTestRowQueue('N/A', 'OK', 1, html_queue)
else:
logging.error('\u001B[1m CAT-M module Attachment Failed\u001B[0m')
html_cell = '<pre style="background-color:white">CAT-M module Attachment Failed</pre>'
html_queue.put(html_cell)
self.CreateHtmlTestRowQueue('N/A', 'KO', 1, html_queue)
if EPC.EPCIPAddress == '' or EPC.EPCUserName == '' or EPC.EPCPassword == '' or EPC.EPCSourceCodePath == '':
Usage()
sys.exit('Insufficient Parameter')
check_eNB = True
check_OAI_UE = False
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
if (pStatus < 0):
self.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
try:
statusQueue = SimpleQueue()
lock = Lock()
SSH.open(EPC.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword)
SSH.command('cd ' + EPC.EPCSourceCodePath, '\$', 5)
Gabriele Perrone
committed
SSH.command('cd scripts', '\$', 5)
if re.match('OAI', EPC.EPCType, re.IGNORECASE):
logging.debug('Using the OAI EPC HSS: not implemented yet')
self.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.CreateHtmlTabFooter(False)
sys.exit(1)
else:
Gabriele Perrone
committed
SSH.command('egrep --color=never "Allocated ipv4 addr" /opt/ltebox/var/log/xGwLog.0', '\$', 5)
result = re.search('Allocated ipv4 addr: (?P<ipaddr>[0-9\.]+) from Pool', SSH.getBefore())
if result is not None:
moduleIPAddr = result.group('ipaddr')
else:
self.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.AutoTerminateUEandeNB()
return
ping_time = re.findall("-c (\d+)",str(self.ping_args))
device_id = 'catm'
Gabriele Perrone
committed
ping_status = SSH.command('stdbuf -o0 ping ' + self.ping_args + ' ' + str(moduleIPAddr) + ' 2>&1 | stdbuf -o0 tee ping_' + self.testCase_id + '_' + device_id + '.log', '\$', int(ping_time[0])*1.5)
# TIMEOUT CASE
if ping_status < 0:
message = 'Ping with UE (' + str(moduleIPAddr) + ') crashed due to TIMEOUT!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, moduleIPAddr, device_id, statusQueue, message)
return
result = re.search(', (?P<packetloss>[0-9\.]+)% packet loss, time [0-9\.]+ms', SSH.getBefore())
if result is None:
message = 'Packet Loss Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, moduleIPAddr, device_id, statusQueue, message)
return
packetloss = result.group('packetloss')
if float(packetloss) == 100:
message = 'Packet Loss is 100%'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, moduleIPAddr, device_id, statusQueue, message)
return
result = re.search('rtt min\/avg\/max\/mdev = (?P<rtt_min>[0-9\.]+)\/(?P<rtt_avg>[0-9\.]+)\/(?P<rtt_max>[0-9\.]+)\/[0-9\.]+ ms', SSH.getBefore())
if result is None:
message = 'Ping RTT_Min RTT_Avg RTT_Max Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
self.ping_iperf_wrong_exit(lock, moduleIPAddr, device_id, statusQueue, message)
return
rtt_min = result.group('rtt_min')
rtt_avg = result.group('rtt_avg')
rtt_max = result.group('rtt_max')
pal_msg = 'Packet Loss : ' + packetloss + '%'
min_msg = 'RTT(Min) : ' + rtt_min + ' ms'
avg_msg = 'RTT(Avg) : ' + rtt_avg + ' ms'
max_msg = 'RTT(Max) : ' + rtt_max + ' ms'
lock.acquire()
logging.debug('\u001B[1;37;44m ping result (' + moduleIPAddr + ') \u001B[0m')
logging.debug('\u001B[1;34m ' + pal_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + min_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + avg_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + max_msg + '\u001B[0m')
qMsg = pal_msg + '\n' + min_msg + '\n' + avg_msg + '\n' + max_msg
packetLossOK = True
if packetloss is not None:
if float(packetloss) > float(self.ping_packetloss_threshold):
qMsg += '\nPacket Loss too high'
logging.debug('\u001B[1;37;41m Packet Loss too high \u001B[0m')
packetLossOK = False
elif float(packetloss) > 0:
qMsg += '\nPacket Loss is not 0%'
logging.debug('\u001B[1;30;43m Packet Loss is not 0% \u001B[0m')
lock.release()
Gabriele Perrone
committed
SSH.close()
html_cell = '<pre style="background-color:white">CAT-M module\nIP Address : ' + moduleIPAddr + '\n' + qMsg + '</pre>'
statusQueue.put(html_cell)
if (packetLossOK):
self.CreateHtmlTestRowQueue(self.ping_args, 'OK', 1, statusQueue)
else:
self.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):
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
if self.ADBCentralized:
if device_id == '84B7N16418004022':
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/on"', '\$', 60)
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/on', '\$', 60)
# airplane mode off // radio on
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOnCmd[idx], '\$', 60)
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "dumpsys telephony.registry" | grep -m 1 mDataConnectionState', '\$', 15)
Gabriele Perrone
committed
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<state>[0-9\-]+)', SSH.getBefore())
if result is None:
logging.debug('\u001B[1;37;41m mDataConnectionState Not Found! \u001B[0m')
lock.acquire()
statusQueue.put(-1)
statusQueue.put(device_id)
statusQueue.put('mDataConnectionState Not Found!')
lock.release()
break
mDataConnectionState = int(result.group('state'))
if mDataConnectionState == 2:
logging.debug('\u001B[1mUE (' + device_id + ') Attach Completed\u001B[0m')
lock.acquire()
statusQueue.put(max_count - count)
statusQueue.put(device_id)
statusQueue.put('Attach Completed')
lock.release()
break
count = count - 1
if count == 15 or count == 30:
logging.debug('\u001B[1;30;43m Retry UE (' + device_id + ') Flight Mode Off \u001B[0m')
if self.ADBCentralized:
if device_id == '84B7N16418004022':
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/off"', '\$', 60)
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/off', '\$', 60)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOffCmd[idx], '\$', 60)
if self.ADBCentralized:
if device_id == '84B7N16418004022':
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/on"', '\$', 60)
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/on', '\$', 60)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOnCmd[idx], '\$', 60)
logging.debug('\u001B[1mWait UE (' + device_id + ') a second until mDataConnectionState=2 (' + str(max_count-count) + ' times)\u001B[0m')
time.sleep(1)
if count == 0:
logging.debug('\u001B[1;37;41m UE (' + device_id + ') Attach Failed \u001B[0m')
lock.acquire()
statusQueue.put(-1)
statusQueue.put(device_id)
statusQueue.put('Attach Failed')
lock.release()
Gabriele Perrone
committed
SSH.close()
except:
os.kill(os.getppid(),signal.SIGUSR1)
def AttachUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
check_eNB = True
check_OAI_UE = False
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
if (pStatus < 0):
self.CreateHtmlTestRow('N/A', 'KO', pStatus)
for device_id in self.UEDevices:
if (self.nbMaxUEtoAttach == -1) or (nb_ue_to_connect < self.nbMaxUEtoAttach):
self.UEDevicesStatus[nb_ue_to_connect] = UE_STATUS_ATTACHING
p = Process(target = self.AttachUE_common, args = (device_id, status_queue, lock,nb_ue_to_connect,))
p.daemon = True
p.start()
multi_jobs.append(p)
nb_ue_to_connect = nb_ue_to_connect + 1
for job in multi_jobs:
job.join()
self.CreateHtmlTestRow('N/A', 'KO', ALL_PROCESSES_OK)
else:
attach_status = True
html_queue = SimpleQueue()
while (not status_queue.empty()):
count = status_queue.get()
if (count < 0):
attach_status = False
device_id = status_queue.get()
message = status_queue.get()
if (count < 0):
html_cell = '<pre style="background-color:white">UE (' + device_id + ')\n' + message + '</pre>'
html_cell = '<pre style="background-color:white">UE (' + device_id + ')\n' + message + ' in ' + str(count + 2) + ' seconds</pre>'
cnt = 0
while cnt < len(self.UEDevices):
if self.UEDevicesStatus[cnt] == UE_STATUS_ATTACHING:
self.UEDevicesStatus[cnt] = UE_STATUS_ATTACHED
cnt += 1
self.CreateHtmlTestRowQueue('N/A', 'OK', len(self.UEDevices), html_queue)
result = re.search('T_stdout', str(self.Initialize_eNB_args))
if result is not None:
logging.debug('Waiting 5 seconds to fill up record file')
time.sleep(5)
else:
self.CreateHtmlTestRowQueue('N/A', 'KO', len(self.UEDevices), html_queue)
def DetachUE_common(self, device_id, idx):
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
if self.ADBCentralized:
if device_id == '84B7N16418004022':
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "su - root -c /data/local/tmp/off"', '\$', 60)
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell /data/local/tmp/off', '\$', 60)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' ' + self.UEDevicesOffCmd[idx], '\$', 60)
logging.debug('\u001B[1mUE (' + device_id + ') Detach Completed\u001B[0m')
Gabriele Perrone
committed
SSH.close()
except:
os.kill(os.getppid(),signal.SIGUSR1)
def DetachUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
if (pStatus < 0):
self.CreateHtmlTestRow('N/A', 'KO', pStatus)
for device_id in self.UEDevices:
self.UEDevicesStatus[cnt] = UE_STATUS_DETACHING
p = Process(target = self.DetachUE_common, args = (device_id,cnt,))
p.daemon = True
p.start()
multi_jobs.append(p)
for job in multi_jobs:
job.join()
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
result = re.search('T_stdout', str(self.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] = UE_STATUS_DETACHED
cnt += 1
def RebootUE_common(self, device_id):
try:
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
previousmDataConnectionStates = []
# Save mDataConnectionState
Gabriele Perrone
committed
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.*=(?P<state>[0-9\-]+)', SSH.getBefore())
if result is None:
logging.debug('\u001B[1;37;41m mDataConnectionState Not Found! \u001B[0m')
sys.exit(1)
previousmDataConnectionStates.append(int(result.group('state')))
# Reboot UE
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell reboot', '\$', 10)
time.sleep(60)
previousmDataConnectionState = previousmDataConnectionStates.pop(0)
count = 180
while count > 0:
count = count - 1
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell dumpsys telephony.registry | grep mDataConnectionState', '\$', 15)
result = re.search('mDataConnectionState.*=(?P<state>[0-9\-]+)', SSH.getBefore())
if result is None:
mDataConnectionState = None
else:
mDataConnectionState = int(result.group('state'))
logging.debug('mDataConnectionState = ' + result.group('state'))
if mDataConnectionState is None or (previousmDataConnectionState == 2 and mDataConnectionState != 2):
logging.debug('\u001B[1mWait UE (' + device_id + ') a second until reboot completion (' + str(180-count) + ' times)\u001B[0m')
time.sleep(1)
else:
logging.debug('\u001B[1mUE (' + device_id + ') Reboot Completed\u001B[0m')
break
if count == 0:
logging.debug('\u001B[1;37;41m UE (' + device_id + ') Reboot Failed \u001B[0m')
sys.exit(1)
Gabriele Perrone
committed
SSH.close()
except:
os.kill(os.getppid(),signal.SIGUSR1)
def RebootUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)
if (pStatus < 0):
self.CreateHtmlTestRow('N/A', 'KO', pStatus)
self.CreateHtmlTabFooter(False)
multi_jobs = []
for device_id in self.UEDevices:
p = Process(target = self.RebootUE_common, args = (device_id,))
p.daemon = True
p.start()
multi_jobs.append(p)
for job in multi_jobs:
job.join()
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
def DataDisableUE_common(self, device_id, idx):

Raphael Defosseux
committed
try:
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
# disable data service
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "svc data disable"', '\$', 60)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "svc data disable"\'', '\$', 60)

Raphael Defosseux
committed
logging.debug('\u001B[1mUE (' + device_id + ') Disabled Data Service\u001B[0m')
Gabriele Perrone
committed
SSH.close()

Raphael Defosseux
committed
except:
os.kill(os.getppid(),signal.SIGUSR1)
def DataDisableUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
multi_jobs = []

Raphael Defosseux
committed
for device_id in self.UEDevices:
p = Process(target = self.DataDisableUE_common, args = (device_id,i,))

Raphael Defosseux
committed
p.daemon = True
p.start()
multi_jobs.append(p)

Raphael Defosseux
committed
for job in multi_jobs:
job.join()
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
def DataEnableUE_common(self, device_id, idx):

Raphael Defosseux
committed
try:
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)

Raphael Defosseux
committed
# enable data service
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "svc data enable"', '\$', 60)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "svc data enable"\'', '\$', 60)

Raphael Defosseux
committed
logging.debug('\u001B[1mUE (' + device_id + ') Enabled Data Service\u001B[0m')
Gabriele Perrone
committed
SSH.close()

Raphael Defosseux
committed
except:
os.kill(os.getppid(),signal.SIGUSR1)
def DataEnableUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
multi_jobs = []

Raphael Defosseux
committed
for device_id in self.UEDevices:
p = Process(target = self.DataEnableUE_common, args = (device_id,i,))

Raphael Defosseux
committed
p.daemon = True
p.start()
multi_jobs.append(p)

Raphael Defosseux
committed
for job in multi_jobs:
job.join()
self.CreateHtmlTestRow('N/A', 'OK', ALL_PROCESSES_OK)
def GetAllUEDevices(self, terminate_ue_flag):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('adb devices', '\$', 15)
#self.UEDevices = re.findall("\\\\r\\\\n([A-Za-z0-9]+)\\\\tdevice",SSH.getBefore())
self.UEDevices = re.findall("\\\\r\\\\n([A-Za-z0-9]+)\\\\tdevice",SSH.getBefore())
Gabriele Perrone
committed
SSH.close()
else:
if (os.path.isfile('./phones_list.txt')):
os.remove('./phones_list.txt')
Gabriele Perrone
committed
SSH.command('ls /etc/*/phones*.txt', '\$', 5)
result = re.search('/etc/ci/phones_list.txt', SSH.getBefore())
Gabriele Perrone
committed
SSH.close()
if (result is not None) and (len(self.UEDevices) == 0):
Gabriele Perrone
committed
SSH.copyin(self.ADBIPAddress, self.ADBUserName, self.ADBPassword, '/etc/ci/phones_list.txt', '.')
if (os.path.isfile('./phones_list.txt')):
phone_list_file = open('./phones_list.txt', 'r')
for line in phone_list_file.readlines():
line = line.strip()
result = re.search('^#', line)
if result is not None:
continue
comma_split = line.split(",")
self.UEDevices.append(comma_split[0])
self.UEDevicesRemoteServer.append(comma_split[1])
self.UEDevicesRemoteUser.append(comma_split[2])
self.UEDevicesOffCmd.append(comma_split[3])
self.UEDevicesOnCmd.append(comma_split[4])
self.UEDevicesRebootCmd.append(comma_split[5])
phone_list_file.close()
if terminate_ue_flag == False:
if len(self.UEDevices) == 0:
logging.debug('\u001B[1;37;41m UE Not Found! \u001B[0m')
sys.exit(1)
if len(self.UEDevicesStatus) == 0:
cnt = 0
while cnt < len(self.UEDevices):
self.UEDevicesStatus.append(UE_STATUS_DETACHED)
cnt += 1
def GetAllCatMDevices(self, terminate_ue_flag):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('lsusb | egrep "Future Technology Devices International, Ltd FT2232C" | sed -e "s#:.*##" -e "s# #_#g"', '\$', 15)
#self.CatMDevices = re.findall("\\\\r\\\\n([A-Za-z0-9_]+)",SSH.getBefore())
self.CatMDevices = re.findall("\\\\r\\\\n([A-Za-z0-9_]+)",SSH.getBefore())
else:
if (os.path.isfile('./modules_list.txt')):
os.remove('./modules_list.txt')
Gabriele Perrone
committed
SSH.command('ls /etc/*/modules*.txt', '\$', 5)
result = re.search('/etc/ci/modules_list.txt', SSH.getBefore())
Gabriele Perrone
committed
SSH.close()
if result is not None:
logging.debug('Found a module list file on ADB server')
if terminate_ue_flag == False:
if len(self.CatMDevices) == 0:
logging.debug('\u001B[1;37;41m CAT-M UE Not Found! \u001B[0m')
sys.exit(1)
Gabriele Perrone
committed
SSH.close()
def CheckUEStatus_common(self, lock, device_id, statusQueue, idx):

Raphael Defosseux
committed
try:
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "dumpsys telephony.registry"', '\$', 15)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "dumpsys telephony.registry"\'', '\$', 60)
result = re.search('mServiceState=(?P<serviceState>[0-9]+)', SSH.getBefore())
serviceState = 'Service State: UNKNOWN'
if result is not None:
lServiceState = int(result.group('serviceState'))
if lServiceState == 3:
serviceState = 'Service State: RADIO_POWERED_OFF'
if lServiceState == 1:
serviceState = 'Service State: OUT_OF_SERVICE'
if lServiceState == 0:
serviceState = 'Service State: IN_SERVICE'
if lServiceState == 2:
serviceState = 'Service State: EMERGENCY_ONLY'
result = re.search('mDataConnectionState=(?P<dataConnectionState>[0-9]+)', SSH.getBefore())
dataConnectionState = 'Data State: UNKNOWN'
if result is not None:
lDataConnectionState = int(result.group('dataConnectionState'))
if lDataConnectionState == 0:
dataConnectionState = 'Data State: DISCONNECTED'
if lDataConnectionState == 1:
dataConnectionState = 'Data State: CONNECTING'
if lDataConnectionState == 2:
dataConnectionState = 'Data State: CONNECTED'
if lDataConnectionState == 3:
dataConnectionState = 'Data State: SUSPENDED'
result = re.search('mDataConnectionReason=(?P<dataConnectionReason>[0-9a-zA-Z_]+)', SSH.getBefore())
dataConnectionReason = 'Data Reason: UNKNOWN'
if result is not None:
dataConnectionReason = 'Data Reason: ' + result.group('dataConnectionReason')

Raphael Defosseux
committed
lock.acquire()
logging.debug('\u001B[1;37;44m Status Check (' + str(device_id) + ') \u001B[0m')
logging.debug('\u001B[1;34m ' + serviceState + '\u001B[0m')
logging.debug('\u001B[1;34m ' + dataConnectionState + '\u001B[0m')
logging.debug('\u001B[1;34m ' + dataConnectionReason + '\u001B[0m')

Raphael Defosseux
committed
statusQueue.put(0)
statusQueue.put(device_id)
qMsg = serviceState + '\n' + dataConnectionState + '\n' + dataConnectionReason
statusQueue.put(qMsg)

Raphael Defosseux
committed
lock.release()
Gabriele Perrone
committed
SSH.close()

Raphael Defosseux
committed
except:
os.kill(os.getppid(),signal.SIGUSR1)
def CheckStatusUE(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
check_eNB = True
check_OAI_UE = False
pStatus = self.CheckProcessExist(check_eNB, check_OAI_UE)

Raphael Defosseux
committed
if (pStatus < 0):
self.CreateHtmlTestRow('N/A', 'KO', pStatus)
self.CreateHtmlTabFooter(False)
sys.exit(1)
multi_jobs = []
lock = Lock()
status_queue = SimpleQueue()

Raphael Defosseux
committed
for device_id in self.UEDevices:
p = Process(target = self.CheckUEStatus_common, args = (lock,device_id,status_queue,i,))

Raphael Defosseux
committed
p.daemon = True
p.start()
multi_jobs.append(p)

Raphael Defosseux
committed
for job in multi_jobs:
job.join()
if self.flexranCtrlInstalled and self.flexranCtrlStarted:
SSH.open(EPC.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword)
Gabriele Perrone
committed
SSH.command('cd /opt/flexran_rtc', '\$', 5)
SSH.command('curl http://localhost:9999/stats | jq \'.\' > log/check_status_' + self.testCase_id + '.log 2>&1', '\$', 5)
SSH.command('cat log/check_status_' + self.testCase_id + '.log | jq \'.eNB_config[0].UE\' | grep -c rnti | sed -e "s#^#Nb Connected UE = #"', '\$', 5)
result = re.search('Nb Connected UE = (?P<nb_ues>[0-9]+)', SSH.getBefore())
passStatus = True
if result is not None:
nb_ues = int(result.group('nb_ues'))
htmlOptions = 'Nb Connected UE(s) to eNB = ' + str(nb_ues)
logging.debug('\u001B[1;37;44m ' + htmlOptions + ' \u001B[0m')
if self.expectedNbOfConnectedUEs > -1:
if nb_ues != self.expectedNbOfConnectedUEs:
passStatus = False
else:
htmlOptions = 'N/A'
Gabriele Perrone
committed
SSH.close()

Raphael Defosseux
committed
passStatus = True

Raphael Defosseux
committed
if (status_queue.empty()):
self.CreateHtmlTestRow(htmlOptions, 'KO', ALL_PROCESSES_OK)

Raphael Defosseux
committed
self.AutoTerminateUEandeNB()
else:
check_status = True
html_queue = SimpleQueue()
while (not status_queue.empty()):
count = status_queue.get()
if (count < 0):
check_status = False
device_id = status_queue.get()
message = status_queue.get()
html_cell = '<pre style="background-color:white">UE (' + device_id + ')\n' + message + '</pre>'
html_queue.put(html_cell)
if check_status and passStatus:
self.CreateHtmlTestRowQueue(htmlOptions, 'OK', len(self.UEDevices), html_queue)

Raphael Defosseux
committed
else:
self.CreateHtmlTestRowQueue(htmlOptions, 'KO', len(self.UEDevices), html_queue)

Raphael Defosseux
committed
self.AutoTerminateUEandeNB()
def GetAllUEIPAddresses(self):
if self.ADBIPAddress == '' or self.ADBUserName == '' or self.ADBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
if (len(self.UEDevices) == 1) and (self.UEDevices[0] == 'OAI-UE'):
if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '' or self.UESourceCodePath == '':
Usage()
sys.exit('Insufficient Parameter')
Gabriele Perrone
committed
SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
SSH.command('ifconfig oaitun_ue1', '\$', 4)
result = re.search('inet addr:(?P<ueipaddress>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)|inet (?P<ueipaddress2>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)', SSH.getBefore())
if result is not None:
if result.group('ueipaddress') is not None:
UE_IPAddress = result.group('ueipaddress')
else:
UE_IPAddress = result.group('ueipaddress2')
logging.debug('\u001B[1mUE (' + self.UEDevices[0] + ') IP Address is ' + UE_IPAddress + '\u001B[0m')
self.UEIPAddresses.append(UE_IPAddress)
else:
logging.debug('\u001B[1;37;41m UE IP Address Not Found! \u001B[0m')
ue_ip_status -= 1
Gabriele Perrone
committed
SSH.close()
return ue_ip_status
Gabriele Perrone
committed
SSH.open(self.ADBIPAddress, self.ADBUserName, self.ADBPassword)
for device_id in self.UEDevices:
if self.UEDevicesStatus[idx] != UE_STATUS_ATTACHED:
idx += 1
continue
count = 0
while count < 4:
if self.ADBCentralized:
Gabriele Perrone
committed
SSH.command('stdbuf -o0 adb -s ' + device_id + ' shell "ip addr show | grep rmnet"', '\$', 15)
Gabriele Perrone
committed
SSH.command('ssh ' + self.UEDevicesRemoteUser[idx] + '@' + self.UEDevicesRemoteServer[idx] + ' \'adb -s ' + device_id + ' shell "ip addr show | grep rmnet"\'', '\$', 60)
result = re.search('inet (?P<ueipaddress>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)\/[0-9]+[0-9a-zA-Z\.\s]+', SSH.getBefore())
if result is None:
logging.debug('\u001B[1;37;41m UE IP Address Not Found! \u001B[0m')
count += 1
else:
count = 10
if count < 9:
ue_ip_status -= 1
continue
UE_IPAddress = result.group('ueipaddress')
logging.debug('\u001B[1mUE (' + device_id + ') IP Address is ' + UE_IPAddress + '\u001B[0m')
for ueipaddress in self.UEIPAddresses:
if ueipaddress == UE_IPAddress:

Raphael Defosseux
committed
logging.debug('\u001B[1mUE (' + device_id + ') IP Address ' + UE_IPAddress + ': has already been allocated to another device !' + '\u001B[0m')
ue_ip_status -= 1
continue
self.UEIPAddresses.append(UE_IPAddress)
Gabriele Perrone
committed
SSH.close()
def ping_iperf_wrong_exit(self, lock, UE_IPAddress, device_id, statusQueue, message):
lock.acquire()
statusQueue.put(-1)
statusQueue.put(device_id)
statusQueue.put(UE_IPAddress)
statusQueue.put(message)
lock.release()
def Ping_common(self, lock, UE_IPAddress, device_id, statusQueue):
# Launch ping on the EPC side (true for ltebox and old open-air-cn)
# But for OAI-Rel14-CUPS, we launch from python executor
launchFromEpc = True
if re.match('OAI-Rel14-CUPS', EPC.EPCType, re.IGNORECASE):
ping_time = re.findall("-c (\d+)",str(self.ping_args))
SSH.open(EPC.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword)
SSH.command('cd ' + EPC.EPCSourceCodePath, '\$', 5)
Gabriele Perrone
committed
SSH.command('cd scripts', '\$', 5)
ping_status = SSH.command('stdbuf -o0 ping ' + self.ping_args + ' ' + UE_IPAddress + ' 2>&1 | stdbuf -o0 tee ping_' + self.testCase_id + '_' + device_id + '.log', '\$', int(ping_time[0])*1.5)
else:
cmd = 'ping ' + self.ping_args + ' ' + UE_IPAddress + ' 2>&1 > ping_' + self.testCase_id + '_' + device_id + '.log'
message = cmd + '\n'
logging.debug(cmd)
ret = subprocess.run(cmd, shell=True)
ping_status = ret.returncode
SSH.copyout(EPC.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword, 'ping_' + self.testCase_id + '_' + device_id + '.log', EPC.EPCSourceCodePath + '/scripts')
SSH.open(EPC.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword)
SSH.command('cat ' + EPC.EPCSourceCodePath + '/scripts/ping_' + self.testCase_id + '_' + device_id + '.log', '\$', 5)
# TIMEOUT CASE
if ping_status < 0:
message = 'Ping with UE (' + str(UE_IPAddress) + ') crashed due to TIMEOUT!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, UE_IPAddress, device_id, statusQueue, message)
return
result = re.search(', (?P<packetloss>[0-9\.]+)% packet loss, time [0-9\.]+ms', SSH.getBefore())
message = 'Packet Loss Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, UE_IPAddress, device_id, statusQueue, message)
packetloss = result.group('packetloss')
if float(packetloss) == 100:
message = 'Packet Loss is 100%'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, UE_IPAddress, device_id, statusQueue, message)
result = re.search('rtt min\/avg\/max\/mdev = (?P<rtt_min>[0-9\.]+)\/(?P<rtt_avg>[0-9\.]+)\/(?P<rtt_max>[0-9\.]+)\/[0-9\.]+ ms', SSH.getBefore())
message = 'Ping RTT_Min RTT_Avg RTT_Max Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
Gabriele Perrone
committed
SSH.close()
self.ping_iperf_wrong_exit(lock, UE_IPAddress, device_id, statusQueue, message)
rtt_min = result.group('rtt_min')
rtt_avg = result.group('rtt_avg')
rtt_max = result.group('rtt_max')
pal_msg = 'Packet Loss : ' + packetloss + '%'
min_msg = 'RTT(Min) : ' + rtt_min + ' ms'
avg_msg = 'RTT(Avg) : ' + rtt_avg + ' ms'
max_msg = 'RTT(Max) : ' + rtt_max + ' ms'
lock.acquire()
logging.debug('\u001B[1;37;44m ping result (' + UE_IPAddress + ') \u001B[0m')
logging.debug('\u001B[1;34m ' + pal_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + min_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + avg_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + max_msg + '\u001B[0m')
qMsg = pal_msg + '\n' + min_msg + '\n' + avg_msg + '\n' + max_msg
packetLossOK = True
if packetloss is not None:
if float(packetloss) > float(self.ping_packetloss_threshold):
logging.debug('\u001B[1;37;41m Packet Loss too high \u001B[0m')
elif float(packetloss) > 0:
logging.debug('\u001B[1;30;43m Packet Loss is not 0% \u001B[0m')
if (packetLossOK):
statusQueue.put(0)
else:
statusQueue.put(-1)
statusQueue.put(device_id)
statusQueue.put(UE_IPAddress)
statusQueue.put(qMsg)
lock.release()
Gabriele Perrone
committed
SSH.close()
except:
os.kill(os.getppid(),signal.SIGUSR1)
def PingNoS1_wrong_exit(self, qMsg):
html_queue = SimpleQueue()
html_cell = '<pre style="background-color:white">OAI UE ping result\n' + qMsg + '</pre>'
html_queue.put(html_cell)
self.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):
self.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.AutoTerminateUEandeNB()
ping_from_eNB = re.search('oaitun_enb1', str(self.ping_args))
if ping_from_eNB is not None:
if self.eNBIPAddress == '' or self.eNBUserName == '' or self.eNBPassword == '':
Usage()
sys.exit('Insufficient Parameter')
else:
if self.UEIPAddress == '' or self.UEUserName == '' or self.UEPassword == '':
Usage()
sys.exit('Insufficient Parameter')
try:
if ping_from_eNB is not None:
Gabriele Perrone
committed
SSH.open(self.eNBIPAddress, self.eNBUserName, self.eNBPassword)
SSH.command('cd ' + self.eNBSourceCodePath + '/cmake_targets/', '\$', 5)
Gabriele Perrone
committed
SSH.open(self.UEIPAddress, self.UEUserName, self.UEPassword)
SSH.command('cd ' + self.UESourceCodePath + '/cmake_targets/', '\$', 5)
SSH.command('cd cmake_targets', '\$', 5)
ping_time = re.findall("-c (\d+)",str(self.ping_args))
Gabriele Perrone
committed
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)
result = re.search(', (?P<packetloss>[0-9\.]+)% packet loss, time [0-9\.]+ms', SSH.getBefore())
if result is None:
message = 'Packet Loss Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
self.PingNoS1_wrong_exit(message)
return
packetloss = result.group('packetloss')
if float(packetloss) == 100:
message = 'Packet Loss is 100%'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
self.PingNoS1_wrong_exit(message)
result = re.search('rtt min\/avg\/max\/mdev = (?P<rtt_min>[0-9\.]+)\/(?P<rtt_avg>[0-9\.]+)\/(?P<rtt_max>[0-9\.]+)\/[0-9\.]+ ms', SSH.getBefore())
if result is None:
message = 'Ping RTT_Min RTT_Avg RTT_Max Not Found!'
logging.debug('\u001B[1;37;41m ' + message + ' \u001B[0m')
self.PingNoS1_wrong_exit(message)
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
return
rtt_min = result.group('rtt_min')
rtt_avg = result.group('rtt_avg')
rtt_max = result.group('rtt_max')
pal_msg = 'Packet Loss : ' + packetloss + '%'
min_msg = 'RTT(Min) : ' + rtt_min + ' ms'
avg_msg = 'RTT(Avg) : ' + rtt_avg + ' ms'
max_msg = 'RTT(Max) : ' + rtt_max + ' ms'
logging.debug('\u001B[1;37;44m OAI UE ping result \u001B[0m')
logging.debug('\u001B[1;34m ' + pal_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + min_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + avg_msg + '\u001B[0m')
logging.debug('\u001B[1;34m ' + max_msg + '\u001B[0m')
qMsg = pal_msg + '\n' + min_msg + '\n' + avg_msg + '\n' + max_msg
packetLossOK = True
if packetloss is not None:
if float(packetloss) > float(self.ping_packetloss_threshold):
qMsg += '\nPacket Loss too high'
logging.debug('\u001B[1;37;41m Packet Loss too high \u001B[0m')
packetLossOK = False
elif float(packetloss) > 0:
qMsg += '\nPacket Loss is not 0%'
logging.debug('\u001B[1;30;43m Packet Loss is not 0% \u001B[0m')
Gabriele Perrone
committed
SSH.close()
html_queue = SimpleQueue()
ip_addr = 'TBD'
html_cell = '<pre style="background-color:white">OAI UE ping result\n' + qMsg + '</pre>'
html_queue.put(html_cell)
if packetLossOK:
self.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue)
else:
self.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue)
# copying on the EPC server for logCollection
if ping_from_eNB is not None:
Gabriele Perrone
committed
copyin_res = SSH.copyin(self.eNBIPAddress, self.eNBUserName, self.eNBPassword, self.eNBSourceCodePath + '/cmake_targets/ping_' + self.testCase_id + '.log', '.')
else:
Gabriele Perrone
committed
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.EPCIPAddress, EPC.EPCUserName, EPC.EPCPassword, 'ping_' + self.testCase_id + '.log', EPC.EPCSourceCodePath + '/scripts')
except:
os.kill(os.getppid(),signal.SIGUSR1)
result = re.search('noS1', str(self.Initialize_eNB_args))
if result is not None:
self.PingNoS1()
return
if EPC.EPCIPAddress == '' or EPC.EPCUserName == '' or EPC.EPCPassword == '' or EPC.EPCSourceCodePath == '':
Usage()
sys.exit('Insufficient Parameter')
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):
self.CreateHtmlTestRow(self.ping_args, 'KO', pStatus)
self.AutoTerminateUEandeNB()
ueIpStatus = self.GetAllUEIPAddresses()
if (ueIpStatus < 0):
self.CreateHtmlTestRow(self.ping_args, 'KO', UE_IP_ADDRESS_ISSUE)
self.AutoTerminateUEandeNB()
multi_jobs = []
i = 0
lock = Lock()
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()
self.CreateHtmlTestRow(self.ping_args, 'KO', ALL_PROCESSES_OK)
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 = '<pre style="background-color:white">UE (' + device_id + ')\nIP Address : ' + ip_addr + '\n' + message + '</pre>'
html_queue.put(html_cell)
if (ping_status):
self.CreateHtmlTestRowQueue(self.ping_args, 'OK', len(self.UEDevices), html_queue)
else:
self.CreateHtmlTestRowQueue(self.ping_args, 'KO', len(self.UEDevices), html_queue)
def Iperf_ComputeTime(self):
result = re.search('-t (?P<iperf_time>\d+)', str(self.iperf_args))
if result is None:
logging.debug('\u001B[1;37;41m Iperf time Not Found! \u001B[0m')
sys.exit(1)
return result.group('iperf_time')
def Iperf_ComputeModifiedBW(self, idx, ue_num):
result = re.search('-b (?P<iperf_bandwidth>[0-9\.]+)[KMG]', str(self.iperf_args))
if result is None:
logging.debug('\u001B[1;37;41m Iperf bandwidth Not Found! \u001B[0m')
sys.exit(1)
iperf_bandwidth = result.group('iperf_bandwidth')