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.openstack import create_flavor
from snaps.openstack import create_image
from snaps.openstack import create_instance
# 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 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))
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
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):
"""
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()