]>
git.ipfire.org Git - nitsi.git/blob - src/nitsi/test.py
dd610383bb286e6ccd589f5bfc331035ace70b9e
10 from . import virtual_environ
11 from . import settings
14 logger
= logging
.getLogger("nitsi.test")
17 class TestException(Exception):
18 def __init__(self
, message
):
19 self
.message
= message
22 def __init__(self
, log_path
, dir=None, recipe_file
=None, settings_file
=None, cmd_settings
=None):
26 # Set default values for the settings dict
27 self
.settings
["name"] = ""
28 self
.settings
["description"] = ""
29 self
.settings
["copy_from"] = None
30 self
.settings
["copy_to"] = None
31 self
.settings
["virtual_environ_path"] = None
32 self
.settings
["interactive_error_handling"] = True
34 self
.cmd_settings
= cmd_settings
35 self
.log_path
= log_path
37 # Init all vars with None
38 self
.settings_file
= None
39 self
.recipe_file
= None
42 # We need at least a path to a recipe file or a dir to a test
43 if not dir and not recipe
:
44 raise TestException("Did not get a path to a test or to a recipe file")
46 # We cannot decide which to use when we get both
47 if (dir and recipe_file
) or (dir and settings_file
):
48 raise TestException("Get dir and path to recipe or settings file")
52 if not os
.path
.isabs(dir):
53 self
.path
= os
.path
.abspath(dir)
54 except BaseException
as e
:
55 logger
.error("Could not get absolute path")
58 logger
.debug("Path of this test is: {}".format(self
.path
))
60 self
.recipe_file
= "{}/recipe".format(self
.path
)
61 self
.settings_file
= "{}/settings".format(self
.path
)
64 if not os
.path
.isabs(recipe_file
):
65 self
.recipe_file
= os
.path
.abspath(recipe_file
)
67 self
.recipe_file
= recipe_file
70 if not os
.path
.isabs(settings_file
):
71 self
.settings_file
= os
.path
.abspath(settings_file
)
73 self
.settings_file
= settings_file
75 # We can also go on without a settings file
76 if self
.settings_file
:
77 if not os
.path
.isfile(self
.settings_file
):
78 logger
.error("No such file: {}".format(self
.settings_file
))
79 raise TestException("No settings file found")
81 # os.path.isfile fails if self.recipe_file is None so we cannot use an and statement
83 if not os
.path
.isfile(self
.recipe_file
):
84 logger
.error("No such file: {}".format(self
.recipe_file
))
85 raise TestException("No recipe file found")
87 logger
.error("No such file: {}".format(self
.recipe_file
))
88 raise TestException("No recipe file found")
93 self
.log
= logger
.getChild(os
.path
.basename(self
.path
))
96 self
.log
= logger
.getChild(os
.path
.basename(self
.recipe_file
))
98 def read_settings(self
):
99 if self
.settings_file
:
100 self
.log
.debug("Going to read all settings from the ini file")
102 self
.config
= configparser
.ConfigParser()
103 self
.config
.read(self
.settings_file
)
104 except BaseException
as e
:
105 self
.log
.error("Failed to parse the config")
108 self
.settings
["name"] = self
.config
.get("GENERAL","name", fallback
="")
109 self
.settings
["description"] = self
.config
.get("GENERAL", "description", fallback
="")
110 self
.settings
["copy_to"] = self
.config
.get("GENERAL", "copy_to", fallback
=None)
111 self
.settings
["copy_from"] = self
.config
.get("GENERAL", "copy_from", fallback
=None)
112 self
.settings
["virtual_environ_path"] = self
.config
.get("VIRTUAL_ENVIRONMENT", "path", fallback
=None)
114 # We need to parse some settings here because they are loaded from a settings file
115 if not os
.path
.isabs(self
.settings
["virtual_environ_path"]):
116 self
.settings
["virtual_environ_path"] = os
.path
.normpath(os
.path
.dirname(
117 self
.settings_file
) + "/" + self
.settings
["virtual_environ_path"])
119 # Parse copy_from setting
120 if self
.settings
["copy_from"]:
121 self
.settings
["copy_from"] = settings
.settings_parse_copy_from(self
.settings
["copy_from"],
122 path
=os
.path
.dirname(self
.settings_file
))
124 # Update all settings from the cmd
125 self
.settings
.update(self
.cmd_settings
)
127 if not os
.path
.isabs(self
.settings
["virtual_environ_path"]):
128 self
.settings
["virtual_environ_path"] = os
.path
.abspath(self
.settings
["virtual_environ_path"])
131 # Check if we get at least a valid a valid path to virtual environ
132 if not self
.settings
["virtual_environ_path"]:
133 self
.log
.error("No path for virtual environment found.")
134 raise TestException("No path for virtual environment found.")
136 # Print all settings for debugging purpose
137 self
.log
.debug("Settings are:")
138 for key
in self
.settings
:
139 self
.log
.debug("{}: {}".format(key
, self
.settings
[key
]))
142 def virtual_environ_setup_stage_1(self
):
143 self
.virtual_environ
= virtual_environ
.VirtualEnviron(self
.settings
["virtual_environ_path"])
145 self
.virtual_networks
= self
.virtual_environ
.get_networks()
147 self
.virtual_machines
= self
.virtual_environ
.get_machines()
149 def virtual_environ_setup_stage_2(self
):
150 # built up which machines which are used in our recipe
153 for line
in self
.recipe
.recipe
:
154 if not line
[0] in used_machines
:
155 used_machines
.append(line
[0])
157 self
.log
.debug("Machines used in this recipe {}".format(used_machines
))
159 self
.used_machine_names
= used_machines
161 for machine
in self
.used_machine_names
:
162 if not machine
in self
.virtual_environ
.machine_names
:
163 raise TestException("{} is listed as machine in the recipe, but the virtual environmet does not have such a machine".format(machine
))
166 def virtual_environ_start(self
):
167 for name
in self
.virtual_environ
.network_names
:
168 self
.virtual_networks
[name
].define()
169 self
.virtual_networks
[name
].start()
171 for name
in self
.used_machine_names
:
172 self
.virtual_machines
[name
].define()
173 self
.virtual_machines
[name
].create_snapshot()
174 # We can only copy files when we know which and to which dir
175 if self
.settings
["copy_from"] and self
.settings
["copy_to"]:
176 self
.virtual_machines
[name
].copy_in(self
.settings
["copy_from"], self
.settings
["copy_to"])
177 self
.virtual_machines
[name
].start()
179 # Time to which all serial output log entries are relativ
180 log_start_time
= time
.time()
182 # Number of chars of the longest machine name
183 longest_machine_name
= self
.virtual_environ
.longest_machine_name
185 self
.log
.info("Try to intialize the serial connection, connect and login on all machines")
186 for name
in self
.used_machine_names
:
187 self
.log
.info("Try to initialize the serial connection connect and login on {}".format(name
))
188 self
.virtual_machines
[name
].serial_init(log_file
="{}/test.log".format(self
.log_path
),
189 log_start_time
=log_start_time
,
190 longest_machine_name
=longest_machine_name
)
191 self
.virtual_machines
[name
].serial_connect()
193 def load_recipe(self
):
194 self
.log
.info("Going to load the recipe")
196 self
.recipe
= recipe
.Recipe(self
.recipe_file
,
197 fallback_machines
=self
.virtual_environ
.machine_names
)
199 for line
in self
.recipe
.recipe
:
202 self
.log
.debug("This was the recipe")
203 except BaseException
as e
:
204 self
.log
.error("Failed to load recipe")
207 # This functions tries to handle an rror of the test (eg. when 'echo "Hello World"' failed)
208 # in an interactive way
209 # returns False when the test should exit right now, and True when the test should go on
210 def interactive_error_handling(self
):
211 if not self
.settings
["interactive_error_handling"]:
214 _cmd
= cmd
.CMD(intro
="You are droppped into an interative debugging shell because of the previous errors",
215 help={"exit": "Exit the test rigth now",
216 "continue": "Continues the test without any error handling, so do not expect that the test succeeds.",
217 "debug": "Disconnects from the serial console and prints the devices to manually connect to the virtual machines." \
218 "This is useful when you can fix th error with some manual commands. Please disconnect from the serial consoles and " \
219 "choose 'exit or 'continue' when you are done"})
221 command
= _cmd
.get_input(valid_commands
=["continue", "exit", "debug"])
223 if command
== "continue":
224 # The test should go on but we do not any debugging, so we return True
226 elif command
== "exit":
227 # The test should exit right now (normal behaviour)
230 # If we get here we are in debugging mode
231 # Disconnect from the serial console:
233 for name
in self
.used_machine_names
:
234 _cmd
.print_to_cmd("Disconnect from the serial console of {}".format(name
))
235 self
.virtual_machines
[name
].serial_disconnect()
237 # Print the serial device for each machine
238 for name
in self
.used_machine_names
:
239 device
= self
.virtual_machines
[name
].get_serial_device()
240 _cmd
.print_to_cmd("Serial device of {} is {}".format(name
, device
))
242 _cmd
.print_to_cmd("You can now connect to all serial devices, and send custom commands to the virtual machines." \
243 "Please type 'continue' or 'exit' when you disconnected from als serial devices and want to go on.")
245 command
= _cmd
.get_input(valid_commands
=["continue", "exit"])
247 if command
== "exit":
250 # We should continue whit the test
251 # Reconnect to the serial devices
253 for name
in self
.used_machine_names
:
254 self
.log
.info("Try to reconnect to {}".format(name
))
255 self
.virtual_machines
[name
].serial_connect()
259 def run_recipe(self
):
260 for line
in self
.recipe
.recipe
:
261 return_value
= self
.virtual_machines
[line
[0]].cmd(line
[2])
262 self
.log
.debug("Return value is: {}".format(return_value
))
263 if return_value
!= "0" and line
[1] == "":
264 err_msg
= "Failed to execute command '{}' on {}, return code: {}".format(line
[2],line
[0], return_value
)
265 # Try to handle this error in an interactive way, if we cannot go on
266 # raise an exception and exit
267 if not self
.interactive_error_handling():
268 raise TestException(err_msg
)
270 elif return_value
== "0" and line
[1] == "!":
271 err_msg
= "Succeded to execute command '{}' on {}, return code: {}".format(line
[2],line
[0],return_value
)
272 self
.log
.error(err_msg
)
273 # Try to handle this error in an interactive way, if we cannot go on
274 # raise an exception and exit
275 if not self
.interactive_error_handling():
276 raise TestException(err_msg
)
278 self
.log
.debug("Command '{}' on {} returned with: {}".format(line
[2],line
[0],return_value
))
280 def virtual_environ_stop(self
):
281 for name
in self
.used_machine_names
:
282 # We just catch exception here to avoid
283 # that we stop the cleanup process if only one command fails
285 self
.virtual_machines
[name
].shutdown()
286 self
.virtual_machines
[name
].revert_snapshot()
287 self
.virtual_machines
[name
].undefine()
288 except BaseException
as e
:
289 self
.log
.exception(e
)
291 for name
in self
.virtual_environ
.network_names
:
293 self
.virtual_networks
[name
].undefine()
294 except BaseException
as e
:
295 self
.log
.exception(e
)