X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=snaps%2Fprovisioning%2Ftests%2Fansible_utils_tests.py;h=b6ace310398e4ba2f0af47ee46b6d4200580b138;hb=refs%2Fchanges%2F15%2F64515%2F1;hp=552171f9d889a1f171f7612d2263c1bfade0b509;hpb=06e3b9ca9abd8bc3d96707bbb2de0c460a196f2f;p=snaps.git diff --git a/snaps/provisioning/tests/ansible_utils_tests.py b/snaps/provisioning/tests/ansible_utils_tests.py index 552171f..b6ace31 100644 --- a/snaps/provisioning/tests/ansible_utils_tests.py +++ b/snaps/provisioning/tests/ansible_utils_tests.py @@ -1,4 +1,4 @@ -# 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"); @@ -13,20 +13,30 @@ # 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 @@ -36,16 +46,19 @@ 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 + 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' @@ -59,6 +72,7 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): # 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 @@ -66,82 +80,141 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): try: # Create Image - os_image_settings = openstack_tests.ubuntu_url_image(name=guid + '-' + '-image') - 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) - - 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_creator.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_creator: - self.image_creator.clean() + 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) @@ -150,69 +223,138 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): 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()