]>
git.ipfire.org Git - nitsi.git/blob - src/nitsi/test.py
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, settings
=None, default_settings_file
=None):
24 self
.settings
= settings
26 self
.log_path
= log_path
28 # Init all vars with None
29 self
.settings_file
= None
30 self
.recipe_file
= None
33 self
.default_settings_file
= default_settings_file
35 # We need at least a path to a recipe file or a dir to a test
36 if not dir and not recipe
:
37 raise TestException("Did not get a path to a test or to a recipe file")
39 # We cannot decide which to use when we get both
40 if (dir and recipe_file
) or (dir and settings_file
):
41 raise TestException("Get dir and path to recipe or settings file")
45 if not os
.path
.isabs(dir):
46 self
.path
= os
.path
.abspath(dir)
47 except BaseException
as e
:
48 logger
.error("Could not get absolute path")
51 logger
.debug("Path of this test is: {}".format(self
.path
))
53 self
.recipe_file
= "{}/recipe".format(self
.path
)
54 self
.settings_file
= "{}/settings".format(self
.path
)
57 # We can also go on without a settings file
58 self
.settings_file
= self
.check_file(self
.settings_file
, log_level
=logging
.WARNING
)
60 self
.recipe_file
= self
.check_file(self
.recipe_file
)
61 if not self
.recipe_file
:
62 raise TestException("No recipe file found")
64 self
.default_settings_file
= self
.check_file(self
.default_settings_file
)
68 self
.log
= logger
.getChild(os
.path
.basename(self
.path
))
69 # We get a recipe when we get here
71 self
.log
= logger
.getChild(os
.path
.basename(self
.recipe_file
))
73 # Parse config and settings:
74 if self
.settings_file
:
75 self
.settings
.set_config_values_from_file(self
.settings_file
, type="settings-file")
77 if self
.default_settings_file
:
78 self
.settings
.set_config_values_from_file(self
.default_settings_file
, type="default-settings-file")
81 self
.settings
.check_config_values()
85 # returns an absolut path, when the file is valid, None when not
86 def check_file(self
, file, log_level
=logging
.ERROR
):
88 logger
.debug("File to check is: {}".format(file))
89 if not os
.path
.isfile(file):
90 err_msg
= "No such file: {}".format(file)
91 logger
.log(log_level
,err_msg
)
93 if not os
.path
.isabs(file):
94 file = os
.path
.abspath(file)
99 def virtual_environ_setup_stage_1(self
):
100 self
.virtual_environ
= virtual_environ
.VirtualEnviron(self
.settings
.get_config_value("virtual_environ_path"))
102 self
.virtual_networks
= self
.virtual_environ
.get_networks()
104 self
.virtual_machines
= self
.virtual_environ
.get_machines()
106 def virtual_environ_setup_stage_2(self
):
107 # built up which machines which are used in our recipe
110 for line
in self
.recipe
.recipe
:
111 if not line
[0] in used_machines
:
112 used_machines
.append(line
[0])
114 self
.log
.debug("Machines used in this recipe {}".format(used_machines
))
116 self
.used_machine_names
= used_machines
118 for machine
in self
.used_machine_names
:
119 if not machine
in self
.virtual_environ
.machine_names
:
120 raise TestException("{} is listed as machine in the recipe, but the virtual environmet does not have such a machine".format(machine
))
123 def virtual_environ_start(self
):
124 for name
in self
.virtual_environ
.network_names
:
125 self
.virtual_networks
[name
].define()
126 self
.virtual_networks
[name
].start()
128 for name
in self
.used_machine_names
:
129 self
.virtual_machines
[name
].define()
130 self
.virtual_machines
[name
].create_snapshot()
131 # We can only copy files when we know which and to which dir
132 if self
.settings
.get_config_value("copy_from") and self
.settings
.get_config_value("copy_to"):
133 self
.virtual_machines
[name
].copy_in(self
.settings
.get_config_value("copy_from"), self
.settings
.get_config_value("copy_to"))
134 self
.virtual_machines
[name
].start()
136 # Time to which all serial output log entries are relativ
137 log_start_time
= time
.time()
139 # Number of chars of the longest machine name
140 longest_machine_name
= self
.virtual_environ
.longest_machine_name
142 self
.log
.info("Try to intialize the serial connection, connect and login on all machines")
143 for name
in self
.used_machine_names
:
144 self
.log
.info("Try to initialize the serial connection connect and login on {}".format(name
))
145 self
.virtual_machines
[name
].serial_init(log_file
="{}/test.log".format(self
.log_path
),
146 log_start_time
=log_start_time
,
147 longest_machine_name
=longest_machine_name
)
148 self
.virtual_machines
[name
].serial_connect()
150 def load_recipe(self
):
151 self
.log
.info("Going to load the recipe")
153 self
.recipe
= recipe
.Recipe(self
.recipe_file
,
154 fallback_machines
=self
.virtual_environ
.machine_names
,
155 include_path
=self
.settings
.get_config_value("include_path"))
157 for line
in self
.recipe
.recipe
:
160 self
.log
.debug("This was the recipe")
161 except BaseException
as e
:
162 self
.log
.error("Failed to load recipe")
165 # This functions tries to handle an error of the test (eg. when 'echo "Hello World"' failed)
166 # in an interactive way
167 # returns False when the test should exit right now, and True when the test should go on
168 def interactive_error_handling(self
):
169 if not self
.settings
.get_config_value("interactive_error_handling"):
172 _cmd
= cmd
.CMD(intro
="You are droppped into an interative debugging shell because of the previous errors",
173 help={"exit": "Exit the test rigth now",
174 "continue": "Continues the test without any error handling, so do not expect that the test succeeds.",
175 "debug": "Disconnects from the serial console and prints the devices to manually connect to the virtual machines." \
176 "This is useful when you can fix th error with some manual commands. Please disconnect from the serial consoles and " \
177 "choose 'exit or 'continue' when you are done"})
179 command
= _cmd
.get_input(valid_commands
=["continue", "exit", "debug"])
181 if command
== "continue":
182 # The test should go on but we do not any debugging, so we return True
184 elif command
== "exit":
185 # The test should exit right now (normal behaviour)
188 # If we get here we are in debugging mode
189 # Disconnect from the serial console:
191 for name
in self
.used_machine_names
:
192 _cmd
.print_to_cmd("Disconnect from the serial console of {}".format(name
))
193 self
.virtual_machines
[name
].serial_disconnect()
195 # Print the serial device for each machine
196 for name
in self
.used_machine_names
:
197 device
= self
.virtual_machines
[name
].get_serial_device()
198 _cmd
.print_to_cmd("Serial device of {} is {}".format(name
, device
))
200 _cmd
.print_to_cmd("You can now connect to all serial devices, and send custom commands to the virtual machines." \
201 "Please type 'continue' or 'exit' when you disconnected from als serial devices and want to go on.")
203 command
= _cmd
.get_input(valid_commands
=["continue", "exit"])
205 if command
== "exit":
208 # We should continue whit the test
209 # Reconnect to the serial devices
211 for name
in self
.used_machine_names
:
212 self
.log
.info("Try to reconnect to {}".format(name
))
213 self
.virtual_machines
[name
].serial_connect()
217 def run_recipe(self
):
218 for line
in self
.recipe
.recipe
:
219 return_value
= self
.virtual_machines
[line
[0]].cmd(line
[2])
220 self
.log
.debug("Return value is: {}".format(return_value
))
221 if return_value
!= "0" and line
[1] == "":
222 err_msg
= "Failed to execute command '{}' on {}, return code: {}".format(line
[2],line
[0], return_value
)
223 # Try to handle this error in an interactive way, if we cannot go on
224 # raise an exception and exit
225 if not self
.interactive_error_handling():
226 raise TestException(err_msg
)
228 elif return_value
== "0" and line
[1] == "!":
229 err_msg
= "Succeded to execute command '{}' on {}, return code: {}".format(line
[2],line
[0],return_value
)
230 self
.log
.error(err_msg
)
231 # Try to handle this error in an interactive way, if we cannot go on
232 # raise an exception and exit
233 if not self
.interactive_error_handling():
234 raise TestException(err_msg
)
236 self
.log
.debug("Command '{}' on {} returned with: {}".format(line
[2],line
[0],return_value
))
238 def virtual_environ_stop(self
):
239 for name
in self
.used_machine_names
:
240 # We just catch exception here to avoid
241 # that we stop the cleanup process if only one command fails
243 self
.virtual_machines
[name
].shutdown()
244 self
.virtual_machines
[name
].revert_snapshot()
245 self
.virtual_machines
[name
].undefine()
246 except BaseException
as e
:
247 self
.log
.exception(e
)
249 for name
in self
.virtual_environ
.network_names
:
251 self
.virtual_networks
[name
].undefine()
252 except BaseException
as e
:
253 self
.log
.exception(e
)