logger = logging.getLogger('create_image')
-DEFAULT_METADATA = {'hw:mem_page_size': 'any'}
+MEM_PAGE_SIZE_ANY = {'hw:mem_page_size': 'any'}
+MEM_PAGE_SIZE_LARGE = {'hw:mem_page_size': 'large'}
class OpenStackFlavor:
"""
def __init__(self, config=None, name=None, flavor_id='auto', ram=None, disk=None, vcpus=None, ephemeral=0, swap=0,
- rxtx_factor=1.0, is_public=True, metadata=DEFAULT_METADATA):
+ rxtx_factor=1.0, is_public=True, metadata=None):
"""
Constructor
:param config: dict() object containing the configuration settings using the attribute names below as each
import uuid
import unittest
-from snaps.openstack import create_flavor
from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
from snaps.openstack.utils import nova_utils
self.assertEquals(0, settings.swap)
self.assertEquals(1.0, settings.rxtx_factor)
self.assertEquals(True, settings.is_public)
- self.assertEquals(create_flavor.DEFAULT_METADATA, settings.metadata)
+ self.assertEquals(None, settings.metadata)
def test_config_with_name_ram_disk_vcpus_only(self):
settings = FlavorSettings(config={'name': 'foo', 'ram': 1, 'disk': 2, 'vcpus': 3})
self.assertEquals(0, settings.swap)
self.assertEquals(1.0, settings.rxtx_factor)
self.assertEquals(True, settings.is_public)
- self.assertEquals(create_flavor.DEFAULT_METADATA, settings.metadata)
+ self.assertEquals(None, settings.metadata)
def test_all(self):
metadata = {'foo': 'bar'}
from snaps.openstack.create_security_group import SecurityGroupSettings, OpenStackSecurityGroup
from snaps.openstack.tests import openstack_tests, validation_utils
from snaps.openstack.utils import nova_utils
-from snaps.openstack.tests.os_source_file_test import OSComponentTestCase, OSIntegrationTestCase
+from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
__author__ = 'spisarski'
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=1024, disk=10, vcpus=1))
+ FlavorSettings(name=guid + '-flavor-name', ram=1024, disk=10, vcpus=1, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
self.tearDown()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Network
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
self.keypair_creator = OpenStackKeypair(
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
except Exception as e:
self.tearDown()
self.inst_creator.create()
-class CreateInstanceOnComputeHost(OSComponentTestCase):
+class CreateInstanceOnComputeHost(OSIntegrationTestCase):
"""
Test for the CreateInstance where one VM is deployed to each compute node
"""
Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
within OpenStack
"""
+ super(self.__class__, self).__start__()
+
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.port_base_name = guid + 'port'
try:
# Create Network
- self.network_creator = OpenStackNetwork(self.os_creds, self.priv_net_config.network_settings)
+ self.network_creator = OpenStackNetwork(self.admin_os_creds, self.priv_net_config.network_settings)
self.network_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
- self.os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=512, disk=1, vcpus=1))
+ self.admin_os_creds,
+ FlavorSettings(name=guid + '-flavor-name', ram=512, disk=1, vcpus=1, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Image
- self.image_creator = OpenStackImage(self.os_creds, self.os_image_settings)
+ self.image_creator = OpenStackImage(self.admin_os_creds, self.os_image_settings)
self.image_creator.create()
except Exception as e:
except Exception as e:
logger.error('Unexpected exception cleaning image with message - ' + e.message)
+ super(self.__class__, self).__clean__()
+
def test_deploy_vm_to_each_compute_node(self):
"""
Tests the creation of OpenStack VM instances to each compute node.
"""
from snaps.openstack.utils import nova_utils
- nova = nova_utils.nova_client(self.os_creds)
+ nova = nova_utils.nova_client(self.admin_os_creds)
zones = nova_utils.get_nova_availability_zones(nova)
# Create Instance on each server/zone
name=inst_name, flavor=self.flavor_creator.flavor_settings.name, availability_zone=zone,
port_settings=[port_settings])
inst_creator = OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings)
+ self.admin_os_creds, instance_settings, self.image_creator.image_settings)
self.inst_creators.append(inst_creator)
inst_creator.create()
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=self.guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
+ metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Keypair
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=self.guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=self.guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
+ metadata=self.flavor_metadata))
self.flavor_creator.create()
self.port_settings = PortSettings(name=self.guid + '-port',
try:
# Create Images
# Create the kernel image
- kernel_image_settings = openstack_tests.cirros_url_image(name=self.image_name+'_kernel',
+ kernel_image_settings = openstack_tests.cirros_url_image(
+ name=self.image_name+'_kernel',
url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-kernel')
self.image_creators.append(OpenStackImage(self.os_creds, kernel_image_settings))
kernel_image = self.image_creators[-1].create()
# Create the ramdisk image
- ramdisk_image_settings = openstack_tests.cirros_url_image(name=self.image_name+'_ramdisk',
+ ramdisk_image_settings = openstack_tests.cirros_url_image(
+ name=self.image_name+'_ramdisk',
url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-initramfs')
self.image_creators.append(OpenStackImage(self.os_creds, ramdisk_image_settings))
ramdisk_image = self.image_creators[-1].create()
self.assertIsNotNone(ramdisk_image)
# Create the main image
- os_image_settings = openstack_tests.cirros_url_image(name=self.image_name,
+ os_image_settings = openstack_tests.cirros_url_image(
+ name=self.image_name,
url='http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-x86_64-disk.img')
- properties = {}
+ properties = dict()
properties['kernel_id'] = kernel_image.id
properties['ramdisk_id'] = ramdisk_image.id
os_image_settings.extra_properties = properties
self.image_creators.append(OpenStackImage(self.os_creds, os_image_settings))
created_image = self.image_creators[-1].create()
+ self.assertIsNotNone(created_image)
# Create Flavor
self.flavor_creator = OpenStackFlavor(
self.admin_os_creds,
- FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2, metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Network
# Exception should not be thrown
self.inst_creator.clean()
-
import openstack_tests
import logging
-# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-# To run these tests from an IDE, the CWD must be set to the python directory of this project
-# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
from snaps.openstack.create_project import ProjectSettings
from snaps.openstack.create_user import UserSettings
from snaps.openstack.utils import deploy_utils, keystone_utils
dev_os_env_file = 'openstack/tests/conf/os_env.yaml'
+# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+# To run these tests from an IDE, the CWD must be set to the snaps directory of this project
+# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+
class OSComponentTestCase(unittest.TestCase):
Super for test classes requiring a connection to OpenStack
"""
def __init__(self, method_name='runTest', os_env_file=None, ext_net_name=None, http_proxy_str=None,
- ssh_proxy_cmd=None, use_keystone=False, log_level=logging.DEBUG):
+ ssh_proxy_cmd=None, use_keystone=False, flavor_metadata=None, log_level=logging.DEBUG):
super(OSIntegrationTestCase, self).__init__(method_name=method_name, os_env_file=os_env_file,
ext_net_name=ext_net_name, http_proxy_str=http_proxy_str,
ssh_proxy_cmd=ssh_proxy_cmd, log_level=log_level)
self.use_keystone = use_keystone
self.keystone = None
+ self.flavor_metadata = flavor_metadata
@staticmethod
def parameterize(testcase_klass, os_env_file, ext_net_name, http_proxy_str=None, ssh_proxy_cmd=None,
- use_keystone=False, log_level=logging.DEBUG):
+ use_keystone=False, flavor_metadata=None, log_level=logging.DEBUG):
""" Create a suite containing all tests taken from the given
subclass, passing them the parameter 'param'.
"""
suite = unittest.TestSuite()
for name in test_names:
suite.addTest(testcase_klass(name, os_env_file, ext_net_name, http_proxy_str, ssh_proxy_cmd, use_keystone,
- log_level))
+ flavor_metadata, log_level))
return suite
"""
# Create Flavor
self.flavor_creator = create_flavor.OpenStackFlavor(
self.admin_os_creds,
- create_flavor.FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2))
+ create_flavor.FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
+ metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Key/Pair
def add_openstack_integration_tests(suite, source_filename, ext_net_name, proxy_settings=None, ssh_proxy_cmd=None,
- use_keystone=True, use_floating_ips=True, log_level=logging.INFO):
+ use_keystone=True, flavor_metadata=None, use_floating_ips=True,
+ log_level=logging.INFO):
"""
Adds tests written to exercise all long-running OpenStack integration tests meaning they will be creating VM
instances and potentially performing some SSH functions through floating IPs
:param ssh_proxy_cmd: the command your environment requires for creating ssh connections through a proxy
:param use_keystone: when True, tests requiring direct access to Keystone are added as these need to be running on
a host that has access to the cloud's private network
+ :param flavor_metadata: dict() object containing the metadata required by your flavor based on your configuration:
+ (i.e. {'hw:mem_page_size': 'large'})
:param use_floating_ips: when true, all tests requiring Floating IPs will be added to the suite
:param log_level: the logging level
:return: None as the tests will be adding to the 'suite' parameter object
# Creator Object tests
suite.addTest(OSIntegrationTestCase.parameterize(CreateSecurityGroupTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateImageSuccessTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateImageNegativeTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateMultiPartImageTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateKeypairsTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateNetworkSuccessTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateRouterSuccessTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateRouterNegativeTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
# VM Instances
suite.addTest(OSIntegrationTestCase.parameterize(SimpleHealthCheck, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateInstanceSimpleTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateInstancePortManipulationTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(InstanceSecurityGroupTests, source_filename, ext_net_name,
http_proxy_str=proxy_settings, use_keystone=use_keystone,
- log_level=log_level))
- suite.addTest(OSComponentTestCase.parameterize(CreateInstanceOnComputeHost, source_filename, ext_net_name,
- http_proxy_str=proxy_settings, log_level=log_level))
- suite.addTest(OSComponentTestCase.parameterize(CreateInstanceFromThreePartImage, source_filename, ext_net_name,
- http_proxy_str=proxy_settings, log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(CreateInstanceOnComputeHost, source_filename, ext_net_name,
+ http_proxy_str=proxy_settings, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(CreateInstanceFromThreePartImage, source_filename, ext_net_name,
+ http_proxy_str=proxy_settings, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, log_level=log_level))
if use_floating_ips:
suite.addTest(OSIntegrationTestCase.parameterize(CreateInstanceSingleNetworkTests, source_filename,
ext_net_name, http_proxy_str=proxy_settings,
ssh_proxy_cmd=ssh_proxy_cmd, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(CreateInstancePubPrivNetTests, source_filename,
ext_net_name, http_proxy_str=proxy_settings,
ssh_proxy_cmd=ssh_proxy_cmd, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(AnsibleProvisioningTests, source_filename,
ext_net_name, http_proxy_str=proxy_settings,
ssh_proxy_cmd=ssh_proxy_cmd, use_keystone=use_keystone,
- log_level=log_level))
+ flavor_metadata=flavor_metadata, log_level=log_level))
def add_openstack_staging_tests(suite, source_filename, ext_net_name, proxy_settings=None, log_level=logging.INFO):
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
+import json
import logging
-import unittest
import os
+import unittest
from snaps import test_suite_builder
'INFO': logging.INFO, 'DEBUG': logging.DEBUG}
-def __create_test_suite(source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd, run_unit_tests, use_keystone,
- use_floating_ips, log_level):
+def __create_test_suite(source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd, run_unit_tests, flavor_metadata,
+ use_keystone, use_floating_ips, log_level):
"""
Compiles the tests that should run
:param source_filename: the OpenStack credentials file (required)
:param run_unit_tests: when true, the tests not requiring OpenStack will be added to the test suite
:param proxy_settings: <host>:<port> of the proxy server (optional)
:param ssh_proxy_cmd: the command used to connect via SSH over some proxy server (optional)
+ :param flavor_metadata: dict() object containing the metadata for flavors created for test VM instance
:param use_keystone: when true, tests creating users and projects will be exercised and must be run on a host that
has access to the cloud's administrative network
:param use_floating_ips: when true, tests requiring floating IPs will be executed
# Long running integration type tests
test_suite_builder.add_openstack_integration_tests(suite, source_filename, ext_net_name, use_keystone=use_keystone,
proxy_settings=proxy_settings, ssh_proxy_cmd=ssh_proxy_cmd,
+ flavor_metadata=flavor_metadata,
use_floating_ips=use_floating_ips, log_level=log_level)
return suite
log_level = LOG_LEVELS.get(arguments.log_level, logging.DEBUG)
+ flavor_metadata = None
+
+ if arguments.flavor_metadata:
+ flavor_metadata = json.loads(arguments.flavor_metadata)
+
suite = None
if arguments.env and arguments.ext_net:
suite = __create_test_suite(arguments.env, arguments.ext_net, arguments.proxy, arguments.ssh_proxy_cmd,
arguments.include_units != ARG_NOT_SET,
+ flavor_metadata,
arguments.use_keystone != ARG_NOT_SET,
arguments.no_floating_ips == ARG_NOT_SET, log_level)
else:
help='When argument is set, all tests requiring Floating IPs will not be executed')
parser.add_argument('-u', '--include-units', dest='include_units', default=ARG_NOT_SET, nargs='?',
help='When argument is set, all tests not requiring OpenStack will be executed')
+ parser.add_argument('-fm', '--flavor-meta', dest='flavor_metadata', required=False,
+ default='{\"hw:mem_page_size\": \"any\"}',
+ help='JSON string to be used as flavor metadata for all test instances created')
args = parser.parse_args()
main(args)
Code Review / snaps.git / commitdiff