flavor_ram = 2048
flavor_vcpus = 1
flavor_disk = 0
+ create_server_timeout = 300
def __init__(self, **kwargs):
if "case_name" not in kwargs:
(_, self.privkey_filename) = tempfile.mkstemp()
(_, self.pubkey_filename) = tempfile.mkstemp()
+ def check_requirements(self):
+ if len(self.orig_cloud.list_hypervisors()) < 2:
+ self.__logger.warning("Vmtp requires at least 2 hypervisors")
+ self.is_skipped = True
+ self.project.clean()
+
def create_network_resources(self):
"""Create router
if not os.path.exists(self.res_dir):
os.makedirs(self.res_dir)
cmd = ['vmtp', '-sc']
- output = subprocess.check_output(cmd)
+ output = subprocess.check_output(cmd).decode("utf-8")
self.__logger.info("%s\n%s", " ".join(cmd), output)
with open(self.config, "w+") as conf:
- vmtp_conf = yaml.load(output)
+ vmtp_conf = yaml.full_load(output)
vmtp_conf["private_key_file"] = self.privkey_filename
vmtp_conf["public_key_file"] = self.pubkey_filename
vmtp_conf["image_name"] = str(self.image.name)
vmtp_conf["security_group_name"] = "pns-security{}".format(
self.guid)
vmtp_conf["dns_nameservers"] = [env.get('NAMESERVER')]
+ vmtp_conf["generic_retry_count"] = self.create_server_timeout // 2
conf.write(yaml.dump(vmtp_conf))
def run_vmtp(self):
OS_PROJECT_NAME=self.project.project.name,
OS_PROJECT_ID=self.project.project.id,
OS_PASSWORD=self.project.password)
+ if not new_env["OS_AUTH_URL"].endswith(('v3', 'v3/')):
+ new_env["OS_AUTH_URL"] = "{}/v3".format(new_env["OS_AUTH_URL"])
try:
del new_env['OS_TENANT_NAME']
del new_env['OS_TENANT_ID']
cmd = ['vmtp', '-d', '--json', '{}/vmtp.json'.format(self.res_dir),
'-c', self.config]
output = subprocess.check_output(
- cmd, stderr=subprocess.STDOUT, env=new_env)
+ cmd, stderr=subprocess.STDOUT, env=new_env).decode("utf-8")
self.__logger.info("%s\n%s", " ".join(cmd), output)
cmd = ['vmtp_genchart', '-c', '{}/vmtp.html'.format(self.res_dir),
'{}/vmtp.json'.format(self.res_dir)]
- output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
+ output = subprocess.check_output(
+ cmd, stderr=subprocess.STDOUT).decode("utf-8")
self.__logger.info("%s\n%s", " ".join(cmd), output)
with open('{}/vmtp.json'.format(self.res_dir), 'r') as res_file:
self.details = json.load(res_file)
status = testcase.TestCase.EX_OK
except subprocess.CalledProcessError as cpe:
self.__logger.error(
- "Exception when calling %s\n%s", cpe.cmd, cpe.output)
+ "Exception when calling %s\n%s", cpe.cmd,
+ cpe.output.decode("utf-8"))
self.result = 0
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot run vmtp")