# $Id$ import imp import sys import inc_sip as sip import inc_const as const import re from inc_cfg import * # Read configuration cfg_file = imp.load_source("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) last_cseq = 0 last_method = "" last_call_id = "" for t in cfg_file.recvfrom_cfg.transaction: # 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)