from xtesting.core import testcase
-from functest.opnfv_tests.openstack.vping import vping_base
+from functest.core import singlevm
from functest.utils import config
-class VPingUserdata(vping_base.VPingBase):
+class VPingUserdata(singlevm.VmReady2):
"""
Class to execute the vPing test using userdata and the VM's console
"""
kwargs["case_name"] = "vping_userdata"
super(VPingUserdata, self).__init__(**kwargs)
self.logger = logging.getLogger(__name__)
+ self.vm1 = None
self.vm2 = None
def run(self, **kwargs):
"""
try:
assert self.cloud
- super(VPingUserdata, self).run()
-
- vm2_name = "{}-{}-{}".format(
- getattr(config.CONF, 'vping_vm_name_2'), "userdata", self.guid)
- self.logger.info(
- "Creating VM 2 instance with name: '%s'", vm2_name)
- self.vm2 = self.cloud.create_server(
- vm2_name, image=self.image.id, flavor=self.flavor.id,
- auto_ip=False, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_boot_timeout'),
- network=self.network.id,
+ assert super(VPingUserdata, self).run(
+ **kwargs) == testcase.TestCase.EX_OK
+ self.result = 0
+ self.vm1 = self.boot_vm()
+ self.vm2 = self.boot_vm(
+ '{}-vm2_{}'.format(self.case_name, self.guid),
userdata=self._get_userdata())
- self.logger.debug("vm2: %s", self.vm2)
self.vm2 = self.cloud.wait_for_server(self.vm2, auto_ip=False)
- p_console = self.cloud.get_server_console(self.vm1.id)
- self.logger.debug("vm2 console: \n%s", p_console)
- return self._execute()
+ result = self._do_vping()
+ self.stop_time = time.time()
+ if result != testcase.TestCase.EX_OK:
+ return testcase.TestCase.EX_RUN_ERROR
+ self.result = 100
+ return testcase.TestCase.EX_OK
except Exception: # pylint: disable=broad-except
self.logger.exception('Unexpected error running vping_userdata')
return testcase.TestCase.EX_RUN_ERROR
"""
Override from super
"""
- if not self.vm1.private_v4:
+ if not (self.vm1.private_v4 or self.vm1.addresses[
+ self.network.name][0].addr):
self.logger.error("vm1: IP addr missing")
return testcase.TestCase.EX_TESTCASE_FAILED
else:
self.logger.debug(
"Pinging %s. Waiting for response...",
- self.vm1.private_v4)
+ self.vm1.private_v4 or self.vm1.addresses[
+ self.network.name][0].addr)
sec += 1
return exit_code
:param test_ip: the IP value to substitute into the script
:return: the bash script contents
"""
- if self.vm1.private_v4:
+ if self.vm1.private_v4 or self.vm1.addresses[
+ self.network.name][0].addr:
return ("#!/bin/sh\n\n"
"while true; do\n"
" ping -c 1 %s 2>&1 >/dev/null\n"
" echo 'vPing KO'\n"
" fi\n"
" sleep 1\n"
- "done\n" % str(self.vm1.private_v4))
+ "done\n" % str(
+ self.vm1.private_v4 or self.vm1.addresses[
+ self.network.name][0].addr))
return None
def clean(self):
assert self.cloud
- self.cloud.delete_server(
- self.vm2, wait=True,
- timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
- self.cloud.delete_server(self.vm2, wait=True)
+ if self.vm1:
+ self.cloud.delete_server(
+ self.vm1, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
+ if self.vm2:
+ self.cloud.delete_server(
+ self.vm2, wait=True,
+ timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
super(VPingUserdata, self).clean()