/usr/lib/python2.7/dist-packages/celery/tests/utils/test_utils.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | from __future__ import absolute_import
import pytz
from datetime import datetime, date, time, timedelta
from kombu import Queue
from celery.utils import (
chunks,
is_iterable,
cached_property,
warn_deprecated,
worker_direct,
gen_task_name,
jsonify,
)
from celery.tests.case import Case, Mock, patch
def double(x):
return x * 2
class test_worker_direct(Case):
def test_returns_if_queue(self):
q = Queue('foo')
self.assertIs(worker_direct(q), q)
class test_gen_task_name(Case):
def test_no_module(self):
app = Mock()
app.name == '__main__'
self.assertTrue(gen_task_name(app, 'foo', 'axsadaewe'))
class test_jsonify(Case):
def test_simple(self):
self.assertTrue(jsonify(Queue('foo')))
self.assertTrue(jsonify(['foo', 'bar', 'baz']))
self.assertTrue(jsonify({'foo': 'bar'}))
self.assertTrue(jsonify(datetime.utcnow()))
self.assertTrue(jsonify(datetime.utcnow().replace(tzinfo=pytz.utc)))
self.assertTrue(jsonify(datetime.utcnow().replace(microsecond=0)))
self.assertTrue(jsonify(date(2012, 1, 1)))
self.assertTrue(jsonify(time(hour=1, minute=30)))
self.assertTrue(jsonify(time(hour=1, minute=30, microsecond=3)))
self.assertTrue(jsonify(timedelta(seconds=30)))
self.assertTrue(jsonify(10))
self.assertTrue(jsonify(10.3))
self.assertTrue(jsonify('hello'))
with self.assertRaises(ValueError):
jsonify(object())
class test_chunks(Case):
def test_chunks(self):
# n == 2
x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2)
self.assertListEqual(
list(x),
[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]],
)
# n == 3
x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3)
self.assertListEqual(
list(x),
[[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]],
)
# n == 2 (exact)
x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), 2)
self.assertListEqual(
list(x),
[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]],
)
class test_utils(Case):
def test_is_iterable(self):
for a in 'f', ['f'], ('f', ), {'f': 'f'}:
self.assertTrue(is_iterable(a))
for b in object(), 1:
self.assertFalse(is_iterable(b))
def test_cached_property(self):
def fun(obj):
return fun.value
x = cached_property(fun)
self.assertIs(x.__get__(None), x)
self.assertIs(x.__set__(None, None), x)
self.assertIs(x.__delete__(None), x)
@patch('warnings.warn')
def test_warn_deprecated(self, warn):
warn_deprecated('Foo')
self.assertTrue(warn.called)
|