# See the License for the specific language governing permissions and
# limitations under the License.
+import uuid
+
import os
import pkg_resources
-import uuid
from scp import SCPClient
-from snaps.openstack.create_security_group import SecurityGroupRuleSettings, Direction, Protocol, \
- OpenStackSecurityGroup, SecurityGroupSettings
+
+from snaps.config.flavor import FlavorConfig
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig
from snaps.openstack import create_flavor
-from snaps.openstack import create_instance
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.tests import openstack_tests, create_instance_tests
+from snaps.openstack.create_security_group import (
+ SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
+ SecurityGroupSettings)
+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)
+
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
try:
# Create Image
- 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)
+ 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)
+ 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_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
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))
+ FlavorConfig(
+ name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
+ metadata=self.flavor_metadata))
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 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
protocol=Protocol.icmp)
- rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
+ rule2 = SecurityGroupRuleSettings(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,
- SecurityGroupSettings(name=sec_grp_name, rule_settings=[rule1, rule2]))
+ SecurityGroupSettings(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))
+ PortConfig(
+ 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,
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=ports_settings,
floating_ip_settings=[create_instance.FloatingIpSettings(
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_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
except:
self.tearDown()
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_creator and not self.image_creator.image_settings.exists:
- self.image_creator.clean()
+ 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.
"""
- vm = self.inst_creator.create(block=True)
-
- # Block until VM's ssh port has been opened
- self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
+ self.inst_creator.create(block=True)
priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(create_instance_tests.check_dhcp_lease(vm, priv_ip))
+ 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())
+ 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))
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
- relative_pb_path = pkg_resources.resource_filename('snaps.provisioning.tests.playbooks', 'simple_playbook.yml')
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks', 'simple_playbook.yml')
retval = self.inst_creator.apply_ansible_playbook(relative_pb_path)
self.assertEqual(0, retval)
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(ip, user, priv_key,
+ 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.assertEqual('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.
"""
- vm = self.inst_creator.create(block=True)
+ 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))
- priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(create_instance_tests.check_dhcp_lease(vm, priv_ip))
-
# Apply Security Group
- self.inst_creator.add_security_group(self.sec_grp_creator.get_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
+ # 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
- relative_pb_path = pkg_resources.resource_filename('snaps.provisioning.tests.playbooks',
- 'template_playbook.yml')
- retval = self.inst_creator.apply_ansible_playbook(relative_pb_path, variables={'name': 'Foo'})
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks',
+ 'template_playbook.yml')
+ retval = self.inst_creator.apply_ansible_playbook(relative_pb_path,
+ variables={
+ 'name': 'Foo'})
self.assertEqual(0, retval)
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(ip, user, priv_key,
+ 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.assertEqual('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()