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
|
# $Id$
import sys
import imp
import re
import subprocess
import time
import inc_param as param
import inc_const as const
# Defaults
G_ECHO=True
G_TRACE=False
G_EXE="..\\..\\bin\\pjsua_vc6d.exe"
###################################
# TestError exception
class TestError:
desc = ""
def __init__(self, desc):
self.desc = desc
###################################
# Poor man's 'expect'-like class
class Expect:
proc = None
echo = False
trace_enabled = False
name = ""
rh = re.compile(const.DESTROYED)
ra = re.compile(const.ASSERT, re.I)
rr = re.compile(const.STDOUT_REFRESH)
def __init__(self, name, exe, args="", echo=G_ECHO, trace_enabled=G_TRACE):
self.name = name
self.echo = echo
self.trace_enabled = trace_enabled
fullcmd = exe + " " + args + " --stdout-refresh=5 --stdout-refresh-text=" + const.STDOUT_REFRESH
self.trace("Popen " + fullcmd)
self.proc = subprocess.Popen(fullcmd, bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
def send(self, cmd):
self.trace("send " + cmd)
self.proc.stdin.writelines(cmd + "\n")
def expect(self, pattern, raise_on_error=True):
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 + ": Timeout expecting pattern: " + pattern)
else:
return None # timeout
# Search for expected text
if r.search(line) != None:
return line
def wait(self):
self.trace("wait")
self.proc.wait()
def trace(self, s):
if self.trace_enabled:
print self.name + ": " + "====== " + s + " ======"
#########################
# Error handling
def handle_error(errmsg, t):
print "====== Caught error: " + errmsg + " ======"
time.sleep(1)
for p in t.process:
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])
# Validate
if script.test == None:
print "Error: no test defined"
sys.exit(1)
if len(script.test.run) == 0:
print "Error: test doesn't contain pjsua run descriptions"
sys.exit(1)
# Instantiate pjsuas
print "====== Running " + script.test.title + " ======"
for run in script.test.run:
try:
p = Expect(run.name, G_EXE, args=run.args, echo=run.echo, trace_enabled=run.trace)
# Wait until initialized
p.expect(const.PROMPT)
p.send("echo 1")
p.send("echo 1")
p.expect("echo 1")
# add running instance
script.test.process.append(p)
# run initial script
for cmd in run.cmds:
if len(cmd) >= 3 and cmd[2]!="":
print "====== " + cmd[2] + " ======"
if len(cmd) >= 1 and cmd[0]!="":
p.send(cmd[0])
if len(cmd) >= 2 and cmd[1]!="":
p.expect(cmd[1])
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)
except TestError, e:
handle_error(e.desc, script.test)
# Shutdown all instances
time.sleep(2)
for p in script.test.process:
p.send("q")
time.sleep(0.5)
p.expect(const.DESTROYED, False)
p.wait()
# Done
print "Test " + script.test.title + " completed successfully"
sys.exit(0)
|
