12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152 |
- 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])
- # Test body function
- def test_func(t):
- pjsua = t.process[0]
- # Create dialog
- dlg = sip.Dialog("127.0.0.1", pjsua.inst_param.sip_port,
- tcp=cfg_file.sendto_cfg.use_tcp)
- #dlg = sip.Dialog("127.0.0.1", 5060, tcp=cfg_file.sendto_cfg.use_tcp)
- cfg = cfg_file.sendto_cfg
-
- if len(cfg.complete_msg) != 0:
- req = dlg.update_fields(cfg.complete_msg)
- else:
- req = dlg.create_invite(cfg.sdp, cfg.extra_headers, cfg.body)
- resp = dlg.send_request_wait(req, 10)
- if resp=="":
- raise TestError("Timed-out waiting for response")
- # Check response code
- code = int(sip.get_code(resp))
- if code != cfg.resp_code:
- dlg.hangup(code)
- raise TestError("Expecting code " + str(cfg.resp_code) +
- " got " + str(code))
- # Check for patterns that must exist
- for p in cfg.resp_include:
- if re.search(p, resp, re.M | re.I)==None:
- dlg.hangup(code)
- raise TestError("Pattern " + p + " not found")
- # Check for patterns that must not exist
- for p in cfg.resp_exclude:
- if re.search(p, resp, re.M | re.I)!=None:
- dlg.hangup(code)
- raise TestError("Excluded pattern " + p + " found")
- pjsua.sync_stdout()
- dlg.hangup(code)
- pjsua.sync_stdout()
- # Here where it all comes together
- test = TestParam(cfg_file.sendto_cfg.name,
- [cfg_file.sendto_cfg.inst_param],
- test_func)
|