# pylint: disable=too-many-instance-attributes
__logger = logging.getLogger(__name__)
- filename = '/home/opnfv/functest/images/cirros-0.4.0-x86_64-disk.img'
+ filename = '/home/opnfv/functest/images/cirros-0.6.1-x86_64-disk.img'
image_format = 'qcow2'
extra_properties = {}
filename_alt = filename
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = 'vmready1'
- super(VmReady1, self).__init__(**kwargs)
+ super().__init__(**kwargs)
self.image = None
self.flavor = None
functest_utils.convert_ini_to_dict(
env.get('IMAGE_PROPERTIES')))
extra_properties.update(
- getattr(config.CONF, '{}_extra_properties'.format(
- self.case_name), {}))
+ getattr(config.CONF, f'{self.case_name}_extra_properties', {}))
image = self.cloud.create_image(
- name if name else '{}-img_{}'.format(self.case_name, self.guid),
+ name if name else f'{self.case_name}-img_{self.guid}',
filename=getattr(
- config.CONF, '{}_image'.format(self.case_name),
+ config.CONF, f'{self.case_name}_image',
self.filename),
meta=extra_properties,
disk_format=getattr(
- config.CONF, '{}_image_format'.format(self.case_name),
+ config.CONF, f'{self.case_name}_image_format',
self.image_format),
visibility=getattr(
- config.CONF, '{}_visibility'.format(self.case_name),
+ config.CONF, f'{self.case_name}_visibility',
self.visibility),
wait=True)
self.__logger.debug("image: %s", image)
functest_utils.convert_ini_to_dict(
env.get('IMAGE_PROPERTIES')))
extra_alt_properties.update(
- getattr(config.CONF, '{}_extra_alt_properties'.format(
- self.case_name), {}))
+ getattr(config.CONF, f'{self.case_name}_extra_alt_properties', {}))
image = self.cloud.create_image(
- name if name else '{}-img_alt_{}'.format(
- self.case_name, self.guid),
+ name if name else f'{self.case_name}-img_alt_{self.guid}',
filename=getattr(
- config.CONF, '{}_image_alt'.format(self.case_name),
+ config.CONF, f'{self.case_name}_image_alt',
self.filename_alt),
meta=extra_alt_properties,
disk_format=getattr(
- config.CONF, '{}_image_alt_format'.format(self.case_name),
+ config.CONF, f'{self.case_name}_image_alt_format',
self.image_format),
visibility=getattr(
- config.CONF, '{}_visibility'.format(self.case_name),
+ config.CONF, f'{self.case_name}_visibility',
self.visibility),
wait=True)
self.__logger.debug("image: %s", image)
"""
assert self.orig_cloud
flavor = self.orig_cloud.create_flavor(
- name if name else '{}-flavor_{}'.format(self.case_name, self.guid),
- getattr(config.CONF, '{}_flavor_ram'.format(self.case_name),
+ name if name else f'{self.case_name}-flavor_{self.guid}',
+ getattr(config.CONF, f'{self.case_name}_flavor_ram',
self.flavor_ram),
- getattr(config.CONF, '{}_flavor_vcpus'.format(self.case_name),
+ getattr(config.CONF, f'{self.case_name}_flavor_vcpus',
self.flavor_vcpus),
- getattr(config.CONF, '{}_flavor_disk'.format(self.case_name),
+ getattr(config.CONF, f'{self.case_name}_flavor_disk',
self.flavor_disk))
self.__logger.debug("flavor: %s", flavor)
flavor_extra_specs = self.flavor_extra_specs.copy()
env.get('FLAVOR_EXTRA_SPECS')))
flavor_extra_specs.update(
getattr(config.CONF,
- '{}_flavor_extra_specs'.format(self.case_name), {}))
+ f'{self.case_name}_flavor_extra_specs', {}))
self.orig_cloud.set_flavor_specs(flavor.id, flavor_extra_specs)
return flavor
"""
assert self.orig_cloud
flavor = self.orig_cloud.create_flavor(
- name if name else '{}-flavor_alt_{}'.format(
- self.case_name, self.guid),
- getattr(config.CONF, '{}_flavor_alt_ram'.format(self.case_name),
+ name if name else f'{self.case_name}-flavor_alt_{self.guid}',
+ getattr(config.CONF, f'{self.case_name}_flavor_alt_ram',
self.flavor_alt_ram),
- getattr(config.CONF, '{}_flavor_alt_vcpus'.format(self.case_name),
+ getattr(config.CONF, f'{self.case_name}_flavor_alt_vcpus',
self.flavor_alt_vcpus),
- getattr(config.CONF, '{}_flavor_alt_disk'.format(self.case_name),
+ getattr(config.CONF, f'{self.case_name}_flavor_alt_disk',
self.flavor_alt_disk))
self.__logger.debug("flavor: %s", flavor)
flavor_alt_extra_specs = self.flavor_alt_extra_specs.copy()
env.get('FLAVOR_EXTRA_SPECS')))
flavor_alt_extra_specs.update(
getattr(config.CONF,
- '{}_flavor_alt_extra_specs'.format(self.case_name), {}))
+ f'{self.case_name}_flavor_alt_extra_specs', {}))
self.orig_cloud.set_flavor_specs(
flavor.id, flavor_alt_extra_specs)
return flavor
"""
assert self.cloud
vm1 = self.cloud.create_server(
- name if name else '{}-vm_{}'.format(self.case_name, self.guid),
+ name if name else f'{self.case_name}-vm_{self.guid}',
image=self.image.id, flavor=self.flavor.id,
- auto_ip=False, network=self.network.id,
+ auto_ip=False,
+ network=self.network.id if self.network else env.get(
+ "EXTERNAL_NETWORK"),
timeout=self.create_server_timeout, wait=True, **kwargs)
self.__logger.debug("vm: %s", vm1)
return vm1
- def check_regex_in_console(self, name, regex=' login: ', loop=1):
+ def check_regex_in_console(self, name, regex=' login: ', loop=6):
"""Wait for specific message in console
Returns: True or False on errors
status = testcase.TestCase.EX_RUN_ERROR
try:
assert self.cloud
- assert super(VmReady1, self).run(
+ assert super().run(
**kwargs) == testcase.TestCase.EX_OK
self.image = self.publish_image()
self.flavor = self.create_flavor()
try:
assert self.orig_cloud
assert self.cloud
- super(VmReady1, self).clean()
+ super().clean()
if self.image:
self.cloud.delete_image(self.image.id)
if self.flavor:
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = 'vmready2'
- super(VmReady2, self).__init__(**kwargs)
+ super().__init__(**kwargs)
try:
assert self.orig_cloud
self.project = tenantnetwork.NewProject(
def clean(self):
try:
- super(VmReady2, self).clean()
+ super().clean()
assert self.project
self.project.clean()
except Exception: # pylint: disable=broad-except
ssh_connect_timeout = 1
ssh_connect_loops = 6
create_floating_ip_timeout = 120
+ check_console_loop = 6
+ check_console_regex = ' login: '
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = 'singlevm1'
- super(SingleVm1, self).__init__(**kwargs)
+ super().__init__(**kwargs)
self.sshvm = None
self.sec = None
self.fip = None
"""
assert self.cloud
self.keypair = self.cloud.create_keypair(
- '{}-kp_{}'.format(self.case_name, self.guid))
+ f'{self.case_name}-kp_{self.guid}')
self.__logger.debug("keypair: %s", self.keypair)
self.__logger.debug("private_key:\n%s", self.keypair.private_key)
- with open(self.key_filename, 'w') as private_key_file:
+ with open(
+ self.key_filename, 'w', encoding='utf-8') as private_key_file:
private_key_file.write(self.keypair.private_key)
self.sec = self.cloud.create_security_group(
- '{}-sg_{}'.format(self.case_name, self.guid),
- 'created by OPNFV Functest ({})'.format(self.case_name))
+ f'{self.case_name}-sg_{self.guid}',
+ f'created by OPNFV Functest ({self.case_name})')
self.cloud.create_security_group_rule(
self.sec.id, port_range_min='22', port_range_max='22',
protocol='tcp', direction='ingress')
- None on error
"""
assert vm1
- fip = self.cloud.create_floating_ip(
- network=self.ext_net.id, server=vm1, wait=True,
- timeout=self.create_floating_ip_timeout)
- self.__logger.debug("floating_ip: %s", fip)
+ fip = None
+ if env.get('NO_TENANT_NETWORK').lower() != 'true':
+ fip = self.cloud.create_floating_ip(
+ network=self.ext_net.id, server=vm1, wait=True,
+ timeout=self.create_floating_ip_timeout)
+ self.__logger.debug("floating_ip: %s", fip)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.client.AutoAddPolicy())
for loop in range(self.ssh_connect_loops):
p_console = self.cloud.get_server_console(vm1)
self.__logger.debug("vm console: \n%s", p_console)
ssh.connect(
- fip.floating_ip_address,
+ fip.floating_ip_address if fip else vm1.public_v4,
username=getattr(
config.CONF,
- '{}_image_user'.format(self.case_name), self.username),
+ f'{self.case_name}_image_user', self.username),
key_filename=self.key_filename,
timeout=getattr(
config.CONF,
- '{}_vm_ssh_connect_timeout'.format(self.case_name),
+ f'{self.case_name}_vm_ssh_connect_timeout',
self.ssh_connect_timeout))
break
except Exception as exc: # pylint: disable=broad-except
self.__logger.debug(
"try %s: cannot connect to %s: %s", loop + 1,
- fip.floating_ip_address, exc)
+ fip.floating_ip_address if fip else vm1.public_v4, exc)
time.sleep(9)
else:
self.__logger.error(
Returns: echo exit codes
"""
(_, stdout, stderr) = self.ssh.exec_command('echo Hello World')
- self.__logger.debug("output:\n%s", stdout.read())
- self.__logger.debug("error:\n%s", stderr.read())
+ self.__logger.debug("output:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("error:\n%s", stderr.read().decode("utf-8"))
return stdout.channel.recv_exit_status()
def run(self, **kwargs):
status = testcase.TestCase.EX_RUN_ERROR
try:
assert self.cloud
- assert super(SingleVm1, self).run(
+ assert super().run(
**kwargs) == testcase.TestCase.EX_OK
self.result = 0
self.prepare()
self.sshvm = self.boot_vm(
key_name=self.keypair.id, security_groups=[self.sec.id])
- (self.fip, self.ssh) = self.connect(self.sshvm)
- if not self.execute():
- self.result = 100
- status = testcase.TestCase.EX_OK
+ if self.check_regex_in_console(
+ self.sshvm.name, regex=self.check_console_regex,
+ loop=self.check_console_loop):
+ (self.fip, self.ssh) = self.connect(self.sshvm)
+ if not self.execute():
+ self.result = 100
+ status = testcase.TestCase.EX_OK
except Exception: # pylint: disable=broad-except
self.__logger.exception('Cannot run %s', self.case_name)
finally:
self.cloud.delete_security_group(self.sec.id)
if self.keypair:
self.cloud.delete_keypair(self.keypair.name)
- super(SingleVm1, self).clean()
+ super().clean()
except Exception: # pylint: disable=broad-except
self.__logger.exception("Cannot clean all resources")
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = 'singlevm2'
- super(SingleVm2, self).__init__(**kwargs)
+ super().__init__(**kwargs)
try:
assert self.orig_cloud
self.project = tenantnetwork.NewProject(
def clean(self):
try:
- super(SingleVm2, self).clean()
+ super().clean()
assert self.project
self.project.clean()
except Exception: # pylint: disable=broad-except
Code Review / functest.git / blobdiff