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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
import pytest
import multiprocessing
import contextlib
from redis.connection import Connection, ConnectionPool
from redis.exceptions import ConnectionError
@contextlib.contextmanager
def exit_callback(callback, *args):
try:
yield
finally:
callback(*args)
class TestMultiprocessing(object):
# Test connection sharing between forks.
# See issue #1085 for details.
def test_connection(self):
conn = Connection()
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
def target(conn):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
conn.disconnect()
proc = multiprocessing.Process(target=target, args=(conn,))
proc.start()
proc.join(3)
assert proc.exitcode is 0
# Check that connection is still alive after fork process has exited.
with pytest.raises(ConnectionError):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
def test_close_connection_in_main(self):
conn = Connection()
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
def target(conn, ev):
ev.wait()
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
ev = multiprocessing.Event()
proc = multiprocessing.Process(target=target, args=(conn, ev))
proc.start()
conn.disconnect()
ev.set()
proc.join(3)
assert proc.exitcode is 1
@pytest.mark.parametrize('max_connections', [1, 2, None])
def test_pool(self, max_connections):
pool = ConnectionPool.from_url('redis://localhost',
max_connections=max_connections)
conn = pool.get_connection('ping')
with exit_callback(pool.release, conn):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
def target(pool):
with exit_callback(pool.disconnect):
conn = pool.get_connection('ping')
with exit_callback(pool.release, conn):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
proc = multiprocessing.Process(target=target, args=(pool,))
proc.start()
proc.join(3)
assert proc.exitcode is 0
# Check that connection is still alive after fork process has exited.
conn = pool.get_connection('ping')
with exit_callback(pool.release, conn):
with pytest.raises(ConnectionError):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
@pytest.mark.parametrize('max_connections', [1, 2, None])
def test_close_pool_in_main(self, max_connections):
pool = ConnectionPool.from_url('redis://localhost',
max_connections=max_connections)
conn = pool.get_connection('ping')
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
def target(pool, disconnect_event):
conn = pool.get_connection('ping')
with exit_callback(pool.release, conn):
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
disconnect_event.wait()
assert conn.send_command('ping') is None
assert conn.read_response() == b'PONG'
ev = multiprocessing.Event()
proc = multiprocessing.Process(target=target, args=(pool, ev))
proc.start()
pool.disconnect()
ev.set()
proc.join(3)
assert proc.exitcode is 0
def test_redis(self, r):
assert r.ping() is True
def target(redis):
assert redis.ping() is True
del redis
proc = multiprocessing.Process(target=target, args=(r,))
proc.start()
proc.join(3)
assert proc.exitcode is 0
assert r.ping() is True
|
