blob: 5b0833ddd8ddab287846ff4912f06e712d9f3a29 (
plain)
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
|
import sys
from unittest.mock import Mock, patch
import pytest
from kombu.message import Message
class test_Message:
def test_repr(self):
assert repr(Message('b', channel=Mock()))
def test_decode(self):
m = Message('body', channel=Mock())
decode = m._decode = Mock()
assert m._decoded_cache is None
assert m.decode() is m._decode.return_value
assert m._decoded_cache is m._decode.return_value
m._decode.assert_called_with()
m._decode = Mock()
assert m.decode() is decode.return_value
def test_reraise_error(self):
m = Message('body', channel=Mock())
callback = Mock(name='callback')
try:
raise KeyError('foo')
except KeyError:
m.errors.append(sys.exc_info())
m._reraise_error(callback)
callback.assert_called()
with pytest.raises(KeyError):
m._reraise_error(None)
@patch('kombu.message.decompress')
def test_decompression_stores_error(self, decompress):
decompress.side_effect = RuntimeError()
m = Message('body', channel=Mock(), headers={'compression': 'zlib'})
with pytest.raises(RuntimeError):
m._reraise_error(None)
|
