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
|
"""atfutils module contains general purpose wrappers
*) assert_success
*) assert_failure
*) print-stdout
*) print_stderr
*) set_active_volume
"""
import re
import ssh
from atfglobals import GlobalObj
def assert_success(exit_status):
"""
"""
if exit_status is None:
return 1
if not exit_status:
return 0
else:
return 1
def assert_failure(exit_status):
"""
"""
if exit_status is None:
return 1
if exit_status:
return 0
else:
return 1
def expect(actual_string, expected_string):
"""
"""
escaped_pattern = re.escape(expected_string)
new_pattern = re.compile(escaped_pattern, re.IGNORECASE)
matched = False
if not expected_string:
matched = True
else:
if re.search(new_pattern, actual_string):
matched = True
return matched
def validate_output(output, expected_exit_status, expected_output, stream="stdout"):
"""
Parameters:
output: This is dictionary which is return value of
executecommand(ssh module)
expected_exit_status: exit_status we are expecting
expected_output : output expected from the execution of command
Returns:
True: if output we got from the command execution matches with
expected_exit_status and expected_output is
False: if output doesn't match with expected_exit_status or
expected_output
"""
output_status = False
if expected_exit_status is not 0:
exit_status = assert_failure(output['exitstatus'])
else:
exit_status = assert_success(output['exitstatus'])
if stream is "stdout":
output_status = expect(str(output['stdoutdata']),
expected_output)
elif stream is "stderr":
output_status = expect(str(output['stderrdata']),
expected_output)
if output_status is True and (not exit_status):
return 0
else:
return 1
def print_stdout(stdoutdata):
"""
"""
logger = GlobalObj.getLoggerObj()
if not stdoutdata == None:
for data in stdoutdata:
logger.debug(data)
def print_stderr(stderrdata):
logger = GlobalObj.getLoggerObj()
if not stderrdata == None:
for data in stderrdata:
logger.debug(data)
def set_active_volume(volumekey):
"""
"""
logger = GlobalObj.getLoggerObj()
env = GlobalObj.getTestenvObj()
return_status = env.setActiveVolume(volumekey)
if return_status:
logger.error("Unable to set Active Volume. \
'%s' Not defined in TestEnvironment" % volumekey )
return return_status
def get_active_volume():
"""
"""
logger = GlobalObj.getLoggerObj()
env = GlobalObj.getTestenvObj()
active_volume = env.getActiveVolume()
if active_volume is None:
logger.error("Active Volume not set in the TestEnvironment")
return active_volume
__all__ = ['assert_success',
'assert_failure',
'print-stdout',
'print_stderr',
'set_active_volume',
'get_active_volume']
|
