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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
|
#!/usr/bin/env python3
#
# (C) 2017 by sysmocom s.f.m.c. GmbH <info@sysmocom.de>
# All rights reserved.
#
# Author: Neels Hofmeyr <nhofmeyr@sysmocom.de>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
'''
Common code for osmo_interact_vty.py and osmo_interact_ctrl.py.
This implements all of application interaction, piping and verification.
osmo_interact_{vty,ctrl}.py plug VTY and CTRL interface specific bits.
'''
import argparse
import sys
import os
import subprocess
import time
import traceback
import socket
import shlex
class Interact:
class StepBase:
command = None
result = None
leading_blanks = None
def __init__(self):
self.result = []
def verify_interact_state(self, interact_instance):
# for example to verify that the last VTY prompt received shows the
# right node.
pass
def command_str(self, interact_instance=None):
return self.command
def __str__(self):
return '%s\n%s' % (self.command_str(), '\n'.join(self.result))
@staticmethod
def is_next_step(line, interact_instance):
assert not "implemented by InteractVty.VtyStep and InteractCtrl.CtrlStep"
socket = None
def __init__(self, step_class, port, host, verbose=False, update=False):
'''
host is the hostname to connect to.
port is the CTRL port to connect on.
'''
self.Step = step_class
self.port = port
self.host = host
self.verbose = verbose
self.update = update
if not port:
raise Exception("You need to provide port number to connect to")
def connect(self):
assert self.socket is None
retries = 30
took = 0
while True:
took += 1
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setblocking(1)
self.socket.connect((self.host, int(self.port)))
except IOError:
retries -= 1
if retries <= 0:
raise
time.sleep(.1)
continue
break
def close(self):
if self.socket is None:
return
self.socket.close()
self.socket = None
def command(self, command):
assert not "implemented separately by InteractVty and InteractCtrl"
def verify_transcript_file(self, transcript_file):
with open(transcript_file, 'r') as f:
content = f.read()
try:
result = self.verify_transcript(content)
except:
print('Error while verifying transcript file %r' % transcript_file, file=sys.stderr)
sys.stderr.flush()
raise
if not self.update:
return
content = '\n'.join(result)
with open(transcript_file, 'w') as f:
f.write(content)
def verify_transcript(self, transcript):
''''
transcript is a "screenshot" of a session, a multi-line string
including commands and expected results.
Feed commands to self.command() and verify the expected results.
'''
# parse steps
steps = []
step = None
blank_lines = 0
for line in transcript.splitlines():
if not line:
blank_lines += 1
continue
next_step_started = self.Step.is_next_step(line, self)
if next_step_started:
if step:
steps.append(step)
step = next_step_started
step.leading_blanks = blank_lines
blank_lines = 0
elif step:
# we only count blank lines directly preceding the start of a
# next step. Insert blank lines in the middle of a response
# back into the response:
if blank_lines:
step.result.extend([''] * blank_lines)
blank_lines = 0
step.result.append(line)
if step:
steps.append(step)
step = None
actual_result = []
# run steps
step_nr = 0
for step in steps:
step_nr += 1
try:
if self.verbose:
if step.leading_blanks:
print('\n' * step.leading_blanks, end='')
print(step.command_str())
sys.stdout.flush()
step.verify_interact_state(self)
res = self.command(step.command)
if self.verbose:
sys.stderr.flush()
sys.stdout.flush()
print('\n'.join(res))
sys.stdout.flush()
if step.leading_blanks:
actual_result.extend([''] * step.leading_blanks)
actual_result.append(step.command_str(self))
match_result = self.match_lines(step.result, res)
if self.update:
if match_result is True:
# preserve any wildcards
actual_result.extend(step.result)
else:
# mismatch, take exactly what came in
actual_result.extend(res)
continue
if match_result is not True:
raise Exception('Result mismatch:\n%s\n\nExpected:\n[\n%s\n]\n\nGot:\n[\n%s\n%s\n]'
% (match_result, step, step.command_str(), '\n'.join(res)))
except:
print('Error during transcript step %d:\n[\n%s\n]' % (step_nr, step),
file=sys.stderr)
sys.stderr.flush()
raise
# final line ending
actual_result.append('')
return actual_result
@staticmethod
def match_lines(expect, got):
'''
Match two lists of strings, allowing certain wildcards:
- In 'expect', if a line is exactly '...', it matches any number of
arbitrary lines in 'got'; the implementation is trivial and skips
lines to the first occurence in 'got' that continues after '...'.
Return 'True' on match, or a string describing the mismatch.
'''
def match_line(expect_line, got_line):
return expect_line == got_line
e = 0
g = 0
while e < len(expect):
if expect[e] == '...':
e += 1
if e >= len(expect):
# anything left in 'got' is accepted.
return True
# look for the next occurence of the expected line in 'got'
while g < len(got) and not match_line(expect[e], got[g]):
g += 1
continue
if g >= len(got):
return 'Cannot find line %r' % expect[e]
if not match_line(expect[e], got[g]):
return 'Mismatch:\nExpect:\n%r\nGot:\n%r' % (expect[e], got[g])
e += 1
g += 1
if g < len(got):
return 'Did not expect line %r' % got[g]
return True
def feed_commands(self, output, command_strs):
for command_str in command_strs:
for command in command_str.splitlines():
res = self.command(command)
output.write('\n'.join(res))
output.write('\n')
def end_process(proc, quiet=False):
if not proc:
return
rc = proc.poll()
if rc is not None:
if not quiet:
print('Process has already terminated with', rc)
proc.wait()
return
proc.terminate()
time_to_wait_for_term = 5
wait_step = 0.001
waited_time = 0
while True:
# poll returns None if proc is still running
if proc.poll() is not None:
break
waited_time += wait_step
# make wait_step approach 1.0
wait_step = (1. + 5. * wait_step) / 6.
if waited_time >= time_to_wait_for_term:
break
time.sleep(wait_step)
if proc.poll() is None:
# termination seems to be slower than that, let's just kill
proc.kill()
if not quiet:
print("Killed child process")
elif waited_time > .002:
if not quiet:
print("Terminating took %.3fs" % waited_time)
proc.wait()
class Application:
proc = None
_devnull = None
@staticmethod
def devnull():
if Application._devnull is None:
Application._devnull = open(os.devnull, 'w')
return Application._devnull
def __init__(self, run_app_str, purge_output=True, quiet=False):
self.command_tuple = shlex.split(run_app_str)
self.purge_output = purge_output
self.quiet = quiet
def run(self):
out_err = None
if self.purge_output:
out_err = Application.devnull()
if not self.quiet:
print('Launching: cd %r; %s' % (os.getcwd(), ' '.join(self.command_tuple)))
self.proc = subprocess.Popen(self.command_tuple, stdout=out_err, stderr=out_err)
def stop(self):
end_process(self.proc, self.quiet)
def verify_application(run_app_str, interact, transcript_file, verbose):
passed = None
application = None
sys.stdout.flush()
sys.stderr.flush()
if run_app_str:
application = Application(run_app_str, purge_output=not verbose)
application.run()
try:
interact.connect()
interact.verify_transcript_file(transcript_file)
passed = True
except:
traceback.print_exc()
passed = False
interact.close()
if application:
application.stop()
sys.stdout.flush()
sys.stderr.flush()
return passed
def common_parser():
parser = argparse.ArgumentParser()
parser.add_argument('-r', '--run', dest='run_app_str',
help='command to run to launch application to test,'
' including command line arguments. If omitted, no'
' application is launched.')
parser.add_argument('-p', '--port', dest='port',
help="Port to reach the application at.")
parser.add_argument('-H', '--host', dest='host', default='localhost',
help="Host to reach the application at.")
return parser
def parser_add_verify_args(parser):
parser.add_argument('-u', '--update', dest='update', action='store_true',
help='Do not verify, but OVERWRITE transcripts based on'
' the application\'s current behavior. OVERWRITES TRANSCRIPT'
' FILES.')
parser.add_argument('-v', '--verbose', action='store_true',
help='Print commands and application output')
parser.add_argument('transcript_files', nargs='*', help='transcript file(s) to verify')
return parser
def parser_add_run_args(parser):
parser.add_argument('-O', '--output', dest='output_path',
help="Write command results to a file instead of stdout."
"('-O -' writes to stdout and is the default)")
parser.add_argument('-c', '--command', dest='cmd_str',
help="Run this command (before reading input files, if any)."
" multiple commands may be separated by ';'")
parser.add_argument('cmd_files', nargs='*', help='file(s) with plain commands to run')
return parser
def main_run_commands(run_app_str, output_path, cmd_str, cmd_files, interact):
to_stdout = False
if not output_path or output_path == '-':
to_stdout = True
output = sys.stdout
else:
output = open(output_path, 'w')
application = None
if run_app_str:
application = Application(run_app_str, quiet=to_stdout)
application.run()
try:
interact.connect()
if cmd_str:
interact.feed_commands(output, cmd_str.split(';'))
for f_path in (cmd_files or []):
with open(f_path, 'r') as f:
interact.feed_commands(output, f.read().decode('utf-8').splitlines())
if not (cmd_str or cmd_files):
while True:
line = sys.stdin.readline()
if not line:
break;
interact.feed_commands(output, line.split(';'))
except:
traceback.print_exc()
finally:
if not to_stdout:
try:
output.close()
except:
traceback.print_exc()
try:
interact.close()
except:
traceback.print_exc()
if application:
try:
application.stop()
except:
traceback.print_exc()
def main_verify_transcripts(run_app_str, transcript_files, interact, verbose):
results = []
for t in transcript_files:
passed = verify_application(run_app_str=run_app_str,
interact=interact,
transcript_file=t,
verbose=verbose)
results.append((passed, t))
print('\nRESULTS:')
all_passed = True
for passed, t in results:
print('%s: %s' % ('pass' if passed else 'FAIL', t))
all_passed = all_passed and passed
print()
if not all_passed:
sys.exit(1)
# vim: tabstop=4 shiftwidth=4 expandtab nocin ai
|