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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
import pytest
import redis
import redis.client
from packaging.version import Version
from test import testtools
from test.testtools import raw_command
REDIS_VERSION = Version(redis.__version__)
def test_zadd_roman(r: redis.Redis):
testtools.raw_command(r, "ZADD", "b", "0", "c")
with pytest.raises(redis.ResponseError) as ctx:
testtools.raw_command(r, "SORT", "b")
assert str(ctx.value) == "One or more scores can't be converted into double"
def test_zadd(r: redis.Redis):
r.zadd("foo", {"four": 4})
r.zadd("foo", {"three": 3})
assert r.zadd("foo", {"two": 2, "one": 1, "zero": 0}) == 3
assert r.zrange("foo", 0, -1) == [b"zero", b"one", b"two", b"three", b"four"]
assert r.zadd("foo", {"zero": 7, "one": 1, "five": 5}) == 1
assert r.zrange("foo", 0, -1) == [b"one", b"two", b"three", b"four", b"five", b"zero"]
def test_zadd_empty(r: redis.Redis):
# Have to add at least one key/value pair
with pytest.raises(redis.RedisError):
r.zadd("foo", {})
@pytest.mark.max_server("6.2.7")
def test_zadd_minus_zero_redis6(r: redis.Redis):
# Changing -0 to +0 is ignored
r.zadd("foo", {"a": -0.0})
r.zadd("foo", {"a": 0.0})
assert raw_command(r, "zscore", "foo", "a") == b"-0"
@pytest.mark.min_server("7")
def test_zadd_minus_zero_redis7(r: redis.Redis):
r.zadd("foo", {"a": -0.0})
r.zadd("foo", {"a": 0.0})
assert raw_command(r, "zscore", "foo", "a") == b"0"
def test_zadd_wrong_type(r: redis.Redis):
r.sadd("foo", "bar")
with pytest.raises(redis.ResponseError):
r.zadd("foo", {"two": 2})
def test_zadd_multiple(r: redis.Redis):
r.zadd("foo", {"one": 1, "two": 2})
assert r.zrange("foo", 0, 0) == [b"one"]
assert r.zrange("foo", 1, 1) == [b"two"]
@pytest.mark.parametrize(
["map_to_add", "expected_ret", "expected_zrange_state"],
[
({"four": 2.0, "three": 1.0}, 0, [(b"three", 3.0), (b"four", 4.0)]),
({"four": 2.0, "three": 1.0, "zero": 0.0}, 1, [(b"zero", 0.0), (b"three", 3.0), (b"four", 4.0)]),
({"two": 2.0, "one": 1.0}, 2, [(b"one", 1.0), (b"two", 2.0), (b"three", 3.0), (b"four", 4.0)]),
],
ids=["no_additions", "partial_additions", "new_additions"],
)
@pytest.mark.parametrize("ch", [False, True])
def test_zadd_with_nx(r: redis.Redis, map_to_add, expected_ret, expected_zrange_state, ch):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", map_to_add, nx=True, ch=ch) == expected_ret
assert r.zrange("foo", 0, -1, withscores=True) == expected_zrange_state
@pytest.mark.parametrize(
["map_to_add", "expected_ret", "expected_zrange_state"],
[
({"four": 2.0, "three": 1.0}, 0, [(b"three", 3.0), (b"four", 4.0)]),
(
{"four": 5.0, "three": 1.0, "zero": 0.0},
2,
[
(b"zero", 0.0),
(b"three", 3.0),
(b"four", 5.0),
],
),
({"two": 2.0, "one": 1.0}, 2, [(b"one", 1.0), (b"two", 2.0), (b"three", 3.0), (b"four", 4.0)]),
],
ids=["no_additions", "partial_additions", "new_additions"],
)
def test_zadd_with_gt_and_ch(r: redis.Redis, map_to_add, expected_ret, expected_zrange_state):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", map_to_add, gt=True, ch=True) == expected_ret
assert r.zrange("foo", 0, -1, withscores=True) == expected_zrange_state
@pytest.mark.parametrize(
["map_to_add", "expected_ret", "expected_zrange_state"],
[
({"four": 2.0, "three": 1.0}, 0, [(b"three", 3.0), (b"four", 4.0)]),
({"four": 5.0, "three": 1.0, "zero": 0.0}, 1, [(b"zero", 0.0), (b"three", 3.0), (b"four", 5.0)]),
({"two": 2.0, "one": 1.0}, 2, [(b"one", 1.0), (b"two", 2.0), (b"three", 3.0), (b"four", 4.0)]),
],
ids=["no_additions", "partial_additions", "new_additions"],
)
def test_zadd_with_gt(r: redis.Redis, map_to_add, expected_ret, expected_zrange_state):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", map_to_add, gt=True) == expected_ret
assert r.zrange("foo", 0, -1, withscores=True) == expected_zrange_state
@pytest.mark.parametrize(
["map_to_add", "expected_ret", "expected_zrange_state"],
[
({"four": 4.0, "three": 1.0}, 1, [(b"three", 1.0), (b"four", 4.0)]),
({"four": 4.0, "three": 1.0, "zero": 0.0}, 2, [(b"zero", 0.0), (b"three", 1.0), (b"four", 4.0)]),
({"two": 2.0, "one": 1.0}, 2, [(b"one", 1.0), (b"two", 2.0), (b"three", 3.0), (b"four", 4.0)]),
],
ids=["no_additions", "partial_additions", "new_additions"],
)
def test_zadd_with_ch(r: redis.Redis, map_to_add, expected_ret, expected_zrange_state):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", map_to_add, ch=True) == expected_ret
assert r.zrange("foo", 0, -1, withscores=True) == expected_zrange_state
@pytest.mark.parametrize(
["map_to_add", "expected_ret", "expected_zrange_state"],
[
({"four": 2.0, "three": 1.0}, 2, [(b"three", 1.0), (b"four", 2.0)]),
({"four": 4.0, "three": 3.0, "zero": 0.0}, 0, [(b"three", 3.0), (b"four", 4.0)]),
({"two": 2.0, "one": 1.0}, 0, [(b"three", 3.0), (b"four", 4.0)]),
],
ids=["new_additions", "no_additions", "no_additions_2"],
)
@pytest.mark.parametrize("ch", [False, True])
def test_zadd_with_xx(r: redis.Redis, map_to_add, expected_ret, expected_zrange_state, ch):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", map_to_add, xx=True, ch=ch) == (expected_ret if ch else 0)
assert r.zrange("foo", 0, -1, withscores=True) == expected_zrange_state
@pytest.mark.parametrize("ch", [False, True])
def test_zadd_with_nx_and_xx(r: redis.Redis, ch):
r.zadd("foo", {"four": 4.0, "three": 3.0})
with pytest.raises(redis.DataError):
r.zadd("foo", {"four": -4.0, "three": -3.0}, nx=True, xx=True, ch=ch)
@pytest.mark.parametrize("ch", [False, True])
def test_zadd_incr(r: redis.Redis, ch):
r.zadd("foo", {"four": 4.0, "three": 3.0})
assert r.zadd("foo", {"four": 1.0}, incr=True, ch=ch) == 5.0
assert r.zadd("foo", {"three": 1.0}, incr=True, nx=True, ch=ch) is None
assert r.zscore("foo", "three") == 3.0
assert r.zadd("foo", {"bar": 1.0}, incr=True, xx=True, ch=ch) is None
assert r.zadd("foo", {"three": 1.0}, incr=True, xx=True, ch=ch) == 4.0
def test_zadd_with_xx_and_gt_and_ch(r: redis.Redis):
r.zadd("test", {"one": 1})
assert r.zscore("test", "one") == 1.0
assert r.zadd("test", {"one": 4}, xx=True, gt=True, ch=True) == 1
assert r.zscore("test", "one") == 4.0
assert r.zadd("test", {"one": 0}, xx=True, gt=True, ch=True) == 0
assert r.zscore("test", "one") == 4.0
def test_zadd_and_zrangebyscore(r: redis.Redis):
raw_command(r, "zadd", "", 0.0, "")
assert raw_command(r, "zrangebyscore", "", 0.0, 0.0, "limit", 0, 0) == []
with pytest.raises(redis.RedisError):
raw_command(r, "zrangebyscore", "", 0.0, 0.0, "limit", 0)
with pytest.raises(redis.RedisError):
raw_command(r, "zadd", "t", 0.0, "xx", "")
|
