1from __future__ import absolute_import, unicode_literals 2 3import pytest 4from case import Mock 5 6from celery.utils import debug 7 8 9def test_on_blocking(patching): 10 getframeinfo = patching('inspect.getframeinfo') 11 frame = Mock(name='frame') 12 with pytest.raises(RuntimeError): 13 debug._on_blocking(1, frame) 14 getframeinfo.assert_called_with(frame) 15 16 17def test_blockdetection(patching): 18 signals = patching('celery.utils.debug.signals') 19 with debug.blockdetection(10): 20 signals.arm_alarm.assert_called_with(10) 21 signals.__setitem__.assert_called_with('ALRM', debug._on_blocking) 22 signals.__setitem__.assert_called_with('ALRM', signals['ALRM']) 23 signals.reset_alarm.assert_called_with() 24 25 26def test_sample_mem(patching): 27 mem_rss = patching('celery.utils.debug.mem_rss') 28 prev, debug._mem_sample = debug._mem_sample, [] 29 try: 30 debug.sample_mem() 31 assert debug._mem_sample[0] is mem_rss() 32 finally: 33 debug._mem_sample = prev 34 35 36def test_sample(): 37 x = list(range(100)) 38 assert list(debug.sample(x, 10)) == [ 39 0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 40 ] 41 x = list(range(91)) 42 assert list(debug.sample(x, 10)) == [ 43 0, 9, 18, 27, 36, 45, 54, 63, 72, 81, 44 ] 45 46 47@pytest.mark.parametrize('f,precision,expected', [ 48 (10, 5, '10'), 49 (10.45645234234, 5, '10.456'), 50]) 51def test_hfloat(f, precision, expected): 52 assert str(debug.hfloat(f, precision)) == expected 53 54 55@pytest.mark.parametrize('byt,expected', [ 56 (2 ** 20, '1MB'), 57 (4 * 2 ** 20, '4MB'), 58 (2 ** 16, '64KB'), 59 (2 ** 16, '64KB'), 60 (2 ** 8, '256b'), 61]) 62def test_humanbytes(byt, expected): 63 assert debug.humanbytes(byt) == expected 64 65 66def test_mem_rss(patching): 67 humanbytes = patching('celery.utils.debug.humanbytes') 68 ps = patching('celery.utils.debug.ps') 69 ret = debug.mem_rss() 70 ps.assert_called_with() 71 ps().memory_info.assert_called_with() 72 humanbytes.assert_called_with(ps().memory_info().rss) 73 assert ret is humanbytes() 74 ps.return_value = None 75 assert debug.mem_rss() is None 76 77 78def test_ps(patching): 79 Process = patching('celery.utils.debug.Process') 80 getpid = patching('os.getpid') 81 prev, debug._process = debug._process, None 82 try: 83 debug.ps() 84 Process.assert_called_with(getpid()) 85 assert debug._process is Process() 86 finally: 87 debug._process = prev 88