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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
# $Id$
import sys
import imp
import re
import os
import subprocess
import random
import time
import inc_const as const
from inc_cfg import *
# Get the pjsua executable name
if sys.platform.find("win32")!=-1:
e = "../../bin/pjsua_vc6d.exe"
st1 = os.stat(e)
if st1 != None:
G_EXE = e
e = "../../bin/pjsua_vc6d.exe"
st2 = os.stat(e)
if st2 != None and st2.st_mtime > st1.st_mtime:
G_EXE = e
st1 = st2
if G_EXE=="":
print "Unable to find valid pjsua. Please build pjsip first"
sys.exit(1)
G_INUNIX = False
else:
f = open("../../../build.mak", "r")
while True:
line = f.readline()
if not line:
break
if line.find("TARGET_NAME")!=-1:
print line
G_EXE="../../bin/pjsua-" + line.split(":= ")[1]
break
if G_EXE=="":
print "Unable to find ../../../build.mak. Please build pjsip first"
sys.exit(1)
G_INUNIX = True
G_EXE = G_EXE.rstrip("\n\r \t")
###################################
# Poor man's 'expect'-like class
class Expect:
proc = None
echo = False
trace_enabled = False
name = ""
inst_param = None
rh = re.compile(const.DESTROYED)
ra = re.compile(const.ASSERT, re.I)
rr = re.compile(const.STDOUT_REFRESH)
t0 = time.time()
def __init__(self, inst_param):
self.inst_param = inst_param
self.name = inst_param.name
self.echo = inst_param.echo_enabled
self.trace_enabled = inst_param.trace_enabled
fullcmd = G_EXE + " " + inst_param.arg + " --stdout-no-buf --stdout-refresh=5 --stdout-refresh-text=" + const.STDOUT_REFRESH
self.trace("Popen " + fullcmd)
self.proc = subprocess.Popen(fullcmd, shell=G_INUNIX, bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=False)
def send(self, cmd):
self.trace("send " + cmd)
self.proc.stdin.writelines(cmd + "\n")
self.proc.stdin.flush()
def expect(self, pattern, raise_on_error=True, title=""):
self.trace("expect " + pattern)
r = re.compile(pattern, re.I)
refresh_cnt = 0
while True:
line = self.proc.stdout.readline()
if line == "":
raise TestError(self.name + ": Premature EOF")
# Print the line if echo is ON
if self.echo:
print self.name + ": " + line,
# Trap assertion error
if self.ra.search(line) != None:
if raise_on_error:
raise TestError(self.name + ": " + line)
else:
return None
# Count stdout refresh text.
if self.rr.search(line) != None:
refresh_cnt = refresh_cnt+1
if refresh_cnt >= 6:
self.trace("Timed-out!")
if raise_on_error:
raise TestError(self.name + " " + title + ": Timeout expecting pattern: \"" + pattern + "\"")
else:
return None # timeout
# Search for expected text
if r.search(line) != None:
return line
def sync_stdout(self):
self.trace("sync_stdout")
cmd = "echo 1" + str(random.randint(1000,9999))
self.send(cmd)
self.expect(cmd)
def wait(self):
self.trace("wait")
self.proc.wait()
def trace(self, s):
if self.trace_enabled:
now = time.time()
fmt = self.name + ": " + "================== " + s + " ==================" + " [at t=%(time)03d]"
print fmt % {'time':int(now - self.t0)}
#########################
# Error handling
def handle_error(errmsg, t, close_processes = True):
print "====== Caught error: " + errmsg + " ======"
if (close_processes):
time.sleep(1)
for p in t.process:
p.send("q")
p.send("q")
p.expect(const.DESTROYED, False)
p.wait()
print "Test completed with error: " + errmsg
sys.exit(1)
#########################
# MAIN
if len(sys.argv)!=3:
print "Usage: run.py MODULE CONFIG"
print "Sample:"
print " run.py mod_run.py scripts-run/100_simple.py"
sys.exit(1)
# Import the test script
script = imp.load_source("script", sys.argv[1])
# Init random seed
random.seed()
# Validate
if script.test == None:
print "Error: no test defined"
sys.exit(1)
if script.test.skip:
print "Test " + script.test.title + " is skipped"
sys.exit(0)
if len(script.test.inst_params) == 0:
print "Error: test doesn't contain pjsua run descriptions"
sys.exit(1)
# Instantiate pjsuas
print "====== Running " + script.test.title + " ======"
print "Using " + G_EXE + " as pjsua executable"
for inst_param in script.test.inst_params:
try:
# Create pjsua's Expect instance from the param
p = Expect(inst_param)
# Wait until registration completes
if inst_param.have_reg:
p.expect(inst_param.uri+".*registration success")
# Synchronize stdout
p.send("")
p.expect(const.PROMPT)
p.send("echo 1")
p.send("echo 1")
p.expect("echo 1")
# add running instance
script.test.process.append(p)
except TestError, e:
handle_error(e.desc, script.test)
# Run the test function
if script.test.test_func != None:
try:
script.test.test_func(script.test, script.test.user_data)
except TestError, e:
handle_error(e.desc, script.test)
# Shutdown all instances
time.sleep(2)
for p in script.test.process:
# Unregister if we have_reg to make sure that next tests
# won't wail
if p.inst_param.have_reg:
p.send("ru")
p.expect(p.inst_param.uri+".*unregistration success")
p.send("q")
p.send("q")
time.sleep(0.5)
p.expect(const.DESTROYED, False)
p.wait()
# Run the post test function
if script.test.post_func != None:
try:
script.test.post_func(script.test, script.test.user_data)
except TestError, e:
handle_error(e.desc, script.test, False)
# Done
print "Test " + script.test.title + " completed successfully"
sys.exit(0)
|