X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=snaps%2Fprovisioning%2Ftests%2Fansible_utils_tests.py;h=851dd644d5328d26d9a9d52f78d14c8ba23fad22;hb=88e9754a4280690acf2ec5942b34336d469673ed;hp=4c8dea6783229d7a2501c1cbae77794db8ae4ed2;hpb=5f3fe6856f07bd1289bac532264eccf6cba68d77;p=snaps.git diff --git a/snaps/provisioning/tests/ansible_utils_tests.py b/snaps/provisioning/tests/ansible_utils_tests.py index 4c8dea6..851dd64 100644 --- a/snaps/provisioning/tests/ansible_utils_tests.py +++ b/snaps/provisioning/tests/ansible_utils_tests.py @@ -18,15 +18,21 @@ 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 ( - SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup, - SecurityGroupSettings) +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 @@ -100,14 +106,14 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): # 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( + self.os_creds, KeypairConfig( name=self.keypair_name, public_filepath=self.keypair_pub_filepath, private_filepath=self.keypair_priv_filepath)) @@ -115,32 +121,30 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): # Create Security Group sec_grp_name = guid + '-sec-grp' - 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) + 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, - SecurityGroupSettings(name=sec_grp_name, - rule_settings=[rule1, rule2])) + 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( + PortConfig( name=self.port_1_name, network_name=self.pub_net_config.network_settings.name)) - instance_settings = create_instance.VmInstanceSettings( + instance_settings = VmInstanceConfig( name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings, - floating_ip_settings=[create_instance.FloatingIpSettings( + floating_ip_settings=[FloatingIpConfig( name=self.floating_ip_name, port_name=self.port_1_name, router_name=self.pub_net_config.router_settings.name)]) @@ -225,10 +229,10 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): 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.nova, vm, priv_ip)) + self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip)) # Apply Security Group self.inst_creator.add_security_group( @@ -237,11 +241,19 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): # 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) + + 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 @@ -254,17 +266,30 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): 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, 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.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): """ @@ -277,10 +302,10 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): 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.nova, vm, priv_ip)) + self.assertTrue(check_dhcp_lease(self.inst_creator, priv_ip)) # Apply Security Group self.inst_creator.add_security_group( @@ -289,6 +314,9 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): # 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()) @@ -302,19 +330,31 @@ class AnsibleProvisioningTests(OSIntegrationTestCase): 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'}) + 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, 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.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()