testtools has these built in - so we should use them.
Change-Id: I2a9f6dbd69d5acef9a0dfbe5e90e9dbf706d8677
LOG = logging.getLogger(__name__)
-class skip_test(object):
- """Decorator that skips a test."""
- # TODO(tr3buchet): remember forever what comstud did here
- def __init__(self, msg):
- self.message = msg
-
- def __call__(self, func):
- @functools.wraps(func)
- def _skipper(*args, **kw):
- """Wrapped skipper function."""
- raise testtools.TestCase.skipException(self.message)
- return _skipper
-
-
-class skip_if(object):
- """Decorator that skips a test if condition is true."""
- def __init__(self, condition, msg):
- self.condition = condition
- self.message = msg
-
- def __call__(self, func):
- @functools.wraps(func)
- def _skipper(*args, **kw):
- """Wrapped skipper function."""
- if self.condition:
- raise testtools.TestCase.skipException(self.message)
- func(*args, **kw)
- return _skipper
-
-
-class skip_unless(object):
- """Decorator that skips a test if condition is not true."""
- def __init__(self, condition, msg):
- self.condition = condition
- self.message = msg
-
- def __call__(self, func):
- @functools.wraps(func)
- def _skipper(*args, **kw):
- """Wrapped skipper function."""
- if not self.condition:
- raise testtools.TestCase.skipException(self.message)
- func(*args, **kw)
- return _skipper
-
-
-def skip_if_fake(func):
- """Decorator that skips a test if running in fake mode."""
- def _skipper(*args, **kw):
- """Wrapped skipper function."""
- if FLAGS.fake_tests:
- raise testtools.TestCase.skipException(
- 'Test cannot be run in fake mode')
- else:
- return func(*args, **kw)
- return _skipper
-
-
class TestingException(Exception):
pass
Tests For Capacity Weigher.
"""
+import testtools
+
from cinder import context
from cinder.openstack.common.scheduler.weights import HostWeightHandler
from cinder import test
self.mox.ResetAll()
return host_states
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_default_of_spreading_first(self):
hostinfo_list = self._get_all_hosts()
self.assertEqual(weighed_host.weight, 921.0)
self.assertEqual(weighed_host.obj.host, 'host1')
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_weight_multiplier1(self):
self.flags(capacity_weight_multiplier=-1.0)
hostinfo_list = self._get_all_hosts()
self.assertEqual(weighed_host.weight, -190.0)
self.assertEqual(weighed_host.obj.host, 'host4')
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_weight_multiplier2(self):
self.flags(capacity_weight_multiplier=2.0)
hostinfo_list = self._get_all_hosts()
Tests For Filter Scheduler.
"""
+import testtools
+
from cinder import context
from cinder import exception
from cinder import test
driver_cls = filter_scheduler.FilterScheduler
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_create_volume_no_hosts(self):
"""
Ensure empty hosts & child_zones result in NoValidHosts exception.
self.assertRaises(exception.NoValidHost, sched.schedule_create_volume,
fake_context, request_spec, {})
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_create_volume_non_admin(self):
"""Test creating an instance locally using run_instance, passing
a non-admin context. DB actions should work."""
fake_context, request_spec, {})
self.assertTrue(self.was_admin)
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_schedule_happy_day(self):
"""Make sure there's nothing glaringly wrong with _schedule()
by doing a happy day pass through."""
self.assertRaises(exception.InvalidParameterValue,
fakes.FakeFilterScheduler)
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_retry_disabled(self):
# Retry info should not get populated when re-scheduling is off.
self.flags(scheduler_max_attempts=1)
# should not have retry info in the populated filter properties:
self.assertFalse("retry" in filter_properties)
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_retry_attempt_one(self):
# Test retry logic on initial scheduling attempt.
self.flags(scheduler_max_attempts=2)
num_attempts = filter_properties['retry']['num_attempts']
self.assertEqual(1, num_attempts)
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed (try setup.py develop')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed (try setup.py develop')
def test_retry_attempt_two(self):
# Test retry logic when re-scheduling.
self.flags(scheduler_max_attempts=2)
import httplib
import stubout
+import testtools
from cinder import context
from cinder import db
return ret_value
self.stubs.Set(utils, 'service_is_up', fake_service_is_up)
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_filter_passes(self):
self._stub_service_is_up(True)
filt_cls = self.class_map['CapacityFilter']()
'service': service})
self.assertTrue(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_filter_fails(self):
self._stub_service_is_up(True)
filt_cls = self.class_map['CapacityFilter']()
'service': service})
self.assertFalse(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_filter_passes_infinite(self):
self._stub_service_is_up(True)
filt_cls = self.class_map['CapacityFilter']()
'service': service})
self.assertTrue(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_capacity_filter_passes_unknown(self):
self._stub_service_is_up(True)
filt_cls = self.class_map['CapacityFilter']()
'service': service})
self.assertTrue(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_retry_filter_disabled(self):
# Test case where retry/re-scheduling is disabled.
filt_cls = self.class_map['RetryFilter']()
filter_properties = {}
self.assertTrue(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_retry_filter_pass(self):
# Node not previously tried.
filt_cls = self.class_map['RetryFilter']()
filter_properties = dict(retry=retry)
self.assertTrue(filt_cls.host_passes(host, filter_properties))
- @test.skip_if(not test_utils.is_cinder_installed(),
- 'Test requires Cinder installed')
+ @testtools.skipIf(not test_utils.is_cinder_installed(),
+ 'Test requires Cinder installed')
def test_retry_filter_fail(self):
# Node was already tried.
filt_cls = self.class_map['RetryFilter']()
from migrate.versioning import repository
import sqlalchemy
+import testtools
import cinder.db.migration as migration
import cinder.db.sqlalchemy.migrate_repo
if _is_mysql_avail(user="openstack_cifail"):
self.fail("Shouldn't have connected")
- @test.skip_unless(_have_mysql(), "mysql not available")
+ @testtools.skipUnless(_have_mysql(), "mysql not available")
def test_mysql_innodb(self):
"""
Test that table creation on mysql only builds InnoDB tables
if _is_backend_avail('postgres', user="openstack_cifail"):
self.fail("Shouldn't have connected")
- @test.skip_unless(_is_backend_avail('postgres'),
- "postgresql not available")
+ @testtools.skipUnless(_is_backend_avail('postgres'),
+ "postgresql not available")
def test_postgresql_opportunistically(self):
# add this to the global lists to make reset work with it, it's removed
# automatically in tearDown so no need to clean it up here.
+++ /dev/null
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# All Rights Reserved.
-#
-# 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.
-
-from cinder import test
-
-
-class ExampleSkipTestCase(test.TestCase):
- test_counter = 0
-
- @test.skip_test("Example usage of @test.skip_test()")
- def test_skip_test_example(self):
- self.fail("skip_test failed to work properly.")
-
- @test.skip_if(True, "Example usage of @test.skip_if()")
- def test_skip_if_example(self):
- self.fail("skip_if failed to work properly.")
-
- @test.skip_unless(False, "Example usage of @test.skip_unless()")
- def test_skip_unless_example(self):
- self.fail("skip_unless failed to work properly.")
-
- @test.skip_if(False, "This test case should never be skipped.")
- def test_001_increase_test_counter(self):
- ExampleSkipTestCase.test_counter += 1
-
- @test.skip_unless(True, "This test case should never be skipped.")
- def test_002_increase_test_counter(self):
- ExampleSkipTestCase.test_counter += 1
-
- def test_003_verify_test_counter(self):
- self.assertEquals(ExampleSkipTestCase.test_counter, 2,
- "Tests were not skipped appropriately")
import urllib2
from oslo.config import cfg
+import testtools
import webob
import webob.dec
server.stop()
server.wait()
- @test.skip_if(not _ipv6_configured(),
- "Test requires an IPV6 configured interface")
+ @testtools.skipIf(not _ipv6_configured(),
+ "Test requires an IPV6 configured interface")
def test_start_random_port_with_ipv6(self):
server = cinder.wsgi.Server("test_random_port",
None,
server.stop()
- @test.skip_if(not _ipv6_configured(),
- "Test requires an IPV6 configured interface")
+ @testtools.skipIf(not _ipv6_configured(),
+ "Test requires an IPV6 configured interface")
def test_app_using_ipv6_and_ssl(self):
CONF.set_default("ssl_cert_file",
os.path.join(TEST_VAR_DIR, 'certificate.crt'))