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