1 # Copyright (c) 2016 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at:
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
19 from snaps.openstack import create_instance
20 from snaps.openstack import create_keypairs
21 from snaps.openstack import create_network
22 from snaps.openstack import create_router
23 from snaps.openstack import create_image
24 from snaps.openstack import create_flavor
25 from scp import SCPClient
27 from snaps.provisioning import ansible_utils
28 from snaps.openstack.tests import openstack_tests
29 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
37 class AnsibleProvisioningTests(OSIntegrationTestCase):
39 Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP and eth1 w/o
44 Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
47 super(self.__class__, self).__start__()
49 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
50 self.keypair_priv_filepath = 'tmp/' + guid
51 self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
52 self.keypair_name = guid + '-kp'
53 self.vm_inst_name = guid + '-inst'
54 self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
55 self.port_1_name = guid + '-port-1'
56 self.port_2_name = guid + '-port-2'
57 self.floating_ip_name = guid + 'fip1'
59 # Setup members to cleanup just in case they don't get created
60 self.inst_creator = None
61 self.keypair_creator = None
62 self.flavor_creator = None
63 self.router_creator = None
64 self.network_creator = None
65 self.image_creators = list()
69 os_image_settings = openstack_tests.ubuntu_url_image(name=guid + '-' + '-image')
70 if self.image_metadata:
71 if self.image_metadata['disk_url']:
72 os_image_settings.url = self.image_metadata['disk_url']
73 if self.image_metadata['extra_properties']:
74 os_image_settings.extra_properties = self.image_metadata['extra_properties']
76 # If this is a 3-part image create the kernel and ramdisk images first
77 if self.image_metadata:
78 if self.image_metadata['kernel_url']:
79 kernel_image_settings = openstack_tests.cirros_url_image(
80 name=os_image_settings.name+'_kernel', url=self.image_metadata['kernel_url'])
81 self.image_creators.append(create_image.OpenStackImage(self.os_creds, kernel_image_settings))
82 kernel_image = self.image_creators[-1].create()
83 os_image_settings.extra_properties['kernel_id'] = kernel_image.id
85 if self.image_metadata['ramdisk_url']:
86 ramdisk_image_settings = openstack_tests.cirros_url_image(
87 name=os_image_settings.name+'_ramdisk', url=self.image_metadata['ramdisk_url'])
88 self.image_creators.append(create_image.OpenStackImage(self.os_creds, ramdisk_image_settings))
89 ramdisk_image = self.image_creators[-1].create()
90 os_image_settings.extra_properties['ramdisk_id'] = ramdisk_image.id
92 self.image_creators.append(create_image.OpenStackImage(self.os_creds, os_image_settings))
93 self.image_creators[-1].create()
95 # First network is public
96 self.pub_net_config = openstack_tests.get_pub_net_config(
97 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
98 router_name=guid + '-pub-router', external_net=self.ext_net_name)
100 self.network_creator = create_network.OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings)
101 self.network_creator.create()
104 self.router_creator = create_router.OpenStackRouter(self.os_creds, self.pub_net_config.router_settings)
105 self.router_creator.create()
108 self.flavor_creator = create_flavor.OpenStackFlavor(
110 create_flavor.FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
111 metadata=self.flavor_metadata))
112 self.flavor_creator.create()
115 self.keypair_creator = create_keypairs.OpenStackKeypair(
116 self.os_creds, create_keypairs.KeypairSettings(
117 name=self.keypair_name, public_filepath=self.keypair_pub_filepath,
118 private_filepath=self.keypair_priv_filepath))
119 self.keypair_creator.create()
122 ports_settings = list()
123 ports_settings.append(
124 create_network.PortSettings(name=self.port_1_name,
125 network_name=self.pub_net_config.network_settings.name))
127 instance_settings = create_instance.VmInstanceSettings(
128 name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings,
129 floating_ip_settings=[create_instance.FloatingIpSettings(
130 name=self.floating_ip_name, port_name=self.port_1_name,
131 router_name=self.pub_net_config.router_settings.name)])
133 self.inst_creator = create_instance.OpenStackVmInstance(
134 self.os_creds, instance_settings, self.image_creators[-1].image_settings,
135 keypair_settings=self.keypair_creator.keypair_settings)
136 except Exception as e:
138 raise Exception(e.message)
142 Cleans the created objects
144 if self.inst_creator:
145 self.inst_creator.clean()
147 if self.keypair_creator:
148 self.keypair_creator.clean()
150 if self.flavor_creator:
151 self.flavor_creator.clean()
153 if os.path.isfile(self.keypair_pub_filepath):
154 os.remove(self.keypair_pub_filepath)
156 if os.path.isfile(self.keypair_priv_filepath):
157 os.remove(self.keypair_priv_filepath)
159 if self.router_creator:
160 self.router_creator.clean()
162 if self.network_creator:
163 self.network_creator.clean()
165 if self.image_creators:
166 while self.image_creators:
167 self.image_creators[-1].clean()
168 self.image_creators.pop()
170 if os.path.isfile(self.test_file_local_path):
171 os.remove(self.test_file_local_path)
173 super(self.__class__, self).__clean__()
175 def test_apply_simple_playbook(self):
177 Tests application of an Ansible playbook that simply copies over a file:
178 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
179 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
180 Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
182 self.inst_creator.create(block=True)
184 # Block until VM's ssh port has been opened
185 self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
187 ssh_client = self.inst_creator.ssh_client()
188 self.assertIsNotNone(ssh_client)
189 out = ssh_client.exec_command('pwd')[1].channel.in_buffer.read(1024)
190 self.assertIsNotNone(out)
191 self.assertGreater(len(out), 1)
193 # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
194 ip = self.inst_creator.get_floating_ip().ip
195 user = self.inst_creator.get_image_user()
196 priv_key = self.inst_creator.keypair_settings.private_filepath
198 retval = ansible_utils.apply_playbook('provisioning/tests/playbooks/simple_playbook.yml', [ip], user, priv_key,
199 proxy_setting=self.os_creds.proxy_settings)
200 self.assertEquals(0, retval)
202 ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
203 self.assertIsNotNone(ssh)
204 scp = SCPClient(ssh.get_transport())
205 scp.get('~/hello.txt', self.test_file_local_path)
207 self.assertTrue(os.path.isfile(self.test_file_local_path))
209 with open(self.test_file_local_path) as f:
210 file_contents = f.readline()
211 self.assertEquals('Hello World!', file_contents)
213 def test_apply_template_playbook(self):
215 Tests application of an Ansible playbook that applies a template to a file:
216 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
217 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
218 Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
220 self.inst_creator.create(block=True)
222 # Block until VM's ssh port has been opened
223 self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
225 # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
226 ip = self.inst_creator.get_floating_ip().ip
227 user = self.inst_creator.get_image_user()
228 priv_key = self.inst_creator.keypair_settings.private_filepath
230 ansible_utils.apply_playbook('provisioning/tests/playbooks/template_playbook.yml', [ip], user, priv_key,
231 variables={'name': 'Foo'}, proxy_setting=self.os_creds.proxy_settings)
233 ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
234 self.assertIsNotNone(ssh)
235 scp = SCPClient(ssh.get_transport())
236 scp.get('/tmp/hello.txt', self.test_file_local_path)
238 self.assertTrue(os.path.isfile(self.test_file_local_path))
240 with open(self.test_file_local_path) as f:
241 file_contents = f.readline()
242 self.assertEquals('Hello Foo!', file_contents)