Split functional test into standalone and utility class
This allows functional tests to be written for each individual template.
See the test_Wordpress_Single_Instance for an example, but the basic
form to follow is:
import util
def test_template():
func_utils = util.FuncUtils()
func_utils.create_stack(template, 'F17')
func_utils.check_cfntools()
func_utils.wait_for_provisioning()
func_utils.check_user_data(template)
ssh = func_utils.get_ssh_client()
# test stuff here
func_utils.cleanup()
Functional test changes:
Changed jeos creation to use new tool.
Added distribution to heat stack creation.
Improved IP parsing from glance.
Improve SSH connection exception handling.
Changed cfntools SHA check to match installed copy instead of locally.
Fixed stack parsing to check correct file and handle user-data injection.
Clean up run_tests a little to make it clear default execution is on host.
Add a short option for unit test running only (may be important due to not
wanting to run functional tests locally).
Also added paramiko to test-requires.
Change-Id: Ib2ae1be32801cb0ee3d1937c4f82ab9f574b4591
Signed-off-by: Jeff Peeler <jpeeler@redhat.com>
This commit is contained in:
Jeff Peeler
parent
ecc5a408a3
commit
8ba6c8ffe8
|
|
@ -0,0 +1,71 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import time # for sleep
|
||||
import util as func_utils
|
||||
from glance import client as glance_client
|
||||
|
||||
|
||||
def setUp(self):
|
||||
if os.geteuid() != 0:
|
||||
print 'test must be run as root'
|
||||
assert False
|
||||
|
||||
if os.environ['OS_AUTH_STRATEGY'] != 'keystone':
|
||||
print 'keystone authentication required'
|
||||
assert False
|
||||
|
||||
prepare_jeos()
|
||||
|
||||
|
||||
def prepare_jeos():
|
||||
# verify JEOS and cloud init
|
||||
args = ('F17', 'x86_64', 'cfntools')
|
||||
imagename = '-'.join(str(i) for i in args)
|
||||
creds = dict(username=os.environ['OS_USERNAME'],
|
||||
password=os.environ['OS_PASSWORD'],
|
||||
tenant=os.environ['OS_TENANT_NAME'],
|
||||
auth_url=os.environ['OS_AUTH_URL'],
|
||||
strategy=os.environ['OS_AUTH_STRATEGY'])
|
||||
|
||||
# -d: debug, -G: register with glance
|
||||
subprocess.call(['heat-jeos', '-d', '-G', 'create', imagename])
|
||||
|
||||
gclient = glance_client.Client(host="0.0.0.0", port=9292,
|
||||
use_ssl=False, auth_tok=None, creds=creds)
|
||||
|
||||
# Nose seems to change the behavior of the subprocess call to be
|
||||
# asynchronous. So poll glance until image is registered.
|
||||
imagelistname = None
|
||||
tries = 0
|
||||
while imagelistname != imagename:
|
||||
tries += 1
|
||||
assert tries < 50
|
||||
time.sleep(15)
|
||||
print "Checking glance for image registration"
|
||||
imageslist = gclient.get_images()
|
||||
for x in imageslist:
|
||||
imagelistname = x['name']
|
||||
if imagelistname == imagename:
|
||||
print "Found image registration for %s" % imagename
|
||||
break
|
||||
|
||||
# technically not necessary, but glance registers image before
|
||||
# completely through with its operations
|
||||
time.sleep(10)
|
||||
|
||||
# TODO: could do teardown and delete jeos
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
#
|
||||
|
||||
import util
|
||||
import nose
|
||||
from nose.plugins.attrib import attr
|
||||
|
||||
|
||||
@attr(speed='slow')
|
||||
@attr(tag=['func', 'wordpress'])
|
||||
def test_template():
|
||||
|
||||
template = 'WordPress_Single_Instance.template'
|
||||
|
||||
func_utils = util.FuncUtils()
|
||||
|
||||
func_utils.create_stack(template, 'F17')
|
||||
func_utils.check_cfntools()
|
||||
func_utils.wait_for_provisioning()
|
||||
func_utils.check_user_data(template)
|
||||
|
||||
ssh = func_utils.get_ssh_client()
|
||||
|
||||
# ensure wordpress was installed
|
||||
wp_file = '/etc/wordpress/wp-config.php'
|
||||
stdin, stdout, sterr = ssh.exec_command('ls ' + wp_file)
|
||||
result = stdout.readlines().pop().rstrip()
|
||||
assert result == wp_file
|
||||
print "Wordpress installation detected"
|
||||
|
||||
func_utils.cleanup()
|
||||
|
|
@ -1,250 +0,0 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
"""Functional test case that utilizes the bin/heat CLI tool"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import optparse
|
||||
import paramiko
|
||||
import subprocess
|
||||
import hashlib
|
||||
import email
|
||||
import json
|
||||
import time # for sleep
|
||||
import nose
|
||||
|
||||
from nose.plugins.attrib import attr
|
||||
from nose import with_setup
|
||||
|
||||
from novaclient.v1_1 import client
|
||||
from glance import client as glance_client
|
||||
from heat import utils
|
||||
from heat.engine import parser
|
||||
|
||||
|
||||
class TestBinHeat():
|
||||
"""Functional tests for the bin/heat CLI tool"""
|
||||
|
||||
def setUp(self):
|
||||
if os.geteuid() != 0:
|
||||
print 'test must be run as root'
|
||||
assert False
|
||||
|
||||
if os.environ['OS_AUTH_STRATEGY'] != 'keystone':
|
||||
print 'keystone authentication required'
|
||||
assert False
|
||||
|
||||
# this test is in heat/tests/functional, so go up 3 dirs
|
||||
self.basepath = os.path.abspath(
|
||||
os.path.dirname(os.path.realpath(__file__)) + '/../../..')
|
||||
|
||||
@attr(speed='slow')
|
||||
@attr(tag=['func', 'jeos'])
|
||||
def test_jeos_create(self):
|
||||
# 0. Verify JEOS and cloud init
|
||||
args = ('F16', 'x86_64', 'cfntools')
|
||||
|
||||
creds = dict(username=os.environ['OS_USERNAME'],
|
||||
password=os.environ['OS_PASSWORD'],
|
||||
tenant=os.environ['OS_TENANT_NAME'],
|
||||
auth_url=os.environ['OS_AUTH_URL'],
|
||||
strategy=os.environ['OS_AUTH_STRATEGY'])
|
||||
dbusername = 'testuser'
|
||||
|
||||
subprocess.call(['heat', '-d', 'jeos-create',
|
||||
args[0], args[1], args[2]])
|
||||
|
||||
gclient = glance_client.Client(host="0.0.0.0", port=9292,
|
||||
use_ssl=False, auth_tok=None, creds=creds)
|
||||
|
||||
# Nose seems to change the behavior of the subprocess call to be
|
||||
# asynchronous. So poll glance until image is registered.
|
||||
imagename = '-'.join(str(i) for i in args)
|
||||
imagelistname = None
|
||||
tries = 0
|
||||
while imagelistname != imagename:
|
||||
tries += 1
|
||||
assert tries < 5000
|
||||
time.sleep(15)
|
||||
print "Checking glance for image registration"
|
||||
imageslist = gclient.get_images()
|
||||
for x in imageslist:
|
||||
imagelistname = x['name']
|
||||
if imagelistname == imagename:
|
||||
print "Found image registration for %s" % imagename
|
||||
break
|
||||
|
||||
# technically not necessary, but glance registers image before
|
||||
# completely through with its operations
|
||||
time.sleep(10)
|
||||
|
||||
nt = client.Client(os.environ['OS_USERNAME'],
|
||||
os.environ['OS_PASSWORD'], os.environ['OS_TENANT_NAME'],
|
||||
os.environ['OS_AUTH_URL'], service_type='compute')
|
||||
|
||||
keyname = nt.keypairs.list().pop().name
|
||||
|
||||
subprocess.call(['heat', '-d', 'create', 'teststack',
|
||||
'--template-file=' + self.basepath +
|
||||
'/templates/WordPress_Single_Instance.template',
|
||||
'--parameters=InstanceType=m1.xlarge;DBUsername=' + dbusername +
|
||||
';DBPassword=' + os.environ['OS_PASSWORD'] +
|
||||
';KeyName=' + keyname])
|
||||
|
||||
print "Waiting for OpenStack to initialize and assign network address"
|
||||
ip = None
|
||||
tries = 0
|
||||
while ip is None:
|
||||
tries += 1
|
||||
assert tries < 500
|
||||
time.sleep(10)
|
||||
|
||||
for server in nt.servers.list():
|
||||
if server.name == 'WikiDatabase': # TODO: get from template
|
||||
address = server.addresses
|
||||
print "Status: %s" % server.status
|
||||
if address:
|
||||
ip = address['wordpress'][0]['addr']
|
||||
print 'IP found:', ip
|
||||
break
|
||||
elif server.status == 'ERROR':
|
||||
print 'Heat error? Aborting'
|
||||
assert False
|
||||
return
|
||||
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
subprocess.check_output(['nc', '-z', ip, '22'])
|
||||
except Exception:
|
||||
print 'SSH not up yet...'
|
||||
time.sleep(10)
|
||||
tries += 1
|
||||
assert tries < 100
|
||||
else:
|
||||
print 'SSH daemon response detected'
|
||||
time.sleep(5) # yuck, sometimes SSH is not *really* up
|
||||
break
|
||||
|
||||
ssh = paramiko.SSHClient()
|
||||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||||
ssh.connect(ip, username='ec2-user', allow_agent=True,
|
||||
look_for_keys=True, password='password')
|
||||
sftp = ssh.open_sftp()
|
||||
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
sftp.stat('/var/lib/cloud/instance/boot-finished')
|
||||
except IOError, e:
|
||||
tries += 1
|
||||
if e[0] == 2:
|
||||
assert tries < 40
|
||||
print "Boot not finished yet..."
|
||||
time.sleep(15)
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
print "boot-finished file found, start host checks"
|
||||
break
|
||||
|
||||
stdin, stdout, stderr = ssh.exec_command('cd /opt/aws/bin; sha1sum *')
|
||||
files = stdout.readlines()
|
||||
|
||||
cfn_tools_files = ['cfn-init', 'cfn-hup', 'cfn-signal',
|
||||
'cfn-get-metadata', 'cfn_helper.py']
|
||||
|
||||
cfntools = {}
|
||||
for file in cfn_tools_files:
|
||||
with open(self.basepath + '/heat/cfntools/' + file, 'rb') as f:
|
||||
sha = hashlib.sha1(f.read()).hexdigest()
|
||||
cfntools[file] = sha
|
||||
|
||||
# 1. make sure cfntools SHA in tree match VM's version
|
||||
for x in range(len(files)):
|
||||
data = files.pop().split(' ')
|
||||
cur_file = data[1].rstrip()
|
||||
if cur_file in cfn_tools_files:
|
||||
assert data[0] == cfntools[cur_file]
|
||||
|
||||
# 2. ensure wordpress was installed
|
||||
wp_file = '/etc/wordpress/wp-config.php'
|
||||
stdin, stdout, sterr = ssh.exec_command('ls ' + wp_file)
|
||||
result = stdout.readlines().pop().rstrip()
|
||||
assert result == wp_file
|
||||
|
||||
# 3. check multipart mime accuracy
|
||||
transport = ssh.get_transport()
|
||||
channel = transport.open_session()
|
||||
channel.get_pty()
|
||||
channel.invoke_shell() # sudo requires tty
|
||||
channel.sendall('sudo chmod 777 \
|
||||
/var/lib/cloud/instance/scripts/startup; \
|
||||
sudo chmod 777 /var/lib/cloud/instance/user-data.txt.i\n')
|
||||
time.sleep(1) # necessary for sendall to complete
|
||||
|
||||
filepaths = {
|
||||
'cloud-config': self.basepath + '/heat/cloudinit/config',
|
||||
'part-handler.py': self.basepath +
|
||||
'/heat/cloudinit/part-handler.py'
|
||||
}
|
||||
f = open(self.basepath +
|
||||
'/templates/WordPress_Single_Instance.template')
|
||||
t = json.loads(f.read())
|
||||
f.close()
|
||||
|
||||
template = parser.Template(t)
|
||||
params = parser.Parameters('test', t,
|
||||
{'KeyName': keyname,
|
||||
'DBUsername': dbusername,
|
||||
'DBPassword': creds['password']})
|
||||
|
||||
stack = parser.Stack(None, 'test', template, params)
|
||||
parsed_t = stack.resolve_static_refs(t)
|
||||
remote_file = sftp.open('/var/lib/cloud/instance/scripts/startup')
|
||||
remote_file_list = remote_file.read().split('\n')
|
||||
remote_file.close()
|
||||
|
||||
t_data = parsed_t['Resources']['WikiDatabase']['Properties']
|
||||
t_data = t_data['UserData']['Fn::Base64']['Fn::Join'].pop()
|
||||
joined_t_data = ''.join(t_data)
|
||||
t_data_list = joined_t_data.split('\n')
|
||||
|
||||
assert t_data_list == remote_file_list
|
||||
|
||||
remote_file = sftp.open('/var/lib/cloud/instance/user-data.txt.i')
|
||||
msg = email.message_from_file(remote_file)
|
||||
remote_file.close()
|
||||
|
||||
for part in msg.walk():
|
||||
# multipart/* are just containers
|
||||
if part.get_content_maintype() == 'multipart':
|
||||
continue
|
||||
|
||||
file = part.get_filename()
|
||||
data = part.get_payload()
|
||||
|
||||
if file in filepaths.keys():
|
||||
with open(filepaths[file]) as f:
|
||||
assert data == f.read()
|
||||
|
||||
# cleanup
|
||||
ssh.close()
|
||||
subprocess.call(['heat', 'delete', 'teststack'])
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.argv.append(__file__)
|
||||
nose.main()
|
||||
|
|
@ -0,0 +1,267 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
|
||||
import sys
|
||||
import os
|
||||
import optparse
|
||||
import paramiko
|
||||
import subprocess
|
||||
import hashlib
|
||||
import email
|
||||
import json
|
||||
import time # for sleep
|
||||
import nose
|
||||
import errno
|
||||
from pkg_resources import resource_string
|
||||
|
||||
from nose.plugins.attrib import attr
|
||||
from nose import with_setup
|
||||
from nose.exc import SkipTest
|
||||
|
||||
from novaclient.v1_1 import client
|
||||
from heat import utils
|
||||
from heat.engine import parser
|
||||
|
||||
|
||||
class FuncUtils:
|
||||
|
||||
# during nose test execution this file will be imported even if
|
||||
# the unit tag was specified
|
||||
try:
|
||||
os.environ['OS_AUTH_STRATEGY']
|
||||
except KeyError:
|
||||
raise SkipTest('OS_AUTH_STRATEGY not set, skipping functional test')
|
||||
|
||||
creds = dict(username=os.environ['OS_USERNAME'],
|
||||
password=os.environ['OS_PASSWORD'],
|
||||
tenant=os.environ['OS_TENANT_NAME'],
|
||||
auth_url=os.environ['OS_AUTH_URL'],
|
||||
strategy=os.environ['OS_AUTH_STRATEGY'])
|
||||
dbusername = 'testuser'
|
||||
stackname = 'teststack'
|
||||
|
||||
# this test is in heat/tests/functional, so go up 3 dirs
|
||||
basepath = os.path.abspath(
|
||||
os.path.dirname(os.path.realpath(__file__)) + '/../../..')
|
||||
|
||||
ssh = paramiko.SSHClient()
|
||||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||||
sftp = None
|
||||
|
||||
def get_ssh_client(self):
|
||||
if self.ssh.get_transport() != None:
|
||||
return self.ssh
|
||||
return None
|
||||
|
||||
def get_sftp_client(self):
|
||||
if self.sftp != None:
|
||||
return self.sftp
|
||||
return None
|
||||
|
||||
def create_stack(self, template_file, distribution):
|
||||
nt = client.Client(os.environ['OS_USERNAME'],
|
||||
os.environ['OS_PASSWORD'], os.environ['OS_TENANT_NAME'],
|
||||
os.environ['OS_AUTH_URL'], service_type='compute')
|
||||
|
||||
keyname = nt.keypairs.list().pop().name
|
||||
|
||||
subprocess.call(['heat', '-d', 'create', self.stackname,
|
||||
'--template-file=' + self.basepath +
|
||||
'/templates/' + template_file,
|
||||
'--parameters=InstanceType=m1.xlarge;DBUsername=' +
|
||||
self.dbusername +
|
||||
';DBPassword=' + os.environ['OS_PASSWORD'] +
|
||||
';KeyName=' + keyname +
|
||||
';LinuxDistribution=' + distribution])
|
||||
|
||||
print "Waiting for OpenStack to initialize and assign network address"
|
||||
ip = None
|
||||
tries = 0
|
||||
while ip is None:
|
||||
tries += 1
|
||||
assert tries < 500
|
||||
time.sleep(10)
|
||||
|
||||
for server in nt.servers.list():
|
||||
# TODO: get PhysicalResourceId instead
|
||||
if server.name == 'WikiDatabase':
|
||||
address = server.addresses
|
||||
print "Status: %s" % server.status
|
||||
if address:
|
||||
ip = address.items()[0][1][0]['addr']
|
||||
print 'IP found:', ip
|
||||
break
|
||||
elif server.status == 'ERROR':
|
||||
print 'Heat error? Aborting'
|
||||
assert False
|
||||
return
|
||||
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
subprocess.check_output(['nc', '-z', ip, '22'])
|
||||
except Exception:
|
||||
print 'SSH not up yet...'
|
||||
time.sleep(10)
|
||||
tries += 1
|
||||
assert tries < 50
|
||||
else:
|
||||
print 'SSH daemon response detected'
|
||||
break
|
||||
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
tries += 1
|
||||
assert tries < 50
|
||||
self.ssh.connect(ip, username='ec2-user', allow_agent=True,
|
||||
look_for_keys=True, password='password')
|
||||
except paramiko.AuthenticationException:
|
||||
print 'Authentication error'
|
||||
time.sleep(2)
|
||||
except Exception, e:
|
||||
if e.errno != errno.EHOSTUNREACH:
|
||||
raise
|
||||
print 'Preparing to connect over SSH'
|
||||
time.sleep(2)
|
||||
else:
|
||||
print 'SSH connected'
|
||||
break
|
||||
self.sftp = self.ssh.open_sftp()
|
||||
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
self.sftp.stat('/var/lib/cloud/instance/boot-finished')
|
||||
except IOError, e:
|
||||
tries += 1
|
||||
if e.errno == errno.ENOENT:
|
||||
assert tries < 50
|
||||
print "Boot not finished yet..."
|
||||
time.sleep(15)
|
||||
else:
|
||||
print e.errno
|
||||
raise
|
||||
else:
|
||||
print "Guest fully booted"
|
||||
break
|
||||
|
||||
def check_cfntools(self):
|
||||
stdin, stdout, stderr = \
|
||||
self.ssh.exec_command('cd /opt/aws/bin; sha1sum *')
|
||||
files = stdout.readlines()
|
||||
|
||||
cfn_tools_files = ['cfn-init', 'cfn-hup', 'cfn-signal',
|
||||
'cfn-get-metadata', 'cfn_helper.py']
|
||||
|
||||
cfntools = {}
|
||||
for file in cfn_tools_files:
|
||||
file_data = resource_string('heat_jeos', 'cfntools/' + file)
|
||||
sha = hashlib.sha1(file_data).hexdigest()
|
||||
cfntools[file] = sha
|
||||
|
||||
# 1. make sure installed cfntools SHA match VM's version
|
||||
for x in range(len(files)):
|
||||
data = files.pop().split(' ')
|
||||
cur_file = data[1].rstrip()
|
||||
if cur_file in cfn_tools_files:
|
||||
assert data[0] == cfntools[cur_file]
|
||||
print 'VM cfntools integrity verified'
|
||||
|
||||
def wait_for_provisioning(self):
|
||||
print "Waiting for provisioning to complete"
|
||||
tries = 0
|
||||
while True:
|
||||
try:
|
||||
self.sftp.stat('/var/lib/cloud/instance/provision-finished')
|
||||
except IOError, e:
|
||||
tries += 1
|
||||
if e.errno == errno.ENOENT:
|
||||
assert tries < 500
|
||||
print "Provisioning not finished yet..."
|
||||
time.sleep(15)
|
||||
else:
|
||||
print e.errno
|
||||
raise
|
||||
else:
|
||||
print "Provisioning completed"
|
||||
break
|
||||
|
||||
def check_user_data(self, template_file):
|
||||
transport = self.ssh.get_transport()
|
||||
channel = transport.open_session()
|
||||
channel.get_pty()
|
||||
channel.invoke_shell() # sudo requires tty
|
||||
channel.sendall('sudo chmod 777 \
|
||||
sudo chmod 777 /var/lib/cloud/instance/user-data.txt.i\n')
|
||||
time.sleep(1) # necessary for sendall to complete
|
||||
|
||||
f = open(self.basepath + '/templates/' + template_file)
|
||||
t = json.loads(f.read())
|
||||
f.close()
|
||||
|
||||
template = parser.Template(t)
|
||||
params = parser.Parameters('test', t,
|
||||
{'KeyName': 'required_parameter',
|
||||
'DBUsername': self.dbusername,
|
||||
'DBPassword': self.creds['password']})
|
||||
|
||||
stack = parser.Stack(None, 'test', template, params)
|
||||
parsed_t = stack.resolve_static_data(t)
|
||||
remote_file = self.sftp.open('/var/lib/cloud/data/cfn-userdata')
|
||||
remote_file_list = remote_file.read().split('\n')
|
||||
remote_file_list_u = map(unicode, remote_file_list)
|
||||
remote_file.close()
|
||||
|
||||
t_data = parsed_t['Resources']['WikiDatabase']['Properties']
|
||||
t_data = t_data['UserData']['Fn::Base64']['Fn::Join'].pop()
|
||||
joined_t_data = ''.join(t_data)
|
||||
t_data_list = joined_t_data.split('\n')
|
||||
# must match user data injection
|
||||
t_data_list.insert(len(t_data_list) - 1,
|
||||
u'touch /var/lib/cloud/instance/provision-finished')
|
||||
|
||||
assert t_data_list == remote_file_list_u
|
||||
|
||||
remote_file = self.sftp.open('/var/lib/cloud/instance/user-data.txt.i')
|
||||
msg = email.message_from_file(remote_file)
|
||||
remote_file.close()
|
||||
|
||||
filepaths = {
|
||||
'cloud-config': self.basepath + '/heat/cloudinit/config',
|
||||
'part-handler.py': self.basepath +
|
||||
'/heat/cloudinit/part-handler.py'
|
||||
}
|
||||
|
||||
# check multipart mime accuracy
|
||||
for part in msg.walk():
|
||||
# multipart/* are just containers
|
||||
if part.get_content_maintype() == 'multipart':
|
||||
continue
|
||||
|
||||
file = part.get_filename()
|
||||
data = part.get_payload()
|
||||
|
||||
if file in filepaths.keys():
|
||||
with open(filepaths[file]) as f:
|
||||
assert data == f.read()
|
||||
|
||||
def cleanup(self):
|
||||
self.ssh.close()
|
||||
subprocess.call(['heat', 'delete', self.stackname])
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.argv.append(__file__)
|
||||
nose.main()
|
||||
11
run_tests.sh
11
run_tests.sh
|
|
@ -5,17 +5,13 @@ function usage {
|
|||
echo "Run Heat's test suite(s)"
|
||||
echo ""
|
||||
echo " -V, --virtual-env Always use virtualenv. Install automatically if not present"
|
||||
echo " -N, --no-virtual-env Don't use virtualenv. Run tests in local environment"
|
||||
echo " -N, --no-virtual-env Don't use virtualenv. Run tests in local environment (default)"
|
||||
echo " -f, --force Force a clean re-build of the virtual environment. Useful when dependencies have been added."
|
||||
echo " --unittests-only Run unit tests only."
|
||||
echo " -u, --unittests-only Run unit tests only."
|
||||
echo " -p, --pep8 Just run pep8"
|
||||
echo " -P, --no-pep8 Don't run static code checks"
|
||||
echo " -c, --coverage Generate coverage report"
|
||||
echo " -h, --help Print this usage message"
|
||||
echo ""
|
||||
echo "Note: with no options specified, the script will try to run the tests in a virtual environment,"
|
||||
echo " If no virtualenv is found, the script will ask if you would like to create one. If you "
|
||||
echo " prefer to run tests NOT in a virtual environment, simply pass the -N option."
|
||||
exit
|
||||
}
|
||||
|
||||
|
|
@ -24,7 +20,7 @@ function process_option {
|
|||
-V|--virtual-env) let always_venv=1; let never_venv=0;;
|
||||
-N|--no-virtual-env) let always_venv=0; let never_venv=1;;
|
||||
-f|--force) let force=1;;
|
||||
--unittests-only) noseargs="$noseargs -a tag=unit";;
|
||||
-u|--unittests-only) noseargs="$noseargs -a tag=unit";;
|
||||
-p|--pep8) let just_pep8=1;;
|
||||
-P|--no-pep8) no_pep8=1;;
|
||||
-c|--coverage) coverage=1;;
|
||||
|
|
@ -35,6 +31,7 @@ function process_option {
|
|||
|
||||
venv=.venv
|
||||
with_venv=tools/with_venv.sh
|
||||
# change usage text if this option is changed:
|
||||
always_venv=0
|
||||
never_venv=1
|
||||
force=0
|
||||
|
|
|
|||
|
|
@ -8,3 +8,4 @@ nosexcover
|
|||
openstack.nose_plugin>=0.7
|
||||
pep8==1.1
|
||||
sphinx>=1.1.2
|
||||
paramiko
|
||||
|
|
|
|||
Loading…
Reference in New Issue