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
|
import pytest
from redis.exceptions import ResponseError
from .conftest import 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"]]],
]
]
assert r.function_list() == res
assert r.function_list(library="*lib") == res
assert (
r.function_list(withcode=True)[0][7]
== f"#!{engine} name={lib} \n {function}".encode()
)
@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"]]],
]
]
primaries = r.get_primaries()
res = {}
for node in primaries:
res[node.name] = function_list
assert r.function_list() == res
assert r.function_list(library="*lib") == res
node = primaries[0].name
assert (
r.function_list(withcode=True)[node][0][7]
== f"#!{engine} name={lib} \n {function}".encode()
)
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")
|
