summaryrefslogtreecommitdiff
path: root/pjsip-apps/src/test-pjsua/mod_call.py
blob: 2cd81951e20a709a41b2c3e60ebe9a0faeceaa76 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# $Id$
import time
import imp
import sys
import inc_param as param
import inc_const as const

# Load configuration
cfg_file = imp.load_source("cfg_file", sys.argv[2])

# Test title
title = cfg_file.config.title
port1 = "9060"

# First pjsua
p1 = param.Pjsua(
		"callee",
		args = cfg_file.config.callee_cfg.arg + " --local-port="+port1,
		echo = cfg_file.config.callee_cfg.echo_enabled,
		trace = cfg_file.config.callee_cfg.trace_enabled
		)

# Second pjsua, make call to the first one
p2 = param.Pjsua(
		"caller",
		args = cfg_file.config.caller_cfg.arg + " --local-port=0",
		echo = cfg_file.config.caller_cfg.echo_enabled,
		trace = cfg_file.config.caller_cfg.trace_enabled
		)

# Test body function
def test_func(t):
	callee = t.process[0]
	caller = t.process[1]

	# Caller making call
	caller.send("m")
	caller.send("sip:localhost:" + port1)
	caller.expect(const.STATE_CALLING)
	
	# Callee answers with 200/OK
	time.sleep(1)
	callee.expect(const.EVENT_INCOMING_CALL)
	callee.send("a")
	callee.send("200")

	# Wait until call is connected in both endpoints
	time.sleep(1)
	if callee.expect(const.STATE_CONFIRMED, False)==None:
		raise TestError("Call failed")
	caller.expect(const.STATE_CONFIRMED)

	# Hold call
	time.sleep(2)
	caller.send("H")
	caller.expect(const.MEDIA_HOLD)
	callee.expect(const.MEDIA_HOLD)
	
	# Release hold
	time.sleep(2)
	caller.send("v")
	caller.expect(const.MEDIA_ACTIVE)
	callee.expect(const.MEDIA_ACTIVE)

	# UPDATE
	time.sleep(2)
	callee.send("U")
	callee.expect(const.MEDIA_ACTIVE)
	caller.expect(const.MEDIA_ACTIVE)
	
	# Send DTMF
	time.sleep(2)
	caller.send("#")
	caller.send("1122")
	callee.expect(const.RX_DTMF + "1")
	callee.expect(const.RX_DTMF + "1")
	callee.expect(const.RX_DTMF + "2")
	callee.expect(const.RX_DTMF + "2")

	# Hangup call
	time.sleep(1)
	caller.send("h")

	# Wait until calls are cleared in both endpoints
	caller.expect(const.STATE_DISCONNECTED)
	callee.expect(const.STATE_DISCONNECTED)
	

# Here where it all comes together
test = param.Test(title, run=[p1, p2], func=test_func)