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
|
import pytest
import redis
import redis.asyncio
from fakeredis import FakeServer, aioredis, FakeAsyncRedis, FakeStrictRedis
from test import testtools
pytestmark = []
fake_only = pytest.mark.parametrize("async_redis", [pytest.param("fake", marks=pytest.mark.fake)], indirect=True)
pytestmark.extend(
[
pytest.mark.asyncio,
]
)
@fake_only
@pytest.mark.disconnected
async def test_not_connected(async_redis: redis.asyncio.Redis):
with pytest.raises(redis.asyncio.ConnectionError):
await async_redis.ping()
@fake_only
async def test_disconnect_server(async_redis, fake_server):
await async_redis.ping()
fake_server.connected = False
with pytest.raises(redis.asyncio.ConnectionError):
await async_redis.ping()
fake_server.connected = True
@pytest.mark.fake
async def test_from_url():
r0 = aioredis.FakeRedis.from_url("redis://localhost?db=0")
r1 = aioredis.FakeRedis.from_url("redis://localhost?db=1")
# Check that they are indeed different databases
await r0.set("foo", "a")
await r1.set("foo", "b")
assert await r0.get("foo") == b"a"
assert await r1.get("foo") == b"b"
await r0.connection_pool.disconnect()
await r1.connection_pool.disconnect()
@pytest.mark.fake
async def test_from_url_with_version():
r0 = aioredis.FakeRedis.from_url("redis://localhost?db=0", version=(6,))
r1 = aioredis.FakeRedis.from_url("redis://localhost?db=1", version=(6,))
# Check that they are indeed different databases
await r0.set("foo", "a")
await r1.set("foo", "b")
assert await r0.get("foo") == b"a"
assert await r1.get("foo") == b"b"
await r0.connection_pool.disconnect()
await r1.connection_pool.disconnect()
@fake_only
async def test_from_url_with_server(async_redis, fake_server):
r2 = aioredis.FakeRedis.from_url("redis://localhost", server=fake_server)
await async_redis.set("foo", "bar")
assert await r2.get("foo") == b"bar"
await r2.connection_pool.disconnect()
@pytest.mark.fake
async def test_without_server():
r = aioredis.FakeRedis()
assert await r.ping()
@pytest.mark.fake
async def test_without_server_disconnected():
r = aioredis.FakeRedis(connected=False)
with pytest.raises(redis.asyncio.ConnectionError):
await r.ping()
@pytest.mark.fake
async def test_async():
# arrange
cache = aioredis.FakeRedis()
# act
await cache.set("fakeredis", "plz")
x = await cache.get("fakeredis")
# assert
assert x == b"plz"
@testtools.run_test_if_redispy_ver("gte", "4.4.0")
@pytest.mark.parametrize("nowait", [False, True])
@pytest.mark.fake
async def test_connection_disconnect(nowait):
server = FakeServer()
r = aioredis.FakeRedis(server=server)
conn = await r.connection_pool.get_connection("_")
assert conn is not None
await conn.disconnect(nowait=nowait)
assert conn._sock is None
@pytest.mark.fake
async def test_init_args():
sync_r1 = FakeStrictRedis()
r1 = FakeAsyncRedis()
r5 = FakeAsyncRedis()
r2 = FakeAsyncRedis(server=FakeServer())
shared_server = FakeServer()
r3 = FakeAsyncRedis(server=shared_server)
r4 = FakeAsyncRedis(server=shared_server)
await r1.set("foo", "bar")
await r3.set("bar", "baz")
assert await r1.get("foo") == b"bar"
assert await r5.get("foo") is None
assert sync_r1.get("foo") is None
assert await r2.get("foo") is None
assert await r3.get("foo") is None
assert await r3.get("bar") == b"baz"
assert await r4.get("bar") == b"baz"
assert await r1.get("bar") is None
|
