self.assertEqual(fake_execute.uid, 2)
self.assertEqual(fake_execute.uid, os.getuid())
+ def test_service_is_up(self):
+ fts_func = datetime.datetime.fromtimestamp
+ fake_now = 1000
+ down_time = 5
+
+ self.flags(service_down_time=down_time)
+ self.mox.StubOutWithMock(timeutils, 'utcnow')
+
+ # Up (equal)
+ timeutils.utcnow().AndReturn(fts_func(fake_now))
+ service = {'updated_at': fts_func(fake_now - down_time),
+ 'created_at': fts_func(fake_now - down_time)}
+ self.mox.ReplayAll()
+ result = utils.service_is_up(service)
+ self.assertTrue(result)
+
+ self.mox.ResetAll()
+ # Up
+ timeutils.utcnow().AndReturn(fts_func(fake_now))
+ service = {'updated_at': fts_func(fake_now - down_time + 1),
+ 'created_at': fts_func(fake_now - down_time + 1)}
+ self.mox.ReplayAll()
+ result = utils.service_is_up(service)
+ self.assertTrue(result)
+
+ self.mox.ResetAll()
+ # Down
+ timeutils.utcnow().AndReturn(fts_func(fake_now))
+ service = {'updated_at': fts_func(fake_now - down_time - 1),
+ 'created_at': fts_func(fake_now - down_time - 1)}
+ self.mox.ReplayAll()
+ result = utils.service_is_up(service)
+ self.assertFalse(result)
+
+ def test_xhtml_escape(self):
+ self.assertEqual('"foo"', utils.xhtml_escape('"foo"'))
+ self.assertEqual(''foo'', utils.xhtml_escape("'foo'"))
+
+ def test_hash_file(self):
+ data = 'Mary had a little lamb, its fleece as white as snow'
+ flo = StringIO.StringIO(data)
+ h1 = utils.hash_file(flo)
+ h2 = hashlib.sha1(data).hexdigest()
+ self.assertEquals(h1, h2)
+
class IsUUIDLikeTestCase(test.TestCase):
def assertUUIDLike(self, val, expected):
in cinder.tests.monkey_patch_example.CALLED_FUNCTION)
-class DeprecationTest(test.TestCase):
- def setUp(self):
- super(DeprecationTest, self).setUp()
-
- def fake_warn_deprecated_class(cls, msg):
- self.warn = ('class', cls, msg)
-
- def fake_warn_deprecated_function(func, msg):
- self.warn = ('function', func, msg)
-
- self.stubs.Set(utils, 'warn_deprecated_class',
- fake_warn_deprecated_class)
- self.stubs.Set(utils, 'warn_deprecated_function',
- fake_warn_deprecated_function)
- self.warn = None
-
- def test_deprecated_function_no_message(self):
- def test_function():
- pass
-
- decorated = utils.deprecated()(test_function)
-
- decorated()
- self.assertEqual(self.warn, ('function', test_function, ''))
-
- def test_deprecated_function_with_message(self):
- def test_function():
- pass
-
- decorated = utils.deprecated('string')(test_function)
-
- decorated()
- self.assertEqual(self.warn, ('function', test_function, 'string'))
-
- def test_deprecated_class_no_message(self):
- @utils.deprecated()
- class TestClass(object):
- pass
-
- TestClass()
- self.assertEqual(self.warn, ('class', TestClass, ''))
-
- def test_deprecated_class_with_message(self):
- @utils.deprecated('string')
- class TestClass(object):
- pass
-
- TestClass()
- self.assertEqual(self.warn, ('class', TestClass, 'string'))
-
- def test_deprecated_classmethod_no_message(self):
- @utils.deprecated()
- class TestClass(object):
- @classmethod
- def class_method(cls):
- pass
-
- TestClass.class_method()
- self.assertEqual(self.warn, ('class', TestClass, ''))
-
- def test_deprecated_classmethod_with_message(self):
- @utils.deprecated('string')
- class TestClass(object):
- @classmethod
- def class_method(cls):
- pass
-
- TestClass.class_method()
- self.assertEqual(self.warn, ('class', TestClass, 'string'))
-
- def test_deprecated_staticmethod_no_message(self):
- @utils.deprecated()
- class TestClass(object):
- @staticmethod
- def static_method():
- pass
-
- TestClass.static_method()
- self.assertEqual(self.warn, ('class', TestClass, ''))
-
- def test_deprecated_staticmethod_with_message(self):
- @utils.deprecated('string')
- class TestClass(object):
- @staticmethod
- def static_method():
- pass
-
- TestClass.static_method()
- self.assertEqual(self.warn, ('class', TestClass, 'string'))
-
- def test_deprecated_instancemethod(self):
- @utils.deprecated()
- class TestClass(object):
- def instance_method(self):
- pass
-
- # Instantiate the class...
- obj = TestClass()
- self.assertEqual(self.warn, ('class', TestClass, ''))
-
- # Reset warn...
- self.warn = None
-
- # Call the instance method...
- obj.instance_method()
-
- # Make sure that did *not* generate a warning
- self.assertEqual(self.warn, None)
-
- def test_service_is_up(self):
- fts_func = datetime.datetime.fromtimestamp
- fake_now = 1000
- down_time = 5
-
- self.flags(service_down_time=down_time)
- self.mox.StubOutWithMock(timeutils, 'utcnow')
-
- # Up (equal)
- timeutils.utcnow().AndReturn(fts_func(fake_now))
- service = {'updated_at': fts_func(fake_now - down_time),
- 'created_at': fts_func(fake_now - down_time)}
- self.mox.ReplayAll()
- result = utils.service_is_up(service)
- self.assertTrue(result)
-
- self.mox.ResetAll()
- # Up
- timeutils.utcnow().AndReturn(fts_func(fake_now))
- service = {'updated_at': fts_func(fake_now - down_time + 1),
- 'created_at': fts_func(fake_now - down_time + 1)}
- self.mox.ReplayAll()
- result = utils.service_is_up(service)
- self.assertTrue(result)
-
- self.mox.ResetAll()
- # Down
- timeutils.utcnow().AndReturn(fts_func(fake_now))
- service = {'updated_at': fts_func(fake_now - down_time - 1),
- 'created_at': fts_func(fake_now - down_time - 1)}
- self.mox.ReplayAll()
- result = utils.service_is_up(service)
- self.assertFalse(result)
-
- def test_xhtml_escape(self):
- self.assertEqual('"foo"', utils.xhtml_escape('"foo"'))
- self.assertEqual(''foo'', utils.xhtml_escape("'foo'"))
-
- def test_hash_file(self):
- data = 'Mary had a little lamb, its fleece as white as snow'
- flo = StringIO.StringIO(data)
- h1 = utils.hash_file(flo)
- h2 = hashlib.sha1(data).hexdigest()
- self.assertEquals(h1, h2)
-
-
class AuditPeriodTest(test.TestCase):
def setUp(self):
setattr(obj, attr, old_value)
-def warn_deprecated_class(cls, msg):
- """
- Issues a warning to indicate that the given class is deprecated.
- If a message is given, it is appended to the deprecation warning.
- """
-
- fullname = '%s.%s' % (cls.__module__, cls.__name__)
- if msg:
- fullmsg = _("Class %(fullname)s is deprecated: %(msg)s")
- else:
- fullmsg = _("Class %(fullname)s is deprecated")
-
- # Issue the warning
- warnings.warn(fullmsg % locals(), DeprecationWarning, stacklevel=3)
-
-
-def warn_deprecated_function(func, msg):
- """
- Issues a warning to indicate that the given function is
- deprecated. If a message is given, it is appended to the
- deprecation warning.
- """
-
- name = func.__name__
-
- # Find the function's definition
- sourcefile = inspect.getsourcefile(func)
-
- # Find the line number, if possible
- if inspect.ismethod(func):
- code = func.im_func.func_code
- else:
- code = func.func_code
- lineno = getattr(code, 'co_firstlineno', None)
-
- if lineno is None:
- location = sourcefile
- else:
- location = "%s:%d" % (sourcefile, lineno)
-
- # Build up the message
- if msg:
- fullmsg = _("Function %(name)s in %(location)s is deprecated: %(msg)s")
- else:
- fullmsg = _("Function %(name)s in %(location)s is deprecated")
-
- # Issue the warning
- warnings.warn(fullmsg % locals(), DeprecationWarning, stacklevel=3)
-
-
-def _stubout(klass, message):
- """
- Scans a class and generates wrapping stubs for __new__() and every
- class and static method. Returns a dictionary which can be passed
- to type() to generate a wrapping class.
- """
-
- overrides = {}
-
- def makestub_class(name, func):
- """
- Create a stub for wrapping class methods.
- """
-
- def stub(cls, *args, **kwargs):
- warn_deprecated_class(klass, message)
- return func(*args, **kwargs)
-
- # Overwrite the stub's name
- stub.__name__ = name
- stub.func_name = name
-
- return classmethod(stub)
-
- def makestub_static(name, func):
- """
- Create a stub for wrapping static methods.
- """
-
- def stub(*args, **kwargs):
- warn_deprecated_class(klass, message)
- return func(*args, **kwargs)
-
- # Overwrite the stub's name
- stub.__name__ = name
- stub.func_name = name
-
- return staticmethod(stub)
-
- for name, kind, _klass, _obj in inspect.classify_class_attrs(klass):
- # We're only interested in __new__(), class methods, and
- # static methods...
- if (name != '__new__' and
- kind not in ('class method', 'static method')):
- continue
-
- # Get the function...
- func = getattr(klass, name)
-
- # Override it in the class
- if kind == 'class method':
- stub = makestub_class(name, func)
- elif kind == 'static method' or name == '__new__':
- stub = makestub_static(name, func)
-
- # Save it in the overrides dictionary...
- overrides[name] = stub
-
- # Apply the overrides
- for name, stub in overrides.items():
- setattr(klass, name, stub)
-
-
-def deprecated(message=''):
- """
- Marks a function, class, or method as being deprecated. For
- functions and methods, emits a warning each time the function or
- method is called. For classes, generates a new subclass which
- will emit a warning each time the class is instantiated, or each
- time any class or static method is called.
-
- If a message is passed to the decorator, that message will be
- appended to the emitted warning. This may be used to suggest an
- alternate way of achieving the desired effect, or to explain why
- the function, class, or method is deprecated.
- """
-
- def decorator(f_or_c):
- # Make sure we can deprecate it...
- if not callable(f_or_c) or isinstance(f_or_c, types.ClassType):
- warnings.warn("Cannot mark object %r as deprecated" % f_or_c,
- DeprecationWarning, stacklevel=2)
- return f_or_c
-
- # If we're deprecating a class, create a subclass of it and
- # stub out all the class and static methods
- if inspect.isclass(f_or_c):
- klass = f_or_c
- _stubout(klass, message)
- return klass
-
- # OK, it's a function; use a traditional wrapper...
- func = f_or_c
-
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- warn_deprecated_function(func, message)
-
- return func(*args, **kwargs)
-
- return wrapper
- return decorator
-
-
-def _showwarning(message, category, filename, lineno, file=None, line=None):
- """
- Redirect warnings into logging.
- """
-
- fmtmsg = warnings.formatwarning(message, category, filename, lineno, line)
- LOG.warning(fmtmsg)
-
-
-# Install our warnings handler
-warnings.showwarning = _showwarning
-
-
def service_is_up(service):
"""Check whether a service is up based on last heartbeat."""
last_heartbeat = service['updated_at'] or service['created_at']
Code Review / openstack-build / cinder-build.git / commitdiff