core.py 15 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
#/*
# * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
# * contributor license agreements.  See the NOTICE file distributed with
# * this work for additional information regarding copyright ownership.
# * The OpenAirInterface Software Alliance licenses this file to You under
# * the OAI Public License, Version 1.0  (the "License"); you may not use this file
# * except in compliance with the License.
# * You may obtain a copy of the License at
# *
# *      http://www.openairinterface.org/?page_id=698
# *
# * Unless required by applicable law or agreed to in writing, software
# * distributed under the License is distributed on an "AS IS" BASIS,
# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# * See the License for the specific language governing permissions and
# * limitations under the License.
# *-------------------------------------------------------------------------------
# * For more information about the OpenAirInterface (OAI) Software Alliance:
# *      contact@openairinterface.org
# */
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112

# \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:
113
           command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' '
114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133
        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:
134
            command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' '
135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174
        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:
Rohit Gupta's avatar
Rohit Gupta committed
175
          #command = 'echo \'' + self.password + '\' | sudo -S -E ' + command
Rohit Gupta's avatar
Rohit Gupta committed
176
          command = 'echo \'' + '' + '\' | sudo -S -E bash -c \' ' + command + '\' '
177 178 179 180 181 182 183 184 185 186
        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:
Rohit Gupta's avatar
Rohit Gupta committed
187
            print "command = " + command + "rsp1  = " + rsp1 + " rsp2 = " + rsp2 + " index = " + str(index)
188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
            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'
235
        else:
Rohit Gupta's avatar
Rohit Gupta committed
236
            print "command = " + command + "expect  = " + expect + "rsp = " + rsp
237
            raise log.err(self.failed(command, expect,debug))
238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343

    
    # **************************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)