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
|
import pytest
from redis.exceptions import ResponseError
from .conftest import assert_resp_response, skip_if_server_version_lt
engine = "lua"
lib = "mylib"
lib2 = "mylib2"
function = "redis.register_function{function_name='myfunc', callback=function(keys, \
args) return args[1] end, flags={ 'no-writes' }}"
function2 = "redis.register_function('hello', function() return 'Hello World' end)"
set_function = "redis.register_function('set', function(keys, args) return \
redis.call('SET', keys[1], args[1]) end)"
get_function = "redis.register_function('get', function(keys, args) return \
redis.call('GET', keys[1]) end)"
@skip_if_server_version_lt("7.0.0")
class TestFunction:
@pytest.fixture(autouse=True)
def reset_functions(self, r):
r.function_flush()
@pytest.mark.onlynoncluster
def test_function_load(self, r):
assert b"mylib" == r.function_load(f"#!{engine} name={lib} \n {function}")
assert b"mylib" == r.function_load(
f"#!{engine} name={lib} \n {function}", replace=True
)
with pytest.raises(ResponseError):
r.function_load(f"#!{engine} name={lib} \n {function}")
with pytest.raises(ResponseError):
r.function_load(f"#!{engine} name={lib2} \n {function}")
def test_function_delete(self, r):
r.function_load(f"#!{engine} name={lib} \n {set_function}")
with pytest.raises(ResponseError):
r.function_load(f"#!{engine} name={lib} \n {set_function}")
assert r.fcall("set", 1, "foo", "bar") == b"OK"
assert r.function_delete("mylib")
with pytest.raises(ResponseError):
r.fcall("set", 1, "foo", "bar")
def test_function_flush(self, r):
r.function_load(f"#!{engine} name={lib} \n {function}")
assert r.fcall("myfunc", 0, "hello") == b"hello"
assert r.function_flush()
with pytest.raises(ResponseError):
r.fcall("myfunc", 0, "hello")
with pytest.raises(ResponseError):
r.function_flush("ABC")
@pytest.mark.onlynoncluster
def test_function_list(self, r):
r.function_load(f"#!{engine} name={lib} \n {function}")
res = [
[
b"library_name",
b"mylib",
b"engine",
b"LUA",
b"functions",
[[b"name", b"myfunc", b"description", None, b"flags", [b"no-writes"]]],
]
]
resp3_res = [
{
b"library_name": b"mylib",
b"engine": b"LUA",
b"functions": [
{b"name": b"myfunc", b"description": None, b"flags": [b"no-writes"]}
],
}
]
assert_resp_response(r, r.function_list(), res, resp3_res)
assert_resp_response(r, r.function_list(library="*lib"), res, resp3_res)
res[0].extend(
[b"library_code", f"#!{engine} name={lib} \n {function}".encode()]
)
resp3_res[0][b"library_code"] = f"#!{engine} name={lib} \n {function}".encode()
assert_resp_response(r, r.function_list(withcode=True), res, resp3_res)
@pytest.mark.onlycluster
def test_function_list_on_cluster(self, r):
r.function_load(f"#!{engine} name={lib} \n {function}")
function_list = [
[
b"library_name",
b"mylib",
b"engine",
b"LUA",
b"functions",
[[b"name", b"myfunc", b"description", None, b"flags", [b"no-writes"]]],
]
]
resp3_function_list = [
{
b"library_name": b"mylib",
b"engine": b"LUA",
b"functions": [
{b"name": b"myfunc", b"description": None, b"flags": [b"no-writes"]}
],
}
]
primaries = r.get_primaries()
res = {}
resp3_res = {}
for node in primaries:
res[node.name] = function_list
resp3_res[node.name] = resp3_function_list
assert_resp_response(r, r.function_list(), res, resp3_res)
assert_resp_response(r, r.function_list(library="*lib"), res, resp3_res)
node = primaries[0].name
code = f"#!{engine} name={lib} \n {function}".encode()
res[node][0].extend([b"library_code", code])
resp3_res[node][0][b"library_code"] = code
assert_resp_response(r, r.function_list(withcode=True), res, resp3_res)
def test_fcall(self, r):
r.function_load(f"#!{engine} name={lib} \n {set_function}")
r.function_load(f"#!{engine} name={lib2} \n {get_function}")
assert r.fcall("set", 1, "foo", "bar") == b"OK"
assert r.fcall("get", 1, "foo") == b"bar"
with pytest.raises(ResponseError):
r.fcall("myfunc", 0, "hello")
def test_fcall_ro(self, r):
r.function_load(f"#!{engine} name={lib} \n {function}")
assert r.fcall_ro("myfunc", 0, "hello") == b"hello"
r.function_load(f"#!{engine} name={lib2} \n {set_function}")
with pytest.raises(ResponseError):
r.fcall_ro("set", 1, "foo", "bar")
def test_function_dump_restore(self, r):
r.function_load(f"#!{engine} name={lib} \n {set_function}")
payload = r.function_dump()
assert r.fcall("set", 1, "foo", "bar") == b"OK"
r.function_delete("mylib")
with pytest.raises(ResponseError):
r.fcall("set", 1, "foo", "bar")
assert r.function_restore(payload)
assert r.fcall("set", 1, "foo", "bar") == b"OK"
r.function_load(f"#!{engine} name={lib2} \n {get_function}")
assert r.fcall("get", 1, "foo") == b"bar"
r.function_delete("mylib")
assert r.function_restore(payload, "FLUSH")
with pytest.raises(ResponseError):
r.fcall("get", 1, "foo")
|
