# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs") # and others. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import uuid import os import pkg_resources from scp import SCPClient from snaps.config.flavor import FlavorConfig 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.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 ip_1 = '10.0.1.100' ip_2 = '10.0.1.200' class AnsibleProvisioningTests(OSIntegrationTestCase): """ 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 """ 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' self.keypair_name = guid + '-kp' self.vm_inst_name = guid + '-inst' self.test_file_local_path = 'tmp/' + guid + '-hello.txt' self.port_1_name = guid + '-port-1' self.port_2_name = guid + '-port-2' self.floating_ip_name = guid + 'fip1' # 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_creator = None 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) self.image_creator.create() # First network is public self.pub_net_config = openstack_tests.get_pub_net_config( project_name=self.os_creds.project_name, 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) 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() # Create Flavor self.flavor_creator = create_flavor.OpenStackFlavor( self.admin_os_creds, 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, 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( 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_creator.image_settings, keypair_settings=self.keypair_creator.keypair_settings) except: self.tearDown() raise def tearDown(self): """ Cleans the created objects """ if self.inst_creator: try: self.inst_creator.clean() except: pass if self.sec_grp_creator: try: self.sec_grp_creator.clean() except: pass if self.keypair_creator: try: self.keypair_creator.clean() except: pass if self.flavor_creator: try: self.flavor_creator.clean() except: pass if os.path.isfile(self.keypair_pub_filepath): try: os.remove(self.keypair_pub_filepath) except: pass if os.path.isfile(self.keypair_priv_filepath): try: os.remove(self.keypair_priv_filepath) except: pass if self.router_creator: try: self.router_creator.clean() except: pass if self.network_creator: try: self.network_creator.clean() except: pass 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) super(self.__class__, self).__clean__() 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. """ 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) 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') retval = self.inst_creator.apply_ansible_playbook(relative_pb_path) self.assertEqual(0, retval) ssh = ansible_utils.ssh_client( ip, user, private_key_filepath=priv_key, proxy_settings=self.os_creds.proxy_settings) self.assertIsNotNone(ssh) 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)) 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. """ 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)) # 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 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, private_key_filepath=priv_key, proxy_settings=self.os_creds.proxy_settings) self.assertIsNotNone(ssh) 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)) 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()