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
# 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))
# 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)])
# 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
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):
"""
# 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())
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()
Code Review / snaps.git / blobdiff