Tests cleanup (and renamed TestCase -> Case

6 files changed, 614 insertions, 0 deletions

diff --git a/kombu/tests/utils/__init__.py b/kombu/tests/utils/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/kombu/tests/utils/__init__.py
diff --git a/kombu/tests/utils/test_amq_manager.py b/kombu/tests/utils/test_amq_manager.py
new file mode 100644
index 00000000..b34b52e8
--- /dev/null
+++ b/kombu/tests/utils/test_amq_manager.py
@@ -0,0 +1,36 @@
+from __future__ import absolute_import
+
+from kombu import Connection
+
+from kombu.tests.case import Case, mask_modules, module_exists, patch
+
+
+class test_get_manager(Case):
+
+    @mask_modules('pyrabbit')
+    def test_without_pyrabbit(self):
+        with self.assertRaises(ImportError):
+            Connection('amqp://').get_manager()
+
+    @module_exists('pyrabbit')
+    def test_with_pyrabbit(self):
+        with patch('pyrabbit.Client', create=True) as Client:
+            manager = Connection('amqp://').get_manager()
+            self.assertIsNotNone(manager)
+            Client.assert_called_with(
+                'localhost:15672', 'guest', 'guest',
+            )
+
+    @module_exists('pyrabbit')
+    def test_transport_options(self):
+        with patch('pyrabbit.Client', create=True) as Client:
+            manager = Connection('amqp://', transport_options={
+                'manager_hostname': 'admin.mq.vandelay.com',
+                'manager_port': 808,
+                'manager_userid': 'george',
+                'manager_password': 'bosco',
+            }).get_manager()
+            self.assertIsNotNone(manager)
+            Client.assert_called_with(
+                'admin.mq.vandelay.com:808', 'george', 'bosco',
+            )
diff --git a/kombu/tests/utils/test_debug.py b/kombu/tests/utils/test_debug.py
new file mode 100644
index 00000000..ea25cb77
--- /dev/null
+++ b/kombu/tests/utils/test_debug.py
@@ -0,0 +1,56 @@
+from __future__ import absolute_import
+
+import logging
+
+from kombu.utils.debug import (
+    setup_logging,
+    Logwrapped,
+)
+from kombu.tests.case import Case, Mock, patch
+
+
+class test_setup_logging(Case):
+
+    def test_adds_handlers_sets_level(self):
+        with patch('kombu.utils.debug.get_logger') as get_logger:
+            logger = get_logger.return_value = Mock()
+            setup_logging(loggers=['kombu.test'])
+
+            get_logger.assert_called_with('kombu.test')
+
+            self.assertTrue(logger.addHandler.called)
+            logger.setLevel.assert_called_with(logging.DEBUG)
+
+
+class test_Logwrapped(Case):
+
+    def test_wraps(self):
+        with patch('kombu.utils.debug.get_logger') as get_logger:
+            logger = get_logger.return_value = Mock()
+
+            W = Logwrapped(Mock(), 'kombu.test')
+            get_logger.assert_called_with('kombu.test')
+            self.assertIsNotNone(W.instance)
+            self.assertIs(W.logger, logger)
+
+            W.instance.__repr__ = lambda s: 'foo'
+            self.assertEqual(repr(W), 'foo')
+            W.instance.some_attr = 303
+            self.assertEqual(W.some_attr, 303)
+
+            W.instance.some_method.__name__ = 'some_method'
+            W.some_method(1, 2, kw=1)
+            W.instance.some_method.assert_called_with(1, 2, kw=1)
+
+            W.some_method()
+            W.instance.some_method.assert_called_with()
+
+            W.some_method(kw=1)
+            W.instance.some_method.assert_called_with(kw=1)
+
+            W.ident = 'ident'
+            W.some_method(kw=1)
+            self.assertTrue(logger.debug.called)
+            self.assertIn('ident', logger.debug.call_args[0][0])
+
+            self.assertEqual(dir(W), dir(W.instance))
diff --git a/kombu/tests/utils/test_encoding.py b/kombu/tests/utils/test_encoding.py
new file mode 100644
index 00000000..a56d2a7e
--- /dev/null
+++ b/kombu/tests/utils/test_encoding.py
@@ -0,0 +1,102 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+from __future__ import unicode_literals
+
+import sys
+
+from contextlib import contextmanager
+
+from kombu.five import bytes_t, string_t
+from kombu.utils.encoding import safe_str, default_encoding
+
+from kombu.tests.case import Case, SkipTest, patch
+
+
+@contextmanager
+def clean_encoding():
+    old_encoding = sys.modules.pop('kombu.utils.encoding', None)
+    import kombu.utils.encoding
+    try:
+        yield kombu.utils.encoding
+    finally:
+        if old_encoding:
+            sys.modules['kombu.utils.encoding'] = old_encoding
+
+
+class test_default_encoding(Case):
+
+    @patch('sys.getdefaultencoding')
+    def test_default(self, getdefaultencoding):
+        getdefaultencoding.return_value = 'ascii'
+        with clean_encoding() as encoding:
+            enc = encoding.default_encoding()
+            if sys.platform.startswith('java'):
+                self.assertEqual(enc, 'utf-8')
+            else:
+                self.assertEqual(enc, 'ascii')
+                getdefaultencoding.assert_called_with()
+
+
+class test_encoding_utils(Case):
+
+    def setUp(self):
+        if sys.version_info >= (3, 0):
+            raise SkipTest('not relevant on py3k')
+
+    def test_str_to_bytes(self):
+        with clean_encoding() as e:
+            self.assertIsInstance(e.str_to_bytes('foobar'), bytes_t)
+
+    def test_from_utf8(self):
+        with clean_encoding() as e:
+            self.assertIsInstance(e.from_utf8('foobar'), bytes_t)
+
+    def test_default_encode(self):
+        with clean_encoding() as e:
+            self.assertTrue(e.default_encode(b'foo'))
+
+
+class test_safe_str(Case):
+
+    def setUp(self):
+        self._cencoding = patch('sys.getdefaultencoding')
+        self._encoding = self._cencoding.__enter__()
+        self._encoding.return_value = 'ascii'
+
+    def tearDown(self):
+        self._cencoding.__exit__()
+
+    def test_when_bytes(self):
+        self.assertEqual(safe_str('foo'), 'foo')
+
+    def test_when_unicode(self):
+        self.assertIsInstance(safe_str('foo'), string_t)
+
+    def test_when_encoding_utf8(self):
+        with patch('sys.getdefaultencoding') as encoding:
+            encoding.return_value = 'utf-8'
+            self.assertEqual(default_encoding(), 'utf-8')
+            s = 'The quiæk fåx jømps øver the lazy dåg'
+            res = safe_str(s)
+            self.assertIsInstance(res, str)
+
+    def test_when_containing_high_chars(self):
+        with patch('sys.getdefaultencoding') as encoding:
+            encoding.return_value = 'ascii'
+            s = 'The quiæk fåx jømps øver the lazy dåg'
+            res = safe_str(s)
+            self.assertIsInstance(res, str)
+            self.assertEqual(len(s), len(res))
+
+    def test_when_not_string(self):
+        o = object()
+        self.assertEqual(safe_str(o), repr(o))
+
+    def test_when_unrepresentable(self):
+
+        class O(object):
+
+            def __repr__(self):
+                raise KeyError('foo')
+
+        self.assertIn('<Unrepresentable', safe_str(O()))
diff --git a/kombu/tests/utils/test_functional.py b/kombu/tests/utils/test_functional.py
new file mode 100644
index 00000000..20b8bbd1
--- /dev/null
+++ b/kombu/tests/utils/test_functional.py
@@ -0,0 +1,59 @@
+from __future__ import absolute_import
+
+import pickle
+
+from kombu.utils.functional import lazy, maybe_evaluate
+
+from kombu.tests.case import Case
+
+
+def double(x):
+    return x * 2
+
+
+class test_lazy(Case):
+
+    def test__str__(self):
+        self.assertEqual(
+            str(lazy(lambda: 'the quick brown fox')),
+            'the quick brown fox',
+        )
+
+    def test__repr__(self):
+        self.assertEqual(
+            repr(lazy(lambda: 'fi fa fo')),
+            "'fi fa fo'",
+        )
+
+    def test__cmp__(self):
+        self.assertEqual(lazy(lambda: 10).__cmp__(lazy(lambda: 20)), -1)
+        self.assertEqual(lazy(lambda: 10).__cmp__(5), 1)
+
+    def test_evaluate(self):
+        self.assertEqual(lazy(lambda: 2 + 2)(), 4)
+        self.assertEqual(lazy(lambda x: x * 4, 2), 8)
+        self.assertEqual(lazy(lambda x: x * 8, 2)(), 16)
+
+    def test_cmp(self):
+        self.assertEqual(lazy(lambda: 10), lazy(lambda: 10))
+        self.assertNotEqual(lazy(lambda: 10), lazy(lambda: 20))
+
+    def test__reduce__(self):
+        x = lazy(double, 4)
+        y = pickle.loads(pickle.dumps(x))
+        self.assertEqual(x(), y())
+
+    def test__deepcopy__(self):
+        from copy import deepcopy
+        x = lazy(double, 4)
+        y = deepcopy(x)
+        self.assertEqual(x._fun, y._fun)
+        self.assertEqual(x._args, y._args)
+        self.assertEqual(x(), y())
+
+
+class test_maybe_evaluate(Case):
+
+    def test_evaluates(self):
+        self.assertEqual(maybe_evaluate(lazy(lambda: 10)), 10)
+        self.assertEqual(maybe_evaluate(20), 20)
diff --git a/kombu/tests/utils/test_utils.py b/kombu/tests/utils/test_utils.py
new file mode 100644
index 00000000..6392feec
--- /dev/null
+++ b/kombu/tests/utils/test_utils.py
@@ -0,0 +1,361 @@
+from __future__ import absolute_import
+from __future__ import unicode_literals
+
+import pickle
+import sys
+
+from functools import wraps
+
+if sys.version_info >= (3, 0):
+    from io import StringIO, BytesIO
+else:
+    from StringIO import StringIO, StringIO as BytesIO  # noqa
+
+from kombu import utils
+from kombu.five import string_t
+
+from kombu.tests.case import (
+    Case, Mock, patch,
+    redirect_stdouts, mask_modules, module_exists, skip_if_module,
+)
+
+
+class OldString(object):
+
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return self.value
+
+    def split(self, *args, **kwargs):
+        return self.value.split(*args, **kwargs)
+
+    def rsplit(self, *args, **kwargs):
+        return self.value.rsplit(*args, **kwargs)
+
+
+class test_kombu_module(Case):
+
+    def test_dir(self):
+        import kombu
+        self.assertTrue(dir(kombu))
+
+
+class test_utils(Case):
+
+    def test_maybe_list(self):
+        self.assertEqual(utils.maybe_list(None), [])
+        self.assertEqual(utils.maybe_list(1), [1])
+        self.assertEqual(utils.maybe_list([1, 2, 3]), [1, 2, 3])
+
+    def test_fxrange_no_repeatlast(self):
+        self.assertEqual(list(utils.fxrange(1.0, 3.0, 1.0)),
+                         [1.0, 2.0, 3.0])
+
+    def test_fxrangemax(self):
+        self.assertEqual(list(utils.fxrangemax(1.0, 3.0, 1.0, 30.0)),
+                         [1.0, 2.0, 3.0, 3.0, 3.0, 3.0,
+                          3.0, 3.0, 3.0, 3.0, 3.0])
+        self.assertEqual(list(utils.fxrangemax(1.0, None, 1.0, 30.0)),
+                         [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0])
+
+    def test_reprkwargs(self):
+        self.assertTrue(utils.reprkwargs({'foo': 'bar', 1: 2, 'k': 'v'}))
+
+    def test_reprcall(self):
+        self.assertTrue(
+            utils.reprcall('add', (2, 2), {'copy': True}),
+        )
+
+
+class test_UUID(Case):
+
+    def test_uuid4(self):
+        self.assertNotEqual(utils.uuid4(),
+                            utils.uuid4())
+
+    def test_uuid(self):
+        i1 = utils.uuid()
+        i2 = utils.uuid()
+        self.assertIsInstance(i1, str)
+        self.assertNotEqual(i1, i2)
+
+    @skip_if_module('__pypy__')
+    def test_uuid_without_ctypes(self):
+        old_utils = sys.modules.pop('kombu.utils')
+
+        @mask_modules('ctypes')
+        def with_ctypes_masked():
+            from kombu.utils import ctypes, uuid
+
+            self.assertIsNone(ctypes)
+            tid = uuid()
+            self.assertTrue(tid)
+            self.assertIsInstance(tid, string_t)
+
+        try:
+            with_ctypes_masked()
+        finally:
+            sys.modules['celery.utils'] = old_utils
+
+
+class test_Misc(Case):
+
+    def test_kwdict(self):
+
+        def f(**kwargs):
+            return kwargs
+
+        kw = {'foo': 'foo',
+              'bar': 'bar'}
+        self.assertTrue(f(**utils.kwdict(kw)))
+
+
+class MyStringIO(StringIO):
+
+    def close(self):
+        pass
+
+
+class MyBytesIO(BytesIO):
+
+    def close(self):
+        pass
+
+
+class test_emergency_dump_state(Case):
+
+    @redirect_stdouts
+    def test_dump(self, stdout, stderr):
+        fh = MyBytesIO()
+
+        utils.emergency_dump_state({'foo': 'bar'}, open_file=lambda n, m: fh)
+        self.assertDictEqual(pickle.loads(fh.getvalue()), {'foo': 'bar'})
+        self.assertTrue(stderr.getvalue())
+        self.assertFalse(stdout.getvalue())
+
+    @redirect_stdouts
+    def test_dump_second_strategy(self, stdout, stderr):
+        fh = MyStringIO()
+
+        def raise_something(*args, **kwargs):
+            raise KeyError('foo')
+
+        utils.emergency_dump_state(
+            {'foo': 'bar'},
+            open_file=lambda n, m: fh, dump=raise_something
+        )
+        self.assertIn('foo', fh.getvalue())
+        self.assertIn('bar', fh.getvalue())
+        self.assertTrue(stderr.getvalue())
+        self.assertFalse(stdout.getvalue())
+
+
+def insomnia(fun):
+
+    @wraps(fun)
+    def _inner(*args, **kwargs):
+        def mysleep(i):
+            pass
+
+        prev_sleep = utils.sleep
+        utils.sleep = mysleep
+        try:
+            return fun(*args, **kwargs)
+        finally:
+            utils.sleep = prev_sleep
+
+    return _inner
+
+
+class test_retry_over_time(Case):
+
+    def setUp(self):
+        self.index = 0
+
+    class Predicate(Exception):
+        pass
+
+    def myfun(self):
+        if self.index < 9:
+            raise self.Predicate()
+        return 42
+
+    def errback(self, exc, intervals, retries):
+        interval = next(intervals)
+        sleepvals = (None, 2.0, 4.0, 6.0, 8.0, 10.0, 12.0, 14.0, 16.0, 16.0)
+        self.index += 1
+        self.assertEqual(interval, sleepvals[self.index])
+        return interval
+
+    @insomnia
+    def test_simple(self):
+        prev_count, utils.count = utils.count, Mock()
+        try:
+            utils.count.return_value = list(range(1))
+            x = utils.retry_over_time(self.myfun, self.Predicate,
+                                      errback=None, interval_max=14)
+            self.assertIsNone(x)
+            utils.count.return_value = list(range(10))
+            cb = Mock()
+            x = utils.retry_over_time(self.myfun, self.Predicate,
+                                      errback=self.errback, callback=cb,
+                                      interval_max=14)
+            self.assertEqual(x, 42)
+            self.assertEqual(self.index, 9)
+            cb.assert_called_with()
+        finally:
+            utils.count = prev_count
+
+    @insomnia
+    def test_retry_once(self):
+        self.assertRaises(
+            self.Predicate, utils.retry_over_time,
+            self.myfun, self.Predicate,
+            max_retries=1, errback=self.errback, interval_max=14,
+        )
+        self.assertEqual(self.index, 1)
+        # no errback
+        self.assertRaises(
+            self.Predicate, utils.retry_over_time,
+            self.myfun, self.Predicate,
+            max_retries=1, errback=None, interval_max=14,
+        )
+
+    @insomnia
+    def test_retry_never(self):
+        self.assertRaises(
+            self.Predicate, utils.retry_over_time,
+            self.myfun, self.Predicate,
+            max_retries=0, errback=self.errback, interval_max=14,
+        )
+        self.assertEqual(self.index, 0)
+
+
+class test_cached_property(Case):
+
+    def test_deleting(self):
+
+        class X(object):
+            xx = False
+
+            @utils.cached_property
+            def foo(self):
+                return 42
+
+            @foo.deleter  # noqa
+            def foo(self, value):
+                self.xx = value
+
+        x = X()
+        del(x.foo)
+        self.assertFalse(x.xx)
+        x.__dict__['foo'] = 'here'
+        del(x.foo)
+        self.assertEqual(x.xx, 'here')
+
+    def test_when_access_from_class(self):
+
+        class X(object):
+            xx = None
+
+            @utils.cached_property
+            def foo(self):
+                return 42
+
+            @foo.setter  # noqa
+            def foo(self, value):
+                self.xx = 10
+
+        desc = X.__dict__['foo']
+        self.assertIs(X.foo, desc)
+
+        self.assertIs(desc.__get__(None), desc)
+        self.assertIs(desc.__set__(None, 1), desc)
+        self.assertIs(desc.__delete__(None), desc)
+        self.assertTrue(desc.setter(1))
+
+        x = X()
+        x.foo = 30
+        self.assertEqual(x.xx, 10)
+
+        del(x.foo)
+
+
+class test_symbol_by_name(Case):
+
+    def test_instance_returns_instance(self):
+        instance = object()
+        self.assertIs(utils.symbol_by_name(instance), instance)
+
+    def test_returns_default(self):
+        default = object()
+        self.assertIs(
+            utils.symbol_by_name('xyz.ryx.qedoa.weq:foz', default=default),
+            default,
+        )
+
+    def test_no_default(self):
+        with self.assertRaises(ImportError):
+            utils.symbol_by_name('xyz.ryx.qedoa.weq:foz')
+
+    def test_imp_reraises_ValueError(self):
+        imp = Mock()
+        imp.side_effect = ValueError()
+        with self.assertRaises(ValueError):
+            utils.symbol_by_name('kombu.Connection', imp=imp)
+
+    def test_package(self):
+        from kombu.entity import Exchange
+        self.assertIs(
+            utils.symbol_by_name('.entity:Exchange', package='kombu'),
+            Exchange,
+        )
+        self.assertTrue(utils.symbol_by_name(':Consumer', package='kombu'))
+
+
+class test_ChannelPromise(Case):
+
+    def test_repr(self):
+        self.assertIn(
+            'foo',
+            repr(utils.ChannelPromise(lambda: 'foo')),
+        )
+
+
+class test_entrypoints(Case):
+
+    @mask_modules('pkg_resources')
+    def test_without_pkg_resources(self):
+        self.assertListEqual(list(utils.entrypoints('kombu.test')), [])
+
+    @module_exists('pkg_resources')
+    def test_with_pkg_resources(self):
+        with patch('pkg_resources.iter_entry_points', create=True) as iterep:
+            eps = iterep.return_value = [Mock(), Mock()]
+
+            self.assertTrue(list(utils.entrypoints('kombu.test')))
+            iterep.assert_called_with('kombu.test')
+            eps[0].load.assert_called_with()
+            eps[1].load.assert_called_with()
+
+
+class test_shufflecycle(Case):
+
+    def test_shuffles(self):
+        prev_repeat, utils.repeat = utils.repeat, Mock()
+        try:
+            utils.repeat.return_value = list(range(10))
+            values = set(['A', 'B', 'C'])
+            cycle = utils.shufflecycle(values)
+            seen = set()
+            for i in range(10):
+                next(cycle)
+            utils.repeat.assert_called_with(None)
+            self.assertTrue(seen.issubset(values))
+            with self.assertRaises(StopIteration):
+                next(cycle)
+                next(cycle)
+        finally:
+            utils.repeat = prev_repeat