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
|
import pytest
from redis._parsers import CommandsParser
from .conftest import (
assert_resp_response,
skip_if_redis_enterprise,
skip_if_server_version_lt,
)
class TestCommandsParser:
def test_init_commands(self, r):
commands_parser = CommandsParser(r)
assert commands_parser.commands is not None
assert "get" in commands_parser.commands
def test_get_keys_predetermined_key_location(self, r):
commands_parser = CommandsParser(r)
args1 = ["GET", "foo"]
args2 = ["OBJECT", "encoding", "foo"]
args3 = ["MGET", "foo", "bar", "foobar"]
assert commands_parser.get_keys(r, *args1) == ["foo"]
assert commands_parser.get_keys(r, *args2) == ["foo"]
assert commands_parser.get_keys(r, *args3) == ["foo", "bar", "foobar"]
@pytest.mark.filterwarnings("ignore:ResponseError")
@skip_if_redis_enterprise()
def test_get_moveable_keys(self, r):
commands_parser = CommandsParser(r)
args1 = [
"EVAL",
"return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}",
2,
"key1",
"key2",
"first",
"second",
]
args2 = ["XREAD", "COUNT", 2, b"STREAMS", "mystream", "writers", 0, 0]
args3 = ["ZUNIONSTORE", "out", 2, "zset1", "zset2", "WEIGHTS", 2, 3]
args4 = ["GEORADIUS", "Sicily", 15, 37, 200, "km", "WITHCOORD", b"STORE", "out"]
args5 = ["MEMORY USAGE", "foo"]
args6 = [
"MIGRATE",
"192.168.1.34",
6379,
"",
0,
5000,
b"KEYS",
"key1",
"key2",
"key3",
]
args7 = ["MIGRATE", "192.168.1.34", 6379, "key1", 0, 5000]
assert_resp_response(
r,
sorted(commands_parser.get_keys(r, *args1)),
["key1", "key2"],
[b"key1", b"key2"],
)
assert_resp_response(
r,
sorted(commands_parser.get_keys(r, *args2)),
["mystream", "writers"],
[b"mystream", b"writers"],
)
assert_resp_response(
r,
sorted(commands_parser.get_keys(r, *args3)),
["out", "zset1", "zset2"],
[b"out", b"zset1", b"zset2"],
)
assert_resp_response(
r,
sorted(commands_parser.get_keys(r, *args4)),
["Sicily", "out"],
[b"Sicily", b"out"],
)
assert sorted(commands_parser.get_keys(r, *args5)) in [["foo"], [b"foo"]]
assert_resp_response(
r,
sorted(commands_parser.get_keys(r, *args6)),
["key1", "key2", "key3"],
[b"key1", b"key2", b"key3"],
)
assert_resp_response(
r, sorted(commands_parser.get_keys(r, *args7)), ["key1"], [b"key1"]
)
# A bug in redis<7.0 causes this to fail: https://github.com/redis/redis/issues/9493
@skip_if_server_version_lt("7.0.0")
def test_get_eval_keys_with_0_keys(self, r):
commands_parser = CommandsParser(r)
args = ["EVAL", "return {ARGV[1],ARGV[2]}", 0, "key1", "key2"]
assert commands_parser.get_keys(r, *args) == []
def test_get_pubsub_keys(self, r):
commands_parser = CommandsParser(r)
args1 = ["PUBLISH", "foo", "bar"]
args2 = ["PUBSUB NUMSUB", "foo1", "foo2", "foo3"]
args3 = ["PUBSUB channels", "*"]
args4 = ["SUBSCRIBE", "foo1", "foo2", "foo3"]
assert commands_parser.get_keys(r, *args1) == ["foo"]
assert commands_parser.get_keys(r, *args2) == ["foo1", "foo2", "foo3"]
assert commands_parser.get_keys(r, *args3) == ["*"]
assert commands_parser.get_keys(r, *args4) == ["foo1", "foo2", "foo3"]
|
