-# Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs")
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
# and others. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# See the License for the specific language governing permissions and
# limitations under the License.
-import os
import uuid
+import os
+import pkg_resources
+from scp import SCPClient
+
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import VmInstanceConfig, FloatingIpConfig
+
+from snaps.openstack import create_flavor
+from snaps.openstack import create_image
from snaps.openstack import create_instance
from snaps.openstack import create_keypairs
from snaps.openstack import create_network
from snaps.openstack import create_router
-from snaps.openstack import create_image
-from snaps.openstack import create_flavor
-from scp import SCPClient
-
-from snaps.provisioning import ansible_utils
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
from snaps.openstack.tests import openstack_tests
+from snaps.openstack.tests.create_instance_tests import check_dhcp_lease
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
+from snaps.openstack.utils import nova_utils
+from snaps.provisioning import ansible_utils
VM_BOOT_TIMEOUT = 600
class AnsibleProvisioningTests(OSIntegrationTestCase):
"""
- Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP and eth1 w/o
+ Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP
+ and eth1 w/o
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
+ self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
+
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
# Setup members to cleanup just in case they don't get created
self.inst_creator = None
self.keypair_creator = None
+ self.sec_grp_creator = None
self.flavor_creator = None
self.router_creator = None
self.network_creator = None
- self.image_creators = list()
+ self.image_creator = None
try:
# Create Image
- os_image_settings = openstack_tests.ubuntu_url_image(name=guid + '-' + '-image')
- if self.image_metadata:
- if 'disk_url' in self.image_metadata and self.image_metadata['disk_url']:
- os_image_settings.url = self.image_metadata['disk_url']
- if 'extra_properties' in self.image_metadata and self.image_metadata['extra_properties']:
- os_image_settings.extra_properties = self.image_metadata['extra_properties']
-
- # If this is a 3-part image create the kernel and ramdisk images first
- if self.image_metadata:
- if 'kernel_url' in self.image_metadata and self.image_metadata['kernel_url']:
- kernel_image_settings = openstack_tests.cirros_url_image(
- name=os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_image_settings))
- kernel_image = self.image_creators[-1].create()
- os_image_settings.extra_properties['kernel_id'] = kernel_image.id
-
- if 'ramdisk_url' in self.image_metadata and self.image_metadata['ramdisk_url']:
- ramdisk_image_settings = openstack_tests.cirros_url_image(
- name=os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_image_settings))
- ramdisk_image = self.image_creators[-1].create()
- os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
-
- self.image_creators.append(create_image.OpenStackImage(self.os_creds, os_image_settings))
- self.image_creators[-1].create()
+ os_image_settings = openstack_tests.ubuntu_image_settings(
+ name=guid + '-' + '-image',
+ image_metadata=self.image_metadata)
+ self.image_creator = create_image.OpenStackImage(
+ self.os_creds, os_image_settings)
+ self.image_creator.create()
# First network is public
self.pub_net_config = openstack_tests.get_pub_net_config(
- net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
- router_name=guid + '-pub-router', external_net=self.ext_net_name)
-
- self.network_creator = create_network.OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings)
+ project_name=self.os_creds.project_name,
+ net_name=guid + '-pub-net',
+ mtu=1442, subnet_name=guid + '-pub-subnet',
+ router_name=guid + '-pub-router',
+ external_net=self.ext_net_name)
+
+ self.network_creator = create_network.OpenStackNetwork(
+ self.os_creds, self.pub_net_config.network_settings)
self.network_creator.create()
# Create routers
- self.router_creator = create_router.OpenStackRouter(self.os_creds, self.pub_net_config.router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, self.pub_net_config.router_settings)
self.router_creator.create()
# Create Flavor
+ flavor_config = openstack_tests.get_flavor_config(
+ name=guid + '-flavor-name', ram=2048, disk=10,
+ vcpus=2, metadata=self.flavor_metadata)
+
self.flavor_creator = create_flavor.OpenStackFlavor(
- self.admin_os_creds,
- create_flavor.FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
- metadata=self.flavor_metadata))
+ self.admin_os_creds, flavor_config)
self.flavor_creator.create()
# Create Key/Pair
self.keypair_creator = create_keypairs.OpenStackKeypair(
- self.os_creds, create_keypairs.KeypairSettings(
- name=self.keypair_name, public_filepath=self.keypair_pub_filepath,
+ self.os_creds, KeypairConfig(
+ name=self.keypair_name,
+ public_filepath=self.keypair_pub_filepath,
private_filepath=self.keypair_priv_filepath))
self.keypair_creator.create()
+ # Create Security Group
+ sec_grp_name = guid + '-sec-grp'
+ rule1 = SecurityGroupRuleConfig(
+ sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ protocol=Protocol.icmp)
+ rule2 = SecurityGroupRuleConfig(
+ sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
+ self.sec_grp_creator = OpenStackSecurityGroup(
+ self.os_creds,
+ SecurityGroupConfig(
+ name=sec_grp_name, rule_settings=[rule1, rule2]))
+ self.sec_grp_creator.create()
+
# Create instance
ports_settings = list()
ports_settings.append(
- create_network.PortSettings(name=self.port_1_name,
- network_name=self.pub_net_config.network_settings.name))
-
- instance_settings = create_instance.VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings,
- floating_ip_settings=[create_instance.FloatingIpSettings(
+ PortConfig(
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name))
+
+ instance_settings = VmInstanceConfig(
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=ports_settings,
+ floating_ip_settings=[FloatingIpConfig(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
self.inst_creator = create_instance.OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creators[-1].image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
- except Exception as e:
+ except:
self.tearDown()
- raise Exception(e.message)
+ raise
def tearDown(self):
"""
Cleans the created objects
"""
if self.inst_creator:
- self.inst_creator.clean()
+ try:
+ self.inst_creator.clean()
+ except:
+ pass
+
+ if self.sec_grp_creator:
+ try:
+ self.sec_grp_creator.clean()
+ except:
+ pass
if self.keypair_creator:
- self.keypair_creator.clean()
+ try:
+ self.keypair_creator.clean()
+ except:
+ pass
if self.flavor_creator:
- self.flavor_creator.clean()
+ try:
+ self.flavor_creator.clean()
+ except:
+ pass
if os.path.isfile(self.keypair_pub_filepath):
- os.remove(self.keypair_pub_filepath)
+ try:
+ os.remove(self.keypair_pub_filepath)
+ except:
+ pass
if os.path.isfile(self.keypair_priv_filepath):
- os.remove(self.keypair_priv_filepath)
+ try:
+ os.remove(self.keypair_priv_filepath)
+ except:
+ pass
if self.router_creator:
- self.router_creator.clean()
+ try:
+ self.router_creator.clean()
+ except:
+ pass
if self.network_creator:
- self.network_creator.clean()
+ try:
+ self.network_creator.clean()
+ except:
+ pass
- if self.image_creators:
- while self.image_creators:
- self.image_creators[-1].clean()
- self.image_creators.pop()
+ if self.image_creator and not self.image_creator.image_settings.exists:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
if os.path.isfile(self.test_file_local_path):
os.remove(self.test_file_local_path)
def test_apply_simple_playbook(self):
"""
- Tests application of an Ansible playbook that simply copies over a file:
- 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
- 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
- Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
+ Tests application of an Ansible playbook that simply copies over a file
+ 1. Have a ~/.ansible.cfg (or alternate means) to
+ set host_key_checking = False
+ 2. Set the following environment variable in your executing shell:
+ ANSIBLE_HOST_KEY_CHECKING=False
+ Should this not be performed, the creation of the host ssh key will
+ cause your ansible calls to fail.
"""
self.inst_creator.create(block=True)
+ priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
+ self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip))
+
+ # Apply Security Group
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
+
# Block until VM's ssh port has been opened
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
+ # Block until cloud-init has completed
+ self.assertTrue(self.inst_creator.cloud_init_complete(block=True))
+
ssh_client = self.inst_creator.ssh_client()
self.assertIsNotNone(ssh_client)
- out = ssh_client.exec_command('pwd')[1].channel.in_buffer.read(1024)
- self.assertIsNotNone(out)
- self.assertGreater(len(out), 1)
- # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
+ try:
+ out = ssh_client.exec_command('pwd')[1].channel.in_buffer.read(
+ 1024)
+ self.assertIsNotNone(out)
+ self.assertGreater(len(out), 1)
+ finally:
+ ssh_client.close()
+
+ # Need to use the first floating IP as subsequent ones are currently
+ # broken with Apex CO
ip = self.inst_creator.get_floating_ip().ip
user = self.inst_creator.get_image_user()
priv_key = self.inst_creator.keypair_settings.private_filepath
- retval = ansible_utils.apply_playbook('provisioning/tests/playbooks/simple_playbook.yml', [ip], user, priv_key,
- proxy_setting=self.os_creds.proxy_settings)
- self.assertEquals(0, retval)
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks', 'simple_playbook.yml')
+ self.inst_creator.apply_ansible_playbook(relative_pb_path)
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(
+ ip, user, private_key_filepath=priv_key,
+ proxy_settings=self.os_creds.proxy_settings)
self.assertIsNotNone(ssh)
- scp = SCPClient(ssh.get_transport())
- scp.get('~/hello.txt', self.test_file_local_path)
+ scp = None
+ try:
+ scp = SCPClient(ssh.get_transport())
+ scp.get('~/hello.txt', self.test_file_local_path)
+ finally:
+ if scp:
+ scp.close()
+ ssh.close()
self.assertTrue(os.path.isfile(self.test_file_local_path))
- with open(self.test_file_local_path) as f:
- file_contents = f.readline()
- self.assertEquals('Hello World!', file_contents)
+ test_file = None
+
+ try:
+ with open(self.test_file_local_path) as test_file:
+ file_contents = test_file.readline()
+ self.assertEqual('Hello World!', file_contents)
+ finally:
+ if test_file:
+ test_file.close()
def test_apply_template_playbook(self):
"""
- Tests application of an Ansible playbook that applies a template to a file:
- 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
- 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
- Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
+ Tests application of an Ansible playbook that applies a template to a
+ file:
+ 1. Have a ~/.ansible.cfg (or alternate means) to set
+ host_key_checking = False
+ 2. Set the following environment variable in your executing shell:
+ ANSIBLE_HOST_KEY_CHECKING=False
+ Should this not be performed, the creation of the host ssh key will
+ cause your ansible calls to fail.
"""
self.inst_creator.create(block=True)
+ priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
+ self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip))
+
+ # Apply Security Group
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
+
# Block until VM's ssh port has been opened
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
- # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
+ # Block until cloud-init has completed
+ self.assertTrue(self.inst_creator.cloud_init_complete(block=True))
+
+ # Apply Security Group
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
+
+ # Need to use the first floating IP as subsequent ones are currently
+ # broken with Apex CO
ip = self.inst_creator.get_floating_ip().ip
user = self.inst_creator.get_image_user()
priv_key = self.inst_creator.keypair_settings.private_filepath
- ansible_utils.apply_playbook('provisioning/tests/playbooks/template_playbook.yml', [ip], user, priv_key,
- variables={'name': 'Foo'}, proxy_setting=self.os_creds.proxy_settings)
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks',
+ 'template_playbook.yml')
+ self.inst_creator.apply_ansible_playbook(
+ relative_pb_path, variables={'name': 'Foo'})
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(
+ ip, user, private_key_filepath=priv_key,
+ proxy_settings=self.os_creds.proxy_settings)
self.assertIsNotNone(ssh)
- scp = SCPClient(ssh.get_transport())
- scp.get('/tmp/hello.txt', self.test_file_local_path)
+ scp = None
+
+ try:
+ scp = SCPClient(ssh.get_transport())
+ scp.get('/tmp/hello.txt', self.test_file_local_path)
+ finally:
+ if scp:
+ scp.close()
+ ssh.close()
self.assertTrue(os.path.isfile(self.test_file_local_path))
- with open(self.test_file_local_path) as f:
- file_contents = f.readline()
- self.assertEquals('Hello Foo!', file_contents)
+ test_file = None
+ try:
+ with open(self.test_file_local_path) as test_file:
+ file_contents = test_file.readline()
+ self.assertEqual('Hello Foo!', file_contents)
+ finally:
+ if test_file:
+ test_file.close()
Code Review / snaps.git / blobdiff