3 # Copyright (c) 2017 All rights reserved
4 # This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
13 import functest.utils.functest_utils as ft_utils
15 logger = logging.getLogger('sdnvpn-results')
18 class Results(object):
20 def __init__(self, line_length, conn=None):
22 self.line_length = line_length
23 self.test_result = "PASS"
27 self.num_tests_failed = 0
29 def get_ping_status(self,
32 expected="PASS", timeout=30):
33 ip_target = self.conn.compute.get_server(vm_target).\
34 addresses.values()[0][0]['addr']
35 self.get_ping_status_target_ip(vm_source, vm_target.name,
36 ip_target, expected, timeout)
38 def get_ping_status_target_ip(self,
42 expected="PASS", timeout=30):
43 console_log = self.conn.compute.\
44 get_server_console_output(vm_source)['output']
45 ip_source = self.conn.compute.get_server(vm_source).\
46 addresses.values()[0][0]['addr']
47 if "request failed" in console_log:
48 # Normally, cirros displays this message when userdata fails
49 logger.debug("It seems userdata is not supported in "
53 tab = ("%s" % (" " * 53))
54 expected_result = 'can ping' if expected == 'PASS' \
56 test_case_name = ("'%s' %s '%s'" %
60 logger.debug("%sPing\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
61 "%s-->Expected result: %s.\n"
62 % (tab, tab, vm_source.name, ip_source,
63 tab, target_name, ip_target,
64 tab, expected_result))
66 console_log = self.conn.compute.\
67 get_server_console_output(vm_source)['output']
68 # the console_log is a long string, we want to take
69 # the last 4 lines (for example)
70 lines = console_log.split('\n')
71 last_n_lines = lines[-5:]
72 if ("ping %s OK" % ip_target) in last_n_lines:
73 msg = ("'%s' can ping '%s'"
74 % (vm_source.name, target_name))
75 if expected == "PASS":
76 logger.debug("[PASS] %s" % msg)
77 self.add_success(test_case_name)
79 logger.debug("[FAIL] %s" % msg)
80 self.add_failure(test_case_name)
81 logger.debug("\n%s" % last_n_lines)
83 elif ("ping %s KO" % ip_target) in last_n_lines:
84 msg = ("'%s' cannot ping '%s'" %
85 (vm_source.name, target_name))
86 if expected == "FAIL":
87 logger.debug("[PASS] %s" % msg)
88 self.add_success(test_case_name)
90 logger.debug("[FAIL] %s" % msg)
91 self.add_failure(test_case_name)
96 logger.debug("[FAIL] Timeout reached for '%s'. "
97 "No ping output captured in the console log"
99 self.add_failure(test_case_name)
102 def add_to_summary(self, num_cols, col1, col2=""):
104 self.summary += ("+%s+\n" % (col1 * (self.line_length - 2)))
106 self.summary += ("| " + col1.ljust(self.line_length - 3) + "|\n")
108 self.summary += ("| %s" % col1.ljust(7) + "| ")
109 self.summary += (col2.ljust(self.line_length - 12) + "|\n")
110 if col1 in ("FAIL", "PASS"):
111 self.details.append({col2: col1})
114 self.num_tests_failed += 1
116 def record_action(self, msg):
117 """Record and log an action and display it in the summary."""
119 self.add_to_summary(1, msg)
121 def add_failure(self, test):
122 self.add_to_summary(2, "FAIL", test)
123 self.test_result = "FAIL"
125 def add_success(self, test):
126 self.add_to_summary(2, "PASS", test)
128 def add_subtest(self, test, successful):
130 self.add_success(test)
132 self.add_failure(test)
134 def check_ssh_output(self, vm_source, vm_target,
135 expected, timeout=30):
136 console_log = self.conn.compute.\
137 get_server_console_output(vm_source)['output']
138 ip_source = self.conn.compute.get_server(vm_source).\
139 addresses.values()[0][0]['addr']
140 ip_target = self.conn.compute.get_server(vm_target).\
141 addresses.values()[0][0]['addr']
143 if "request failed" in console_log:
144 # Normally, cirros displays this message when userdata fails
145 logger.debug("It seems userdata is not supported in "
149 tab = ("%s" % (" " * 53))
150 test_case_name = ("[%s] returns 'I am %s' to '%s'[%s]" %
151 (ip_target, expected,
152 vm_source.name, ip_source))
153 logger.debug("%sSSH\n%sfrom '%s' (%s)\n%sto '%s' (%s).\n"
154 "%s-->Expected result: %s.\n"
155 % (tab, tab, vm_source.name, ip_source,
156 tab, vm_target.name, ip_target,
159 console_log = self.conn.compute.\
160 get_server_console_output(vm_source)['output']
161 # the console_log is a long string, we want to take
162 # the last 4 lines (for example)
163 lines = console_log.split('\n')
164 last_n_lines = lines[-5:]
165 if ("%s %s" % (ip_target, expected)) in last_n_lines:
166 logger.debug("[PASS] %s" % test_case_name)
167 self.add_success(test_case_name)
169 elif ("%s not reachable" % ip_target) in last_n_lines:
170 logger.debug("[FAIL] %s" % test_case_name)
171 self.add_failure(test_case_name)
176 logger.debug("[FAIL] Timeout reached for '%s'."
177 " No ping output captured in the console log"
179 self.add_failure(test_case_name)
182 def ping_ip_test(self, address):
183 ping = "ping %s -c 10" % address
184 testcase_name = "Ping IP %s" % address
185 logger.debug(testcase_name)
186 exit_code = ft_utils.execute_command(ping)
189 self.add_failure(testcase_name)
191 self.add_success(testcase_name)
193 def compile_summary(self):
194 success_message = "All the subtests have passed."
195 failure_message = "One or more subtests have failed."
197 self.add_to_summary(0, "=")
198 logger.info("\n%s" % self.summary)
199 if self.test_result == "PASS":
200 logger.info(success_message)
202 logger.info(failure_message)
204 return {"status": self.test_result, "details": self.details}