3 # Copyright (c) 2015 All rights reserved
4 # This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 from datetime import datetime
16 from functest.core.testcase import TestCase
17 from functest.utils import functest_utils
18 from functest.utils.constants import CONST
20 from snaps.openstack import create_flavor
21 from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
22 from snaps.openstack.create_network import NetworkSettings, SubnetSettings
23 from snaps.openstack.tests import openstack_tests
24 from snaps.openstack.utils import deploy_utils, nova_utils
27 class VPingBase(TestCase):
30 Base class for vPing tests that check connectivity between two VMs shared
32 This class is responsible for creating the image, internal network.
34 def __init__(self, **kwargs):
35 super(VPingBase, self).__init__(**kwargs)
37 self.logger = logging.getLogger(self.__class__.__name__)
39 self.functest_repo = CONST.__getattribute__('dir_repo_functest')
41 if CONST.__getattribute__('vping_unique_names'):
42 self.guid = '-' + str(uuid.uuid4())
44 self.os_creds = openstack_tests.get_credentials(
45 os_env_file=CONST.__getattribute__('openstack_creds'))
47 self.repo = CONST.__getattribute__('dir_vping')
49 self.creators = list()
50 self.image_creator = None
51 self.network_creator = None
52 self.vm1_creator = None
53 self.vm2_creator = None
55 self.self_cleanup = CONST.__getattribute__('vping_cleanup_objects')
59 CONST.__getattribute__('vping_image_name') + self.guid
62 self.vm1_name = CONST.__getattribute__('vping_vm_name_1') + self.guid
63 self.vm2_name = CONST.__getattribute__('vping_vm_name_2') + self.guid
64 self.vm_boot_timeout = CONST.__getattribute__('vping_vm_boot_timeout')
65 self.vm_delete_timeout =\
66 CONST.__getattribute__('vping_vm_delete_timeout')
67 self.vm_ssh_connect_timeout = CONST.vping_vm_ssh_connect_timeout
68 self.ping_timeout = CONST.__getattribute__('vping_ping_timeout')
69 self.flavor_name = 'vping-flavor' + self.guid
71 # NEUTRON Private Network parameters
72 self.private_net_name =\
73 CONST.__getattribute__('vping_private_net_name') + self.guid
74 self.private_subnet_name =\
75 CONST.__getattribute__('vping_private_subnet_name') + self.guid
76 self.private_subnet_cidr =\
77 CONST.__getattribute__('vping_private_subnet_cidr')
79 scenario = functest_utils.get_scenario()
81 self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_ANY
82 if 'ovs' in scenario or 'fdio' in scenario:
83 self.flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
85 self.cirros_image_config = None
87 # Move this configuration option up for all tests to leverage
88 if hasattr(CONST, 'snaps_images_cirros'):
89 self.cirros_image_config = CONST.__getattribute__(
90 'snaps_images_cirros')
94 Begins the test execution which should originate from the subclass
97 if not os.path.exists(self.functest_repo):
99 "Functest repository not found '%s'" % self.functest_repo)
101 self.logger.info('Begin virtual environment setup')
103 self.start_time = time.time()
104 self.logger.info("vPing Start Time:'%s'" % (
105 datetime.fromtimestamp(self.start_time).strftime(
106 '%Y-%m-%d %H:%M:%S')))
108 self.__delete_exist_vms()
110 image_base_name = self.image_name + '-' + str(self.guid)
111 os_image_settings = openstack_tests.cirros_image_settings(
112 image_base_name, image_metadata=self.cirros_image_config)
113 self.logger.info("Creating image with name: '%s'" % self.image_name)
115 self.image_creator = deploy_utils.create_image(
116 self.os_creds, os_image_settings)
117 self.creators.append(self.image_creator)
120 "Creating network with name: '%s'" % self.private_net_name)
121 self.network_creator = deploy_utils.create_network(
123 NetworkSettings(name=self.private_net_name,
124 subnet_settings=[SubnetSettings(
125 name=self.private_subnet_name,
126 cidr=self.private_subnet_cidr)]))
127 self.creators.append(self.network_creator)
130 "Creating flavor with name: '%s'" % self.flavor_name)
131 flavor_creator = OpenStackFlavor(
133 FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
134 metadata=self.flavor_metadata))
135 flavor_creator.create()
136 self.creators.append(flavor_creator)
140 Method called by subclasses after environment has been setup
141 :return: the exit code
143 self.logger.info('Begin test execution')
145 test_ip = self.vm1_creator.get_port_ip(
146 self.vm1_creator.instance_settings.port_settings[0].name)
148 if self.vm1_creator.vm_active(
149 block=True) and self.vm2_creator.vm_active(block=True):
150 result = self._do_vping(self.vm2_creator, test_ip)
152 raise Exception('VMs never became active')
154 if result == TestCase.EX_RUN_ERROR:
155 return TestCase.EX_RUN_ERROR
157 self.stop_time = time.time()
159 return TestCase.EX_OK
163 Cleanup all OpenStack objects. Should be called on completion
166 if self.self_cleanup:
167 for creator in reversed(self.creators):
170 except Exception as e:
171 self.logger.error('Unexpected error cleaning - %s', e)
173 def _do_vping(self, vm_creator, test_ip):
175 Method to be implemented by subclasses
176 Begins the real test after the OpenStack environment has been setup
177 :param vm_creator: the SNAPS VM instance creator object
178 :param test_ip: the IP to which the VM needs to issue the ping
181 raise NotImplementedError('vping execution is not implemented')
183 def __delete_exist_vms(self):
185 Cleans any existing VMs using the same name.
187 nova_client = nova_utils.nova_client(self.os_creds)
188 servers = nova_client.servers.list()
189 for server in servers:
190 if server.name == self.vm1_name or server.name == self.vm2_name:
191 self.logger.info("Deleting instance %s..." % server.name)
195 class VPingMain(object):
196 def __init__(self, vping_cls):
197 self.vping = vping_cls()
199 def main(self, **kwargs):
201 result = self.vping.run(**kwargs)
202 if result != VPingBase.EX_OK:
205 return self.vping.publish_report()
207 return VPingBase.EX_RUN_ERROR