123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282 |
- ## Automatic test module for SIPp.
- ##
- ## This module will need a test driver for each SIPp scenario:
- ## - For simple scenario, i.e: make/receive call (including auth), this
- ## test module can auto-generate a default test driver, i.e: make call
- ## or apply auto answer. Just name the SIPp scenario using "uas" or
- ## "uac" prefix accordingly.
- ## - Custom test driver can be defined in a python script file containing
- ## a list of the PJSUA instances and another list for PJSUA expects/
- ## commands. The custom test driver file must use the same filename as
- ## the SIPp XML scenario. See samples of SIPp scenario + its driver
- ## in tests/pjsua/scripts-sipp/ folder for detail.
- ##
- ## Here are defined macros that can be used in the custom driver:
- ## - $SIPP_PORT : SIPp binding port
- ## - $SIPP_URI : SIPp SIP URI
- ## - $PJSUA_PORT[N] : binding port of PJSUA instance #N
- ## - $PJSUA_URI[N] : SIP URI of PJSUA instance #N
- import ctypes
- import time
- import sys
- import os
- import re
- import subprocess
- from inc_cfg import *
- import inc_const
- import inc_util as util
- # flags that test is running in Unix
- G_INUNIX = False
- if sys.platform.lower().find("win32")!=-1 or sys.platform.lower().find("microsoft")!=-1:
- G_INUNIX = False
- else:
- G_INUNIX = True
- # /dev/null handle, for redirecting output when SIPP is not in background mode
- FDEVNULL = None
- # SIPp executable path and param
- #SIPP_PATH = '"C:\\devs\\bin\\Sipp_3.2\\sipp.exe"'
- SIPP_PATH = 'sipp'
- SIPP_PORT = 50070
- SIPP_PARAM = "-m 1 -i 127.0.0.1 -p " + str(SIPP_PORT)
- SIPP_TIMEOUT = 60
- # On BG mode, SIPp doesn't require special terminal
- # On non-BG mode, on win, it needs env var: "TERMINFO=c:\cygwin\usr\share\terminfo"
- # TODO: on unix with BG mode, waitpid() always fails, need to be fixed
- SIPP_BG_MODE = False
- #SIPP_BG_MODE = not G_INUNIX
- # Will be updated based on the test driver file (a .py file whose the same name as SIPp XML file)
- PJSUA_INST_PARAM = []
- PJSUA_EXPECTS = []
- # Default PJSUA param if test driver is not available:
- # - no-tcp as SIPp is on UDP only
- # - id, username, and realm: to allow PJSUA sending re-INVITE with auth after receiving 401/407 response
- PJSUA_DEF_PARAM = "--null-audio --max-calls=1 --no-tcp --id=sip:a@localhost --username=a --realm=*"
- # Get SIPp scenario (XML file)
- SIPP_SCEN_XML = ""
- if ARGS[1].endswith('.xml'):
- SIPP_SCEN_XML = ARGS[1]
- else:
- exit(-99)
- # Functions for resolving macros in the test driver
- def resolve_pjsua_port(mo):
- return str(PJSUA_INST_PARAM[int(mo.group(1))].sip_port)
- def resolve_pjsua_uri(mo):
- return PJSUA_INST_PARAM[int(mo.group(1))].uri[1:-1]
- def resolve_driver_macros(st):
- st = re.sub("\$SIPP_PORT", str(SIPP_PORT), st)
- st = re.sub("\$SIPP_URI", "sip:sipp@127.0.0.1:"+str(SIPP_PORT), st)
- st = re.sub("\$PJSUA_PORT\[(\d+)\]", resolve_pjsua_port, st)
- st = re.sub("\$PJSUA_URI\[(\d+)\]", resolve_pjsua_uri, st)
- return st
- # Init test driver
- if os.access(SIPP_SCEN_XML[:-4]+".py", os.R_OK):
- # Load test driver file (the corresponding .py file), if any
- cfg_file = util.load_module_from_file("cfg_file", SIPP_SCEN_XML[:-4]+".py")
- for ua_idx, ua_param in enumerate(cfg_file.PJSUA):
- ua_param = resolve_driver_macros(ua_param)
- PJSUA_INST_PARAM.append(InstanceParam("pjsua"+str(ua_idx), ua_param))
- if DEFAULT_TELNET and hasattr(cfg_file, 'PJSUA_CLI_EXPECTS'):
- PJSUA_EXPECTS = cfg_file.PJSUA_CLI_EXPECTS
- else:
- PJSUA_EXPECTS = cfg_file.PJSUA_EXPECTS
- else:
- # Generate default test driver
- if os.path.basename(SIPP_SCEN_XML)[0:3] == "uas":
- # auto make call when SIPp is as UAS
- ua_param = PJSUA_DEF_PARAM + " sip:127.0.0.1:" + str(SIPP_PORT)
- else:
- # auto answer when SIPp is as UAC
- ua_param = PJSUA_DEF_PARAM + " --auto-answer=200"
- PJSUA_INST_PARAM.append(InstanceParam("pjsua", ua_param))
- # Start SIPp process, returning PID
- def start_sipp():
- global SIPP_BG_MODE
- sipp_proc = None
- sipp_param = SIPP_PARAM + " -sf " + SIPP_SCEN_XML
- if SIPP_BG_MODE:
- sipp_param = sipp_param + " -bg"
- if SIPP_TIMEOUT:
- sipp_param = sipp_param + " -timeout "+str(SIPP_TIMEOUT)+"s -timeout_error" + " -deadcall_wait "+str(SIPP_TIMEOUT)+"s"
- # add target param
- sipp_param = sipp_param + " 127.0.0.1:" + str(PJSUA_INST_PARAM[0].sip_port)
- # run SIPp
- fullcmd = os.path.normpath(SIPP_PATH) + " " + sipp_param
- print("Running SIPP: " + fullcmd)
- if SIPP_BG_MODE:
- sipp_proc = subprocess.Popen(fullcmd, bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE, shell=G_INUNIX, universal_newlines=False)
- else:
- # redirect output to NULL
- global FDEVNULL
- #FDEVNULL = open(os.devnull, 'w')
- FDEVNULL = open("logs/sipp_output.tmp", 'w')
- sipp_proc = subprocess.Popen(fullcmd, shell=G_INUNIX, stdout=FDEVNULL, stderr=FDEVNULL)
- if not SIPP_BG_MODE:
- if sipp_proc == None or sipp_proc.poll():
- return None
- return sipp_proc
- else:
- # get SIPp child process PID
- pid = 0
- r = re.compile("PID=\[(\d+)\]", re.I)
- while True:
- line = sipp_proc.stdout.readline()
- pid_r = r.search(line)
- if pid_r:
- pid = int(pid_r.group(1))
- break
- if not sipp_proc.poll():
- break
- if pid != 0:
- # Win specific: get process handle from PID, as on win32, os.waitpid() takes process handle instead of pid
- if (sys.platform == "win32"):
- SYNCHRONIZE = 0x00100000
- PROCESS_QUERY_INFORMATION = 0x0400
- hnd = ctypes.windll.kernel32.OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION, False, pid)
- pid = hnd
- return pid
- # Wait SIPp process to exit, returning SIPp exit code
- def wait_sipp(sipp):
- if not SIPP_BG_MODE:
- global FDEVNULL
- sipp.wait()
- FDEVNULL.close()
- return sipp.returncode
- else:
- print("Waiting SIPp (PID=" + str(sipp) + ") to exit..")
- wait_cnt = 0
- while True:
- try:
- wait_cnt = wait_cnt + 1
- [pid_, ret_code] = os.waitpid(sipp, 0)
- if sipp == pid_:
- #print "SIPP returned ", ret_code
- ret_code = ret_code >> 8
- # Win specific: Close process handle
- if (sys.platform == "win32"):
- ctypes.windll.kernel32.CloseHandle(sipp)
-
- return ret_code
- except os.error:
- if wait_cnt <= 5:
- print("Retry ("+str(wait_cnt)+") waiting SIPp..")
- else:
- return -99
- # Execute PJSUA flow
- def exec_pjsua_expects(t, sipp):
- # Get all PJSUA instances
- ua = []
- for ua_idx in range(len(PJSUA_INST_PARAM)):
- ua.append(t.process[ua_idx])
- ua_err_st = ""
- while len(PJSUA_EXPECTS):
- expect = PJSUA_EXPECTS.pop(0)
- ua_idx = expect[0]
- expect_st = expect[1]
- send_cmd = resolve_driver_macros(expect[2])
- timeout = expect[3] if len(expect)>=4 else 0
- # Handle exception in pjsua flow, to avoid zombie SIPp process
- try:
- if expect_st != "":
- if timeout > 0:
- ua[ua_idx].expect(expect_st, raise_on_error = True, timeout = timeout)
- else:
- ua[ua_idx].expect(expect_st, raise_on_error = True)
- if send_cmd != "":
- ua[ua_idx].send(send_cmd)
- except TestError as e:
- ua_err_st = e.desc
- break;
- except:
- ua_err_st = "Unknown error"
- break;
- # Need to poll here for handling these cases:
- # - If there is no PJSUA EXPECT scenario, we must keep polling the stdout,
- # otherwise PJSUA process may stuck (due to stdout pipe buffer full?).
- # - last PJSUA_EXPECT contains a pjsua command that needs time to
- # finish, for example "v" (re-INVITE), the SIPp XML scenario may expect
- # that re-INVITE transaction to be completed and without stdout poll
- # PJSUA process may stuck.
- # Ideally the poll should be done contiunously until SIPp process is
- # terminated.
- # Update: now pjsua stdout is polled continuously by a dedicated thread,
- # so the poll is no longer needed
- #for ua_idx in range(len(ua)):
- # ua[ua_idx].expect(inc_const.STDOUT_REFRESH, raise_on_error = False)
- return ua_err_st
- def sipp_err_to_str(err_code):
- if err_code == 0:
- return "All calls were successful"
- elif err_code == 1:
- return "At least one call failed"
- elif err_code == 97:
- return "exit on internal command. Calls may have been processed"
- elif err_code == 99:
- return "Normal exit without calls processed"
- elif err_code == -1:
- return "Fatal error (timeout)"
- elif err_code == -2:
- return "Fatal error binding a socket"
- else:
- return "Unknown error"
- # Test body function
- def TEST_FUNC(t):
- sipp_ret_code = 0
- ua_err_st = ""
- sipp = start_sipp()
- if not sipp:
- raise TestError("Failed starting SIPp")
- ua_err_st = exec_pjsua_expects(t, sipp)
- sipp_ret_code = wait_sipp(sipp)
- if ua_err_st != "":
- raise TestError(ua_err_st)
- if sipp_ret_code:
- rc = ctypes.c_byte(sipp_ret_code).value
- raise TestError("SIPp returned error " + str(rc) + ": " + sipp_err_to_str(rc))
- # Here where it all comes together
- test = TestParam(SIPP_SCEN_XML[:-4],
- PJSUA_INST_PARAM,
- TEST_FUNC)
|