import xml.etree.ElementTree as ET
-import inspect
import os
import configparser
class vm():
- def __init__(self, vm_xml_file, snapshot_xml_file, image, root_uid):
+ def __init__(self, vm_xml_file, snapshot_xml_file, image, root_uid, username, password):
self.log = log(4)
self.con = libvirt_con("qemu:///system")
try:
self.root_uid = root_uid
+ self.username = username
+ self.password = password
+
def define(self):
self.dom = self.con.con.defineXML(self.vm_xml)
if self.dom == None:
raise BaseException
def revert_snapshot(self):
- print(inspect.getmembers(self.dom, predicate=inspect.ismethod))
self.dom.revertToSnapshot(self.snapshot)
self.snapshot.delete()
#serial_con.close()
- def login(self, username, password):
+ def login(self):
try:
- self.serial_con = connection(self.get_serial_device(), username="root")
- self.serial_con.login("25814@root")
+ self.serial_con = connection(self.get_serial_device(), username=self.username)
+ self.serial_con.login(self.password)
except BaseException as e:
self.log.error("Could not connect to the domain via serial console")
return self.con.in_waiting
+ def line_in_buffer(self):
+ if b"\n" in self.buffer:
+ return True
+
+ return False
def readline2(self, pattern=None):
string = ""
print("We are not logged in")
return False
+ def print_lines_in_buffer(self):
+ while True:
+ self.log.debug("Fill buffer ...")
+ self.peek(len(self.buffer) + self.in_waiting)
+ self.log.debug("Current buffer length: {}".format(len(self.buffer)))
+ if self.line_in_buffer() == True:
+ while self.line_in_buffer() == True:
+ data = self.readline()
+ self.log_console_line(data.decode())
+ else:
+ self.log.debug("We have printed all lines in the buffer")
+ break
+
def login(self, password):
if self.username == None:
self.log.error("Username cannot be blank")
return False
+ self.print_lines_in_buffer()
+
# Hit enter to see what we get
self.con.write(b'\n')
# We get two new lines \r\n ?
data = self.readline()
self.log_console_line(data.decode())
+ self.print_lines_in_buffer()
if self.back_at_prompt():
self.log.debug("We are already logged in.")
# Read all line till we get login:
while 1:
- data = self.peek()
- if not data.decode() == "l":
- self.log.debug("We get no l at the start")
- self.log_console_line(self.readline().decode())
-
# We need to use self.in_waiting because with self.con.in_waiting we get
# not the complete string
size = len(self.buffer) + self.in_waiting
self.con.flush()
def command(self, command):
- self.write("{}\n".format(command))
+ self.write("{}; echo \"END: $?\"\n".format(command))
# We need to read out the prompt for this command first
# If we do not do this we will break the loop immediately
data = self.readline()
self.log_console_line(data.decode())
+ # We saved our exit code in data (the last line)
+ self.log.debug(data.decode())
+ data = data.decode().replace("END: ", "")
+ self.log.debug(data)
+ self.log.debug(data.strip())
+ return data.strip()
+
# A class which define and undefine a virtual network based on an xml file
class network():
self.virtual_machines = self.virtual_environ.get_machines()
def virtual_environ_start(self):
- pass
+ for name in self.virtual_environ.network_names:
+ self.virtual_networks[name].define()
+ self.virtual_networks[name].start()
+
+ for name in self.virtual_environ.machine_names:
+ self.virtual_machines[name].define()
+ self.virtual_machines[name].create_snapshot()
+ self.virtual_machines[name].start()
+
+ self.log.debug("Try to login on all machines")
+ for name in self.virtual_environ.machine_names:
+ self.virtual_machines[name].login()
def load_recipe(self):
- pass
+ try:
+ self.recipe = recipe(self.recipe_file)
+ except BaseException:
+ self.log.error("Failed to load recipe")
+ exit(1)
+
+ def run_recipe(self):
+ for line in self.recipe.recipe:
+ return_value = self.virtual_machines[line[0]].cmd(line[2])
+ self.log.debug("Return value is: {}".format(return_value))
+ if return_value != "0" and line[1] == "":
+ self.log.error("Failed to execute command '{}' on {}, return code: {}".format(line[2],line[0], return_value))
+ return False
+ elif return_value == "0" and line[1] == "!":
+ self.log.error("Succeded to execute command '{}' on {}, return code: {}".format(line[2],line[0],return_value))
+ return False
+ else:
+ self.log.debug("Command '{}' on {} returned with: {}".format(line[2],line[0],return_value))
- def run_recipe():
- pass
+ def virtual_environ_stop(self):
+ for name in self.virtual_environ.machine_names:
+ self.virtual_machines[name].shutdown()
+ self.virtual_machines[name].revert_snapshot()
+ self.virtual_machines[name].undefine()
- def virtual_environ_stop():
- pass
+ for name in self.virtual_environ.network_names:
+ self.virtual_networks[name].undefine()
# Should return all vms and networks in a list
self.log.debug(_machine)
machines.setdefault(_machine, vm(
os.path.normpath(self.path + "/" + self.config[_machine]["xml_file"]),
- os.path.normpath(self.path + "/" + self.config[_machine]["snapshot_xml_file"])))
+ os.path.normpath(self.path + "/" + self.config[_machine]["snapshot_xml_file"]),
+ self.config[_machine]["image"],
+ self.config[_machine]["root_uid"],
+ self.config[_machine]["username"],
+ self.config[_machine]["password"]))
return machines
+ @property
+ def machine_names(self):
+ return self.machines
+
+ @property
+ def network_names(self):
+ return self.networks
+
if __name__ == "__main__":
import argparse
args = parser.parse_args()
- _recipe = recipe("/home/jonatan/python-testing-kvm/test/recipe")
currenttest = test(args.dir)
currenttest.read_settings()
- currenttest.virtual_environ_setup()
\ No newline at end of file
+ currenttest.virtual_environ_setup()
+ currenttest.load_recipe()
+ currenttest.virtual_environ_start()
+ currenttest.run_recipe()
+ currenttest.virtual_environ_stop()
\ No newline at end of file