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
|
import pytest
from fakeredis._command_args_parsing import extract_args
from fakeredis._helpers import SimpleError
def test_extract_args():
args = (
b"nx",
b"ex",
b"324",
b"xx",
)
(xx, nx, ex, keepttl), _ = extract_args(args, ("nx", "xx", "+ex", "keepttl"))
assert xx
assert nx
assert ex == 324
assert not keepttl
def test_extract_args__should_raise_error():
args = (b"nx", b"ex", b"324", b"xx", b"something")
with pytest.raises(SimpleError):
_, _ = extract_args(args, ("nx", "xx", "+ex", "keepttl"))
def test_extract_args__should_return_something():
args = (b"nx", b"ex", b"324", b"xx", b"something")
(xx, nx, ex, keepttl), left = extract_args(args, ("nx", "xx", "+ex", "keepttl"), error_on_unexpected=False)
assert xx
assert nx
assert ex == 324
assert not keepttl
assert left == (b"something",)
args = (
b"nx",
b"something",
b"ex",
b"324",
b"xx",
)
(xx, nx, ex, keepttl), left = extract_args(
args, ("nx", "xx", "+ex", "keepttl"), error_on_unexpected=False, left_from_first_unexpected=False
)
assert xx
assert nx
assert ex == 324
assert not keepttl
assert left == [
b"something",
]
def test_extract_args__multiple_numbers():
args = (
b"nx",
b"limit",
b"324",
b"123",
b"xx",
)
(xx, nx, limit, keepttl), _ = extract_args(args, ("nx", "xx", "++limit", "keepttl"))
assert xx
assert nx
assert limit == [324, 123]
assert not keepttl
(xx, nx, limit, keepttl), _ = extract_args(
(
b"nx",
b"xx",
),
("nx", "xx", "++limit", "keepttl"),
)
assert xx
assert nx
assert not keepttl
assert limit == [None, None]
def test_extract_args__extract_non_numbers():
args = (
b"by",
b"dd",
b"nx",
b"limit",
b"324",
b"123",
b"xx",
)
(xx, nx, limit, sortby), _ = extract_args(args, ("nx", "xx", "++limit", "*by"))
assert xx
assert nx
assert limit == [324, 123]
assert sortby == b"dd"
def test_extract_args__extract_maxlen():
args = (b"MAXLEN", b"5")
(nomkstream, limit, maxlen, maxid), left_args = extract_args(
args, ("nomkstream", "+limit", "~+maxlen", "~maxid"), error_on_unexpected=False
)
assert not nomkstream
assert limit is None
assert maxlen == 5
assert maxid is None
args = (b"MAXLEN", b"~", b"5", b"maxid", b"~", b"1")
(nomkstream, limit, maxlen, maxid), left_args = extract_args(
args, ("nomkstream", "+limit", "~+maxlen", "~maxid"), error_on_unexpected=False
)
assert not nomkstream
assert limit is None
assert maxlen == 5
assert maxid == b"1"
args = (
b"by",
b"dd",
b"nx",
b"maxlen",
b"~",
b"10",
b"limit",
b"324",
b"123",
b"xx",
)
(nx, maxlen, xx, limit, sortby), _ = extract_args(args, ("nx", "~+maxlen", "xx", "++limit", "*by"))
assert xx
assert nx
assert maxlen == 10
assert limit == [324, 123]
assert sortby == b"dd"
|
