Merge branch 'resource_unit_tests'

changes/40/40/1
Chris Alfonso 11 years ago
commit bf859e6f4b
  1. 1
      .gitignore
  2. 2
      heat/common/config.py
  3. 23
      heat/db/api.py
  4. 31
      heat/db/migration.py
  5. 111
      heat/db/sqlalchemy/migration.py
  6. 60
      heat/engine/resources.py
  7. 60
      heat/testing/README.rst
  8. 0
      heat/testing/__init__.py
  9. 1
      heat/testing/fake/__init__.py
  10. 362
      heat/testing/runner.py
  11. 24
      heat/tests/__init__.py
  12. 2
      heat/tests/examples/test3.py
  13. 82
      heat/tests/test_resources.py
  14. 26
      heat/utils.py
  15. 13
      run_tests.sh

1
.gitignore vendored

@ -6,3 +6,4 @@ heat.egg-info
heat/vcsversion.py
tags
*.log
heat/tests/heat-test.db

@ -178,8 +178,6 @@ class HeatEngineConfigOpts(cfg.CommonConfigOpts):
help='port for os volume api to listen'),
]
db_opts = [
cfg.StrOpt('db_backend', default='heat.db.sqlalchemy.api',
help='The backend to use for db'),
cfg.StrOpt('sql_connection',
default='mysql://heat:heat@localhost/heat',
help='The SQLAlchemy connection string used to connect to the '

@ -25,19 +25,32 @@ Usage:
The underlying driver is loaded . SQLAlchemy is currently the only
supported backend.
'''
import heat.utils
from heat.openstack.common import utils
from heat.openstack.common import cfg
from heat.common import config
import heat.utils
SQL_CONNECTION = 'sqlite:///heat-test.db/'
SQL_IDLE_TIMEOUT = 3600
db_opts = [
cfg.StrOpt('db_backend',
default='sqlalchemy',
help='The backend to use for db'),
]
conf = config.HeatEngineConfigOpts()
conf.db_backend = 'heat.db.sqlalchemy.api'
IMPL = heat.utils.LazyPluggable('db_backend',
sqlalchemy='heat.db.sqlalchemy.api')
def configure(conf):
global IMPL
global SQL_CONNECTION
global SQL_IDLE_TIMEOUT
IMPL = utils.import_object(conf.db_backend)
SQL_CONNECTION = conf.sql_connection
SQL_IDLE_TIMEOUT = conf.sql_idle_timeout
def raw_template_get(context, template_id):
return IMPL.raw_template_get(context, template_id)

@ -0,0 +1,31 @@
# 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.
"""Database setup and migration commands."""
from heat import utils
IMPL = utils.LazyPluggable('db_backend',
sqlalchemy='heat.db.sqlalchemy.migration')
def db_sync(version=None):
"""Migrate the database to `version` or the most recent version."""
return IMPL.db_sync(version=version)
def db_version():
"""Display the current database version."""
return IMPL.db_version()

@ -0,0 +1,111 @@
# 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 distutils.version as dist_version
import os
import sys
from heat.db.sqlalchemy.session import get_engine
import sqlalchemy
import migrate
from migrate.versioning import util as migrate_util
_REPOSITORY = None
@migrate_util.decorator
def patched_with_engine(f, *a, **kw):
url = a[0]
engine = migrate_util.construct_engine(url, **kw)
try:
kw['engine'] = engine
return f(*a, **kw)
finally:
if isinstance(engine, migrate_util.Engine) and engine is not url:
migrate_util.log.debug('Disposing SQLAlchemy engine %s', engine)
engine.dispose()
# TODO(jkoelker) When migrate 0.7.3 is released and nova depends
# on that version or higher, this can be removed
MIN_PKG_VERSION = dist_version.StrictVersion('0.7.3')
if (not hasattr(migrate, '__version__') or
dist_version.StrictVersion(migrate.__version__) < MIN_PKG_VERSION):
migrate_util.with_engine = patched_with_engine
# NOTE(jkoelker) Delay importing migrate until we are patched
from migrate.versioning import api as versioning_api
from migrate.versioning.repository import Repository
try:
from migrate.versioning import exceptions as versioning_exceptions
except ImportError:
try:
from migrate import exceptions as versioning_exceptions
except ImportError:
sys.exit(_("python-migrate is not installed. Exiting."))
#_REPOSITORY = None
def db_sync(version=None):
if version is not None:
try:
version = int(version)
except ValueError:
raise exception.Error(_("version should be an integer"))
current_version = db_version()
repository = _find_migrate_repo()
if version is None or version > current_version:
return versioning_api.upgrade(get_engine(), repository, version)
else:
return versioning_api.downgrade(get_engine(), repository,
version)
def db_version():
repository = _find_migrate_repo()
try:
return versioning_api.db_version(get_engine(), repository)
except versioning_exceptions.DatabaseNotControlledError:
# If we aren't version controlled we may already have the database
# in the state from before we started version control, check for that
# and set up version_control appropriately
meta = sqlalchemy.MetaData()
engine = get_engine()
meta.reflect(bind=engine)
try:
for table in ('stack', 'resource', 'event',
'parsed_template', 'raw_template'):
assert table in meta.tables
return db_version_control(1)
except AssertionError:
return db_version_control(0)
def db_version_control(version=None):
repository = _find_migrate_repo()
versioning_api.version_control(get_engine(), repository, version)
return version
def _find_migrate_repo():
"""Get the path for the migrate repository."""
path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'migrate_repo')
assert os.path.exists(path)
global _REPOSITORY
if _REPOSITORY is None:
_REPOSITORY = Repository(path)
return _REPOSITORY

@ -20,7 +20,6 @@ import os
import string
import json
import sys
from email import encoders
from email.message import Message
from email.mime.base import MIMEBase
@ -36,7 +35,6 @@ from heat.db import api as db_api
from heat.common.config import HeatEngineConfigOpts
logger = logging.getLogger('heat.engine.resources')
# If ../heat/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
@ -78,7 +76,6 @@ class Resource(object):
self.instance_id = None
self.state = None
self.id = None
self._nova = {}
if not 'Properties' in self.t:
# make a dummy entry to prevent having to check all over the
@ -522,6 +519,33 @@ class Instance(Resource):
logger.info('%s.GetAtt(%s) == %s' % (self.name, key, res))
return unicode(res)
def _build_userdata(self, userdata):
# Build mime multipart data blob for cloudinit userdata
mime_blob = MIMEMultipart()
fp = open('%s/%s' % (cloudinit_path, 'config'), 'r')
msg = MIMEText(fp.read(), _subtype='cloud-config')
fp.close()
msg.add_header('Content-Disposition', 'attachment',
filename='cloud-config')
mime_blob.attach(msg)
fp = open('%s/%s' % (cloudinit_path, 'part-handler.py'), 'r')
msg = MIMEText(fp.read(), _subtype='part-handler')
fp.close()
msg.add_header('Content-Disposition', 'attachment',
filename='part-handler.py')
mime_blob.attach(msg)
msg = MIMEText(json.dumps(self.t['Metadata']),
_subtype='x-cfninitdata')
msg.add_header('Content-Disposition', 'attachment',
filename='cfn-init-data')
mime_blob.attach(msg)
msg = MIMEText(userdata, _subtype='x-shellscript')
msg.add_header('Content-Disposition', 'attachment', filename='startup')
return mime_blob.attach(msg)
def create(self):
def _null_callback(p, n, out):
"""
@ -533,7 +557,6 @@ class Instance(Resource):
return
self.state_set(self.CREATE_IN_PROGRESS)
Resource.create(self)
props = self.t['Properties']
if not 'KeyName' in props:
raise exception.UserParameterMissing(key='KeyName')
@ -576,37 +599,12 @@ class Instance(Resource):
if o.name == flavor:
flavor_id = o.id
# Build mime multipart data blob for cloudinit userdata
mime_blob = MIMEMultipart()
fp = open('%s/%s' % (cloudinit_path, 'config'), 'r')
msg = MIMEText(fp.read(), _subtype='cloud-config')
fp.close()
msg.add_header('Content-Disposition', 'attachment',
filename='cloud-config')
mime_blob.attach(msg)
fp = open('%s/%s' % (cloudinit_path, 'part-handler.py'), 'r')
msg = MIMEText(fp.read(), _subtype='part-handler')
fp.close()
msg.add_header('Content-Disposition', 'attachment',
filename='part-handler.py')
mime_blob.attach(msg)
msg = MIMEText(json.dumps(self.t['Metadata']),
_subtype='x-cfninitdata')
msg.add_header('Content-Disposition', 'attachment',
filename='cfn-init-data')
mime_blob.attach(msg)
msg = MIMEText(userdata, _subtype='x-shellscript')
msg.add_header('Content-Disposition', 'attachment', filename='startup')
mime_blob.attach(msg)
server_userdata = self._build_userdata(userdata)
server = self.nova().servers.create(name=self.name, image=image_id,
flavor=flavor_id,
key_name=key_name,
security_groups=security_groups,
userdata=mime_blob.as_string())
userdata=server_userdata)
while server.status == 'BUILD':
server.get()
eventlet.sleep(1)

@ -0,0 +1,60 @@
=====================================
Heat Testing Infrastructure
=====================================
A note of clarification is in order, to help those who are new to testing in
Heat:
- actual unit tests are created in the "tests" directory;
- the "testing" directory is used to house the infrastructure needed to support
testing in Heat.
This README file attempts to provide current and prospective contributors with
everything they need to know in order to start creating unit tests and
utilizing the convenience code provided in nova.testing.
Test Types: Unit vs. Functional vs. Integration
-----------------------------------------------
TBD
Writing Unit Tests
------------------
TBD
Using Fakes
~~~~~~~~~~~
TBD
test.TestCase
-------------
The TestCase class from heat.test (generally imported as test) will
automatically manage self.stubs using the stubout module and self.mox
using the mox module during the setUp step. They will automatically
verify and clean up during the tearDown step.
If using test.TestCase, calling the super class setUp is required and
calling the super class tearDown is required to be last if tearDown
is overriden.
Writing Functional Tests
------------------------
TBD
Writing Integration Tests
-------------------------
TBD
Tests and assertRaises
----------------------
When asserting that a test should raise an exception, test against the
most specific exception possible. An overly broad exception type (like
Exception) can mask errors in the unit test itself.
Example::
TBD

@ -0,0 +1,362 @@
#!/usr/bin/env python
# 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.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Unittest runner for Heat.
To run all tests
python heat/testing/runner.py
To run a single test module:
python heat/testing/runner.py test_resources
To run a single test:
python heat/testing/runner.py
test_resources:ResourceTestCase.test_resource_from_template
"""
import gettext
import heapq
import os
import unittest
import sys
import time
import eventlet
from nose import config
from nose import core
from nose import result
gettext.install('heat', unicode=1)
reldir = os.path.join(os.path.dirname(__file__), '..', '..')
absdir = os.path.abspath(reldir)
sys.path.insert(0, absdir)
from heat.openstack.common import cfg
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except Exception:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
import win32console as win
red, green, blue, bold = (win.FOREGROUND_RED, win.FOREGROUND_GREEN,
win.FOREGROUND_BLUE, win.FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = win.GetStdHandle(win.STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold
}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
def get_elapsed_time_color(elapsed_time):
if elapsed_time > 1.0:
return 'red'
elif elapsed_time > 0.25:
return 'yellow'
else:
return 'green'
class HeatTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
self.show_elapsed = kw.pop('show_elapsed')
result.TextTestResult.__init__(self, *args, **kw)
self.num_slow_tests = 5
self.slow_tests = [] # this is a fixed-sized heap
self._last_case = None
self.colorizer = None
# NOTE(vish): reset stdout for the terminal check
stdout = sys.stdout
sys.stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
# NOTE(lorinh): Initialize start_time in case a sqlalchemy-migrate
# error results in it failing to be initialized later. Otherwise,
# _handleElapsedTime will fail, causing the wrong error message to
# be outputted.
self.start_time = time.time()
def getDescription(self, test):
return str(test)
def _handleElapsedTime(self, test):
self.elapsed_time = time.time() - self.start_time
item = (self.elapsed_time, test)
# Record only the n-slowest tests using heap
if len(self.slow_tests) >= self.num_slow_tests:
heapq.heappushpop(self.slow_tests, item)
else:
heapq.heappush(self.slow_tests, item)
def _writeElapsedTime(self, test):
color = get_elapsed_time_color(self.elapsed_time)
self.colorizer.write(" %.2f" % self.elapsed_time, color)
def _writeResult(self, test, long_result, color, short_result, success):
if self.showAll:
self.colorizer.write(long_result, color)
if self.show_elapsed and success:
self._writeElapsedTime(test)
self.stream.writeln()
elif self.dots:
self.stream.write(short_result)
self.stream.flush()
# NOTE(vish): copied from unittest with edit to add color
def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
self._handleElapsedTime(test)
self._writeResult(test, 'OK', 'green', '.', True)
# NOTE(vish): copied from unittest with edit to add color
def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
self._handleElapsedTime(test)
self._writeResult(test, 'FAIL', 'red', 'F', False)
# NOTE(vish): copied from nose with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for
errorClasses. If the exception is a registered class, the
error will be added to the list for that class, not errors.
"""
self._handleElapsedTime(test)
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# 2.3 compat
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passed = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_detail(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
self._writeResult(test, 'ERROR', 'red', 'E', False)
def startTest(self, test):
unittest.TestResult.startTest(self, test)
self.start_time = time.time()
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class HeatTestRunner(core.TextTestRunner):
def __init__(self, *args, **kwargs):
self.show_elapsed = kwargs.pop('show_elapsed')
core.TextTestRunner.__init__(self, *args, **kwargs)
def _makeResult(self):
return HeatTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config,
show_elapsed=self.show_elapsed)
def _writeSlowTests(self, result_):
# Pare out 'fast' tests
slow_tests = [item for item in result_.slow_tests
if get_elapsed_time_color(item[0]) != 'green']
if slow_tests:
slow_total_time = sum(item[0] for item in slow_tests)
self.stream.writeln("Slowest %i tests took %.2f secs:"
% (len(slow_tests), slow_total_time))
for elapsed_time, test in sorted(slow_tests, reverse=True):
time_str = "%.2f" % elapsed_time
self.stream.writeln(" %s %s" % (time_str.ljust(10), test))
def run(self, test):
result_ = core.TextTestRunner.run(self, test)
if self.show_elapsed:
self._writeSlowTests(result_)
return result_
def run():
# This is a fix to allow the --hide-elapsed flag while accepting
# arbitrary nosetest flags as well
argv = [x for x in sys.argv if x != '--hide-elapsed']
hide_elapsed = argv != sys.argv
# If any argument looks like a test name but doesn't have "heat.tests" in
# front of it, automatically add that so we don't have to type as much
for i, arg in enumerate(argv):
if arg.startswith('test_'):
argv[i] = 'heat.tests.%s' % arg
testdir = os.path.abspath(os.path.join("heat", "tests"))
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=testdir,
plugins=core.DefaultPluginManager())
runner = HeatTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c,
show_elapsed=not hide_elapsed)
sys.exit(not core.run(config=c, testRunner=runner, argv=argv))
if __name__ == '__main__':
eventlet.monkey_patch()
run()

@ -16,3 +16,27 @@
# The code below enables nosetests to work with i18n _() blocks
import __builtin__
setattr(__builtin__, '_', lambda x: x)
import os
import shutil
from heat.db.sqlalchemy.session import get_engine
_DB = None
def reset_db():
engine = get_engine()
engine.dispose()
conn = engine.connect()
conn.connection.executescript(_DB)
def setup():
import mox # Fail fast if you don't have mox. Workaround for bug 810424
from heat import db
from heat.db import migration
migration.db_sync()
engine = get_engine()
conn = engine.connect()
# _DB = "".join(line for line in conn.connection.dump())

@ -11,7 +11,7 @@ from nose.plugins.attrib import attr
# sets attribute on all test methods
@attr(tag=['example', 'unittest'])
@attr(tag=['example', 'unit'])
@attr(speed='fast')
class ExampleTest(unittest.TestCase):
def test_a(self):

@ -0,0 +1,82 @@
import sys
import os
sys.path.append(os.environ['PYTHON_NOVACLIENT_SRC'])
import nose
import unittest
import mox
import json
import sqlalchemy
from nose.plugins.attrib import attr
from nose import with_setup
from tests.v1_1 import fakes
from heat.engine import resources
import heat.db as db_api
from heat.engine import parser
@attr(tag=['unit', 'resource'])
@attr(speed='fast')
class ResourcesTest(unittest.TestCase):
def setUp(self):
self.m = mox.Mox()
self.fc = fakes.FakeClient()
def tearDown(self):
self.m.UnsetStubs()
print "ResourcesTest teardown complete"
def test_initialize_instance_from_template(self):
f = open('../../templates/WordPress_Single_Instance_gold.template')
t = json.loads(f.read())
f.close()
params = {}
parameters = {}
params['KeyStoneCreds'] = None
t['Parameters']['KeyName']['Value'] = 'test'
stack = parser.Stack('test_stack', t, 0, params)
self.m.StubOutWithMock(db_api, 'resource_get_by_name_and_stack')
db_api.resource_get_by_name_and_stack(None, 'test_resource_name',\
stack).AndReturn(None)
self.m.StubOutWithMock(resources.Instance, 'nova')
resources.Instance.nova().AndReturn(self.fc)
resources.Instance.nova().AndReturn(self.fc)
resources.Instance.nova().AndReturn(self.fc)
resources.Instance.nova().AndReturn(self.fc)
#Need to find an easier way
userdata = t['Resources']['WebServer']['Properties']['UserData']
self.m.ReplayAll()
t['Resources']['WebServer']['Properties']['ImageId'] = 'CentOS 5.2'
t['Resources']['WebServer']['Properties']['InstanceType'] = '256 MB Server'
instance = resources.Instance('test_resource_name',\
t['Resources']['WebServer'], stack)
server_userdata = instance._build_userdata(json.dumps(userdata))
self.m.StubOutWithMock(self.fc.servers, 'create')
self.fc.servers.create(image=1, flavor=1, key_name='test',\
name='test_resource_name', security_groups=None,\
userdata=server_userdata).\
AndReturn(self.fc.servers.list()[1])
self.m.ReplayAll()
instance.itype_oflavor['256 MB Server'] = '256 MB Server'
instance.create()
self.m.ReplayAll()
instance.itype_oflavor['256 MB Server'] = '256 MB Server'
instance.create()
# allows testing of the test directly, shown below
if __name__ == '__main__':
sys.argv.append(__file__)
nose.main()

@ -0,0 +1,26 @@
class LazyPluggable(object):
"""A pluggable backend loaded lazily based on some value."""
def __init__(self, pivot, **backends):
self.__backends = backends
self.__pivot = pivot
self.__backend = None
def __get_backend(self):
if not self.__backend:
print self.__backends.values()
backend_name = 'sqlalchemy'
backend = self.__backends[backend_name]
if isinstance(backend, tuple):
name = backend[0]
fromlist = backend[1]
else:
name = backend
fromlist = backend
self.__backend = __import__(name, None, None, fromlist)
return self.__backend
def __getattr__(self, key):
backend = self.__get_backend()
return getattr(backend, key)

@ -42,9 +42,10 @@ for arg in "$@"; do
process_option $arg
done
NOSETESTS="python run_tests.py $noseargs"
NOSETESTS="python heat/testing/runner.py $noseopts $noseargs"
function run_tests {
echo 'Running tests'
# Just run the test suites in current environment
${wrapper} $NOSETESTS 2> run_tests.err.log
}
@ -52,7 +53,7 @@ function run_tests {
function run_pep8 {
echo "Running pep8 ..."
PEP8_OPTIONS="--exclude=$PEP8_EXCLUDE --repeat"
PEP8_INCLUDE="bin/heat bin/heat-api bin/heat-engine heat tools setup.py run_tests.py"
PEP8_INCLUDE="bin/heat bin/heat-api bin/heat-engine heat tools setup.py heat/testing/runner.py"
${wrapper} pep8 $PEP8_OPTIONS $PEP8_INCLUDE
}
@ -87,9 +88,9 @@ if [ $just_pep8 -eq 1 ]; then
exit
fi
run_tests || exit
run_tests
if [ -z "$noseargs" ]; then
run_pep8
fi
#if [ -z "$noseargs" ]; then
# run_pep8
#fi

Loading…
Cancel
Save