voice-gateway/tests/pjsua/mod_recvfrom.py

import sys
import inc_sip as sip
import inc_const as const
import inc_util as util
import re
from inc_cfg import *

# Read configuration
cfg_file = util.load_module_from_file("cfg_file", ARGS[1])

# Default server port (should we randomize?)
srv_port = 50070

def test_func(test):
    pjsua = test.process[0]
    dlg = sip.Dialog("127.0.0.1", pjsua.inst_param.sip_port, 
             local_port=srv_port, 
             tcp=cfg_file.recvfrom_cfg.tcp)

    config = pjsua.get_config(cfg_file.recvfrom_cfg.pj_config)
    print("Config : " + config)

    last_cseq = 0
    last_method = ""
    last_call_id = ""
    for t in cfg_file.recvfrom_cfg.transaction:
        # Check if transaction requires configuration
        if t.pj_config != "":
            r = re.compile(t.pj_config, re.I)
            if r.search(config) == None:
                print("Configuration : " + t.pj_config + " not found, skipping")
                continue

        # Print transaction title
        if t.title != "":
            dlg.trace(t.title)
        # Run command and expect patterns
        for c in t.cmds:
            if c[0] and c[0] != "":
                pjsua.send(c[0])
            if len(c)>1 and c[1] and c[1] != "":
                pjsua.expect(c[1])
        # Wait for request
        if t.check_cseq:
            # Absorbs retransmissions
            cseq = 0
            method = last_method
            call_id = last_call_id
            while cseq <= last_cseq and method == last_method and call_id == last_call_id:
                request, src_addr = dlg.wait_msg_from(30)
                if request==None or request=="":
                    raise TestError("Timeout waiting for request")
                method = request.split(" ", 1)[0]
                cseq_hval = sip.get_header(request, "CSeq")
                cseq_hval = cseq_hval.split(" ")[0]
                cseq = int(cseq_hval)
                call_id = sip.get_header(request, "Call-ID")
            last_cseq = cseq
            last_method = method
        else:
            request, src_addr = dlg.wait_msg_from(30)
            if request==None or request=="":
                raise TestError("Timeout waiting for request")

        # Check for include patterns
        for pat in t.include:
            if re.search(pat, request, re.M | re.I)==None:
                if t.title:
                    tname = " in " + t.title + " transaction"
                else:
                    tname = ""
                raise TestError("Pattern " + pat + " not found" + tname)
        # Check for exclude patterns
        for pat in t.exclude:
            if re.search(pat, request, re.M | re.I)!=None:
                if t.title:
                    tname = " in " + t.title + " transaction"
                else:
                    tname = ""
                raise TestError("Excluded pattern " + pat + " found" + tname)
        # Create response
        if t.resp_code!=0:
            response = dlg.create_response(request, t.resp_code, "Status reason")
            # Add headers to response
            for h in t.resp_hdr:
                response = response + h + "\r\n"
            # Add message body if required
            if t.body:
                response = response + t.body
            # Send response
            dlg.send_msg(response, src_addr)

        # Expect something to happen in pjsua
        if t.expect != "":
            pjsua.expect(t.expect)
        # Sync
        pjsua.sync_stdout()

# Replace "$PORT" with server port in pjsua args
cfg_file.recvfrom_cfg.inst_param.arg = cfg_file.recvfrom_cfg.inst_param.arg.replace("$PORT", str(srv_port))

# Here where it all comes together
test = TestParam(cfg_file.recvfrom_cfg.name, 
                 [cfg_file.recvfrom_cfg.inst_param], 
                 test_func)