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
|
import struct
import pytest
from consul import ConsulException
class TestConsul:
def test_kv(self, consul_obj) -> None:
c, _consul_version = consul_obj
_index, data = c.kv.get("foo")
assert data is None
assert c.kv.put("foo", "bar") is True
_index, data = c.kv.get("foo")
assert data["Value"] == b"bar"
def test_kv_wait(self, consul_obj) -> None:
c, _consul_version = consul_obj
assert c.kv.put("foo", "bar") is True
index, _data = c.kv.get("foo")
check, _data = c.kv.get("foo", index=index, wait="20ms")
assert index == check
def test_kv_encoding(self, consul_obj) -> None:
c, _consul_version = consul_obj
# test binary
c.kv.put("foo", struct.pack("i", 1000))
_index, data = c.kv.get("foo")
assert struct.unpack("i", data["Value"]) == (1000,)
# test unicode
c.kv.put("foo", "bar")
_index, data = c.kv.get("foo")
assert data["Value"] == b"bar"
# test empty-string comes back as `None`
c.kv.put("foo", "")
_index, data = c.kv.get("foo")
assert data["Value"] is None
# test None
c.kv.put("foo", None)
_index, data = c.kv.get("foo")
assert data["Value"] is None
# check unencoded values raises assert
pytest.raises(AssertionError, c.kv.put, "foo", {1: 2})
def test_kv_put_cas(self, consul_obj) -> None:
c, _consul_version = consul_obj
assert c.kv.put("foo", "bar", cas=50) is False
assert c.kv.put("foo", "bar", cas=0) is True
_index, data = c.kv.get("foo")
assert c.kv.put("foo", "bar2", cas=data["ModifyIndex"] - 1) is False
assert c.kv.put("foo", "bar2", cas=data["ModifyIndex"]) is True
_index, data = c.kv.get("foo")
assert data["Value"] == b"bar2"
def test_kv_put_flags(self, consul_obj) -> None:
c, _consul_version = consul_obj
c.kv.put("foo", "bar")
_index, data = c.kv.get("foo")
assert data["Flags"] == 0
assert c.kv.put("foo", "bar", flags=50) is True
_index, data = c.kv.get("foo")
assert data["Flags"] == 50
def test_kv_recurse(self, consul_obj) -> None:
c, _consul_version = consul_obj
_index, data = c.kv.get("foo/", recurse=True)
assert data is None
c.kv.put("foo/", None)
_index, data = c.kv.get("foo/", recurse=True)
assert len(data) == 1
c.kv.put("foo/bar1", "1")
c.kv.put("foo/bar2", "2")
c.kv.put("foo/bar3", "3")
_index, data = c.kv.get("foo/", recurse=True)
assert [x["Key"] for x in data] == ["foo/", "foo/bar1", "foo/bar2", "foo/bar3"]
assert [x["Value"] for x in data] == [None, b"1", b"2", b"3"]
def test_kv_delete(self, consul_obj) -> None:
c, _consul_version = consul_obj
c.kv.put("foo1", "1")
c.kv.put("foo2", "2")
c.kv.put("foo3", "3")
_index, data = c.kv.get("foo", recurse=True)
assert [x["Key"] for x in data] == ["foo1", "foo2", "foo3"]
assert c.kv.delete("foo2") is True
_index, data = c.kv.get("foo", recurse=True)
assert [x["Key"] for x in data] == ["foo1", "foo3"]
assert c.kv.delete("foo", recurse=True) is True
_index, data = c.kv.get("foo", recurse=True)
assert data is None
def test_kv_delete_cas(self, consul_obj) -> None:
c, _consul_version = consul_obj
c.kv.put("foo", "bar")
index, data = c.kv.get("foo")
assert c.kv.delete("foo", cas=data["ModifyIndex"] - 1) is False
assert c.kv.get("foo") == (index, data)
assert c.kv.delete("foo", cas=data["ModifyIndex"]) is True
index, data = c.kv.get("foo")
assert data is None
def test_kv_acquire_release(self, consul_obj) -> None:
c, _consul_version = consul_obj
pytest.raises(ConsulException, c.kv.put, "foo", "bar", acquire="foo")
s1 = c.session.create()
s2 = c.session.create()
assert c.kv.put("foo", "1", acquire=s1) is True
assert c.kv.put("foo", "2", acquire=s2) is False
assert c.kv.put("foo", "1", acquire=s1) is True
assert c.kv.put("foo", "1", release="foo") is False
assert c.kv.put("foo", "2", release=s2) is False
assert c.kv.put("foo", "2", release=s1) is True
c.session.destroy(s1)
c.session.destroy(s2)
def test_kv_keys_only(self, consul_obj) -> None:
c, _consul_version = consul_obj
assert c.kv.put("bar", "4") is True
assert c.kv.put("base/foo", "1") is True
assert c.kv.put("base/base/foo", "5") is True
_index, data = c.kv.get("base/", keys=True, separator="/")
assert data == ["base/base/", "base/foo"]
|
