#****************************************************************************** # Eurecom OpenAirInterface # Copyright(c) 1999 - 2013 Eurecom # This program is free software; you can redistribute it and/or modify it # under the terms and conditions of the GNU General Public License, # version 2, as published by the Free Software Foundation. # This program is distributed in the hope it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or # FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for # more details. # You should have received a copy of the GNU General Public License along with # this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA. # The full GNU General Public License is included in this distribution in # the file called "COPYING". # Contact Information # Openair Admin: openair_admin@eurecom.fr # Openair Tech : openair_tech@eurecom.fr # Forums : http://forums.eurecom.fsr/openairinterface # Address : Eurecom, Compus SophiaTech 450, route des chappes, 06451 Biot, France #***************************************************************************** # \file core.py # \brief OAI core testing class that provides various primitives to send/recv cmd to openair class searching for patterns and process the responses and tag the test case as passed/failed/skipped # \author Navid Nikaein # \date 2013 # \version 0.1 # @ingroup _test import pexpect import time import re import string import sys import os import openair import log class core: def __init__(self): self.send_cr = 1 self.expect_echo = 0 self.expect_response = 1 self.flag_errors = 1 self.log = None def clean(self, obj): if type(obj) is str: return obj else: return repr(obj) def mark(self, marker): if self.log: print >> self.log, "\n\n{" + marker + "}\n\n" def expected(self, expected, got): return "================================= Failure ===================================\n"+\ "_________________________________ Expected __________________________________\n"+\ "--->" + expected + "<-\n" +\ "_________________________________ Received __________________________________\n"+\ "--->" + got + "<-\n" +\ "=============================================================================\n" def unexpected(self, notexpected, got): return "================================= Failure ===================================\n"+\ "__________________________ not expect to find _______________________________\n"+\ "---> " + self.clean(notexpected) + "\n" +\ "_________________________________ Received___________________________________\n"+\ "---> " + self.clean(got) + "\n" +\ "=============================================================================\n" def failed(self, command, expect,debug): time.sleep(2) ret = "================================= Failure =================================\n" ret +="_________________________________ Sent ____________________________________\n" ret +="---> " + command + "\n" ret +="______________________________Searching for _______________________________\n" ret +="---> " + self.clean(expect) + "\n" if debug >= 1 : ret +="________________________________ Received _________________________________\n" ret +="---> " + self.clean(self.oai.before) + "\n" ret +="_______________________________ Remaining _________________________________\n" ret +="---> " + self.clean(self.oai.after) + "\n" ret +="===========================================================================\n" return ret def err(self, command): return "============================ Error received ================================\n"+\ "__________________________________ Sent ____________________________________\n"+\ '---> ' + command + "\n"+\ "_________________________________ Error was ________________________________\n"+\ "---> " + self.oai.before + "\n" +\ "============================================================================\n" def wait_quiet(self, timeout=0.5): while 1: try: self.oai.expect(['..*'], timeout=0.5) except pexpect.TIMEOUT, e: return # print '[Flushing ' + self.oai.after + ']' # **************************Send***************************** # 1) send a command and return, do not wait # ************************************************************ def send_nowait(self, command, sudo=False): rsp1 = self.prompt1 rsp2 = self.prompt2 if sudo == True: command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' ' self.wait_quiet() if self.send_cr: log.stats['cmd'] += 1 self.oai.sendline(command) else: self.oai.send(command) # **************************Send***************************** # 1) send a command # 2) wait for a return prompt. Don't capture the response. # 3) Check for error or timeout. # ************************************************************ def send(self, command,sudo=False, timeout = 50, rsp1=None, rsp2=None,debug=0): if not rsp1: rsp1 = self.prompt1 if not rsp2: rsp2 = self.prompt2 self.wait_quiet() if sudo == True: command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' ' if self.send_cr: log.stats['cmd'] += 1 self.oai.sendline(command) else: self.oai.send(command) if self.expect_echo: #cmd = self.oai.expect([re.escape(command), pexpect.TIMEOUT], timeout=timeout); cmd = self.oai.expect_exact([command, pexpect.TIMEOUT], timeout=timeout); if cmd != 0: raise log.err(self.failed(command, command,debug)) if self.expect_response: #index = self.oai.expect([re.escape(rsp1), re.escape(rsp2),'%', pexpect.TIMEOUT], timeout=timeout) index = self.oai.expect_exact([rsp1, rsp2, pexpect.TIMEOUT], timeout=timeout) if index == 0 or index == 1: return 'OK' elif index == 2: #self.oai.expect([re.escape(rsp1), re.escape(rsp2), pexpect.TIMEOUT], timeout=timeout) self.oai.expect_exact([rsp1, rsp2, pexpect.TIMEOUT], timeout=timeout) if self.flag_errors: raise log.err(self.err(command)) else: return 'OK' else: raise log.err(self.failed(command, rsp1 + ' or ' + rsp2,debug)) # **************************send_recv************************* # 1) send a command # 2) wait for either rsp1 or rsp2 is found (normally prompts) # 3) return everything seen before that # ************************************************************ def send_recv(self, command, sudo=False, timeout=100, rsp1=None, rsp2=None,debug=0): if not rsp1: rsp1 = self.prompt1 if not rsp2: rsp2 = self.prompt2 self.wait_quiet() if sudo == True: #command = 'echo \'' + self.password + '\' | sudo -S -E ' + command command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' ' if self.send_cr: log.stats['cmd'] += 1 self.oai.sendline(command) else: self.oai.send(command) #index = self.oai.expect([re.escape(rsp1), re.escape(rsp2), pexpect.TIMEOUT], timeout=timeout); index = self.oai.expect_exact([rsp1, rsp2, pexpect.TIMEOUT], timeout=timeout); if index == 0 or index == 1 : return self.oai.before else: print "command = " + command + "rsp1 = " + rsp1 + " rsp2 = " + rsp2 + " index = " + str(index) raise log.err(self.failed(command, rsp1 + ' or ' + rsp2,debug)) # **************************send_expect************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expected pattern in the response # 3) raise an error if not found # ************************************************************** def send_expect(self, command, expect, sudo=False, delay = 50, rsp1=None, rsp2=None,debug=0): if debug : print command print expect print delay rsp = self.send_recv(command, sudo, delay, rsp1, rsp2) #print rsp if (rsp.find(expect) != -1): return 'Ok' raise log.err(self.failed(command, expect,debug)) # **************************send_expect_re************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expected pattern defined by a regular expression in the response # 3) return a error if raise_err flag is set and re not found, otherwise return 'Ok' # ***************************************************************** def send_expect_re(self, command, expect, sudo=False, raise_err=1, delay = 50, rsp1=None, rsp2=None,debug=0): rsp = self.send_recv(command, sudo, delay, rsp1, rsp2) # print rsp match = re.compile(expect).search(rsp) if match: return match if raise_err: raise log.err(self.failed(command, expect,debug)) else : return None # **************************send_expect************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expected pattern defined by a re in the response # 3) return ok if not found # ************************************************************** def send_expect_false(self, command, expect, sudo=False, delay = 5, rsp1=None, rsp2=None,debug=0): rsp = self.send_recv(command, sudo, delay, rsp1, rsp2) # print rsp if (rsp.find(expect) == -1): return 'OK' else: print "command = " + command + "expect = " + expect + "rsp = " + rsp raise log.err(self.failed(command, expect,debug)) # **************************send_wait************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expected pattern in the response # 3) retry for a numretries if not found # 4) return an error if not found after the numtries # 3) return the response if found # ************************************************************** def send_wait(self, command, expect, sudo=False, numretries=3, rsp1=None, rsp2=None,debug=0): timer = 0 for i in range(numretries): rsp = self.send_recv(command, sudo, 10, rsp1, rsp2) if (rsp.find(expect) != -1): return rsp; time.sleep(2) timer = timer+2 raise log.err(self.failed(command, expect,debug)) # **************************send_wait_re************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expected pattern defined by a re in the response # 3) retry for a numretries if not found # 4) return an error if not found after the numtries # 3) return the response if found # ************************************************************** def send_wait_re(self, command, expect, sudo=False, numretries=3, rsp1=None, rsp2=None,debug=0): timer = 0 for i in range(numretries): rsp = self.send_recv(command,sudo) if re.compile(expect).search(rsp): # print "Found in",i,"attempts" return rsp; time.sleep(2) timer = timer+2 raise log.err(self.failed(command, expect,debug)) # **************************send_wait_false************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expect pattern in the response # 3) return the response if not found # 4) return an error if the pattern found after the numtries # ************************************************************** def send_wait_false(self, command, expect, sudo=False, numretries=3, rsp1=None, rsp2=None,debug=0): timer = 1 for i in range(numretries): rsp = self.send_recv(command,sudo) if (rsp.find(expect) == -1): return rsp; time.sleep(2) timer = timer+2 raise log.err(self.failed(command, expect,debug)) # **************************send_wait_false************************* # 1) send a command, and optionally specify a the time to wait # 2) search for an expect pattern defined by a re in the response # 3) return the response if not found # 4) return an error if the pattern found after the numtries # ************************************************************** def send_wait_false_re(self, command, expect, sudo=False, numretries=3, rsp1=None, rsp2=None,debug=0): timer = 0 for i in range(numretries): rsp = self.send_recv(command,sudo) if not re.compile(expect).search(rsp): return rsp; time.sleep(2) timer = timer+2 raise log.err(self.failed(command, expect,debug)) # **************************find************************* # 1) find an exact pattern in a given string # 2) raise an error if not found # ************************************************************** def find(self, string, pattern): word = string.replace(pattern,'*','\*') words = string.replace(word,' ','\s*') if re.search(words,string): pass else: raise log.err(string) # **************************find_false************************** # 1) find an exact pattern in a given string # 2) raise an error if found # ************************************************************** def find_false(self, string, pattern): if string.find(pattern) != -1: raise log.err(string) # **************************find_re************************* # 1) find an exact re pattern in a given string # 2) raise an error if not found # ************************************************************** def find_re(self, string, pattern): if not re.compile(pattern).search(string): raise log.err(string) # **************************find_false_re************************* # 1) find an exact re pattern in a given string # 2) raise an error if found # ************************************************************** def find_false_re(self, string, pattern): if re.compile(pattern).search(string): raise log.err(string)