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
|
import pytest
from .conftest import (
skip_if_redis_enterprise,
skip_ifnot_redis_enterprise,
wait_for_command,
)
@pytest.mark.onlynoncluster
class TestMonitor:
def test_wait_command_not_found(self, r):
"Make sure the wait_for_command func works when command is not found"
with r.monitor() as m:
response = wait_for_command(r, m, "nothing")
assert response is None
def test_response_values(self, r):
db = r.connection_pool.connection_kwargs.get("db", 0)
with r.monitor() as m:
r.ping()
response = wait_for_command(r, m, "PING")
assert isinstance(response["time"], float)
assert response["db"] == db
assert response["client_type"] in ("tcp", "unix")
assert isinstance(response["client_address"], str)
assert isinstance(response["client_port"], str)
assert response["command"] == "PING"
def test_command_with_quoted_key(self, r):
with r.monitor() as m:
r.get('foo"bar')
response = wait_for_command(r, m, 'GET foo"bar')
assert response["command"] == 'GET foo"bar'
def test_command_with_binary_data(self, r):
with r.monitor() as m:
byte_string = b"foo\x92"
r.get(byte_string)
response = wait_for_command(r, m, "GET foo\\x92")
assert response["command"] == "GET foo\\x92"
def test_command_with_escaped_data(self, r):
with r.monitor() as m:
byte_string = b"foo\\x92"
r.get(byte_string)
response = wait_for_command(r, m, "GET foo\\\\x92")
assert response["command"] == "GET foo\\\\x92"
@skip_if_redis_enterprise()
def test_lua_script(self, r):
with r.monitor() as m:
script = 'return redis.call("GET", "foo")'
assert r.eval(script, 0) is None
response = wait_for_command(r, m, "GET foo")
assert response["command"] == "GET foo"
assert response["client_type"] == "lua"
assert response["client_address"] == "lua"
assert response["client_port"] == ""
@skip_ifnot_redis_enterprise()
def test_lua_script_in_enterprise(self, r):
with r.monitor() as m:
script = 'return redis.call("GET", "foo")'
assert r.eval(script, 0) is None
response = wait_for_command(r, m, "GET foo")
assert response is None
|
