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
|
import pytest
import redis
from test import testtools
json_tests = pytest.importorskip("probables")
@pytest.mark.unsupported_server_types("dragonfly")
def test_cms_create(r: redis.Redis):
assert r.cms().initbydim("cmsDim", 100, 5)
assert r.cms().initbyprob("cmsProb", 0.01, 0.01)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbydim("cmsDim", 1, 5)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbydim("cmsDim2", 0, 5)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbydim("cmsDim2", 3, 0)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbyprob("cmsProb", 0.01, 0.1)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbyprob("cmsProb2", 2, 0.01)
with pytest.raises(redis.exceptions.ResponseError):
r.cms().initbyprob("cmsProb2", 0.01, 0)
@pytest.mark.unsupported_server_types("dragonfly")
def test_cms_incrby(r: redis.Redis):
assert r.cms().initbydim("cmsDim", 100, 5)
assert r.cms().initbyprob("cmsProb", 0.01, 0.01)
assert r.cms().incrby("cmsDim", ["foo"], [3]) == [3]
assert r.cms().incrby("cmsDim", ["foo", "bar"], [4, 1]) == [7, 1]
assert r.cms().query("cmsDim", "foo") == [7]
assert r.cms().query("cmsDim", "foo", "bar") == [7, 1]
assert r.cms().query("cmsDim", "noexist") == [0]
with pytest.raises(redis.exceptions.ResponseError):
r.cms().query("cmsDim")
with pytest.raises(redis.exceptions.ResponseError):
r.cms().query("noexist", "foo")
with pytest.raises(redis.exceptions.ResponseError):
testtools.raw_command(r, "CMS.INCRBY", "cmsDim", "foo", 1, "bar")
with pytest.raises(redis.exceptions.ResponseError, match="CMS: key does not exist"):
r.cms().incrby("noexist", ["foo", "bar"], [3, 4])
with pytest.raises(redis.exceptions.ResponseError, match="CMS: Cannot parse number"):
r.cms().incrby("cmsDim", ["foo", "bar"], [3, "four"])
@pytest.mark.unsupported_server_types("dragonfly")
def test_cms_merge(r: redis.Redis):
assert r.cms().initbydim("cmsDim", 100, 5)
assert r.cms().initbydim("cms2", 100, 5)
assert r.cms().incrby("cmsDim", ["foo"], [3]) == [3]
assert r.cms().incrby("cms2", ["foo", "bar"], [4, 1]) == [4, 1]
assert r.cms().merge("cmsDim", 1, ["cms2"])
assert r.cms().query("cmsDim", "foo", "bar") == [4, 1]
with pytest.raises(redis.exceptions.ResponseError, match="CMS: key does not exist"):
r.cms().merge("noexist", 1, ["cms2"])
with pytest.raises(redis.exceptions.ResponseError, match="CMS: Number of keys must be positive"):
r.cms().merge("cms2", 0, ["cmsDim"])
with pytest.raises(redis.exceptions.ResponseError, match="wrong number of arguments for '.*' command"):
r.cms().merge("cms2", 1, [])
with pytest.raises(redis.exceptions.ResponseError, match="CMS: wrong number of keys/weights"):
r.cms().merge("cmsDim", 1, ["cms2", "cms1"], [4, 3])
with pytest.raises(redis.exceptions.ResponseError, match="CMS: key does not exist"):
r.cms().merge("cmsDim", 2, ["cms2", "noexist"], [4, 3])
@pytest.mark.unsupported_server_types("dragonfly")
def test_cms_info(r: redis.Redis):
assert r.cms().initbydim("A", 1000, 5)
assert r.cms().initbydim("B", 1000, 5)
assert r.cms().initbydim("C", 1000, 5)
assert r.cms().incrby("A", ["foo", "bar", "baz"], [5, 3, 9])
assert r.cms().incrby("B", ["foo", "bar", "baz"], [2, 3, 1])
assert r.cms().query("A", "foo", "bar", "baz") == [5, 3, 9]
assert r.cms().query("B", "foo", "bar", "baz") == [2, 3, 1]
assert r.cms().merge("C", 2, ["A", "B"])
assert r.cms().query("C", "foo", "bar", "baz") == [7, 6, 10]
assert r.cms().merge("C", 2, ["A", "B"], ["1", "2"])
assert r.cms().query("C", "foo", "bar", "baz") == [9, 9, 11]
assert r.cms().merge("C", 2, ["A", "B"], ["2", "3"])
assert r.cms().query("C", "foo", "bar", "baz") == [16, 15, 21]
info = r.cms().info("A")
assert info.width == 1000
assert info.depth == 5
assert info.count == 17
with pytest.raises(redis.exceptions.ResponseError, match="CMS: key does not exist"):
r.cms().info("noexist")
@pytest.mark.xfail(reason="Bug in pyprobables")
@pytest.mark.unsupported_server_types("dragonfly")
def test_cms_merge_fail(r: redis.Redis):
assert r.cms().initbydim("A", 1000, 5)
assert r.cms().initbydim("B", 1000, 5)
assert r.cms().initbydim("C", 1000, 5)
assert r.cms().incrby("A", ["foo", "bar", "baz"], [5, 3, 9])
assert r.cms().incrby("B", ["foo", "bar", "baz"], [2, 3, 1])
assert r.cms().query("A", "foo", "bar", "baz") == [5, 3, 9]
assert r.cms().query("B", "foo", "bar", "baz") == [2, 3, 1]
assert r.cms().merge("C", 2, ["A", "B"])
assert r.cms().query("C", "foo", "bar", "baz") == [7, 6, 10]
assert r.cms().merge("C", 2, ["A", "B"], ["2", "3"])
info = r.cms().info("C")
assert info.width == 1000
assert info.depth == 5
assert info.count == 52
|
