voice-gateway-service/pjproject/tests/pjsua/mod_sipp.py

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