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
|
from __future__ import annotations
import socket
import sys
import types
from unittest.mock import Mock, patch
import pytest
from kombu.utils import compat
from kombu.utils.compat import entrypoints, maybe_fileno
def test_entrypoints():
with patch(
'kombu.utils.compat.importlib_metadata.entry_points', create=True
) as iterep:
eps = [Mock(), Mock()]
iterep.return_value = (
{'kombu.test': eps} if sys.version_info < (3, 10) else eps)
assert list(entrypoints('kombu.test'))
if sys.version_info < (3, 10):
iterep.assert_called_with()
else:
iterep.assert_called_with(group='kombu.test')
eps[0].load.assert_called_with()
eps[1].load.assert_called_with()
def test_maybe_fileno():
assert maybe_fileno(3) == 3
f = Mock(name='file')
assert maybe_fileno(f) is f.fileno()
f.fileno.side_effect = ValueError()
assert maybe_fileno(f) is None
class test_detect_environment:
def test_detect_environment(self):
try:
compat._environment = None
X = compat.detect_environment()
assert compat._environment == X
Y = compat.detect_environment()
assert Y == X
finally:
compat._environment = None
@pytest.mark.ensured_modules('eventlet', 'eventlet.patcher')
def test_detect_environment_eventlet(self, module_exists):
with patch('eventlet.patcher.is_monkey_patched', create=True) as m:
assert sys.modules['eventlet']
m.return_value = True
env = compat._detect_environment()
m.assert_called_with(socket)
assert env == 'eventlet'
@pytest.mark.ensured_modules('gevent')
def test_detect_environment_gevent(self, module_exists):
with patch('gevent.socket', create=True) as m:
prev, socket.socket = socket.socket, m.socket
try:
assert sys.modules['gevent']
env = compat._detect_environment()
assert env == 'gevent'
finally:
socket.socket = prev
def test_detect_environment_no_eventlet_or_gevent(self):
try:
sys.modules['eventlet'] = types.ModuleType('eventlet')
sys.modules['eventlet.patcher'] = types.ModuleType('patcher')
assert compat._detect_environment() == 'default'
finally:
sys.modules.pop('eventlet.patcher', None)
sys.modules.pop('eventlet', None)
compat._detect_environment()
try:
sys.modules['gevent'] = types.ModuleType('gevent')
assert compat._detect_environment() == 'default'
finally:
sys.modules.pop('gevent', None)
compat._detect_environment()
|
