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
|
"""Test snapshot action functionality"""
# pylint: disable=missing-function-docstring, missing-class-docstring, line-too-long
import os
from datetime import datetime, timedelta, timezone
from curator.helpers.getters import get_indices, get_snapshot
from curator.exceptions import FailedExecution
from . import CuratorTestCase
from . import testvars
HOST = os.environ.get('TEST_ES_SERVER', 'http://127.0.0.1:9200')
class TestActionFileSnapshot(CuratorTestCase):
def test_snapshot(self):
self.create_indices(5)
self.create_repository()
snap_name = 'snapshot1'
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'],
testvars.snapshot_test.format(self.args['repository'], snap_name, 1, 30),
)
self.invoke_runner()
snapshot = get_snapshot(self.client, self.args['repository'], '*')
assert 1 == len(snapshot['snapshots'])
assert snap_name == snapshot['snapshots'][0]['snapshot']
def test_snapshot_datemath(self):
self.create_indices(5)
self.create_repository()
snap_name = '<snapshot-{now-1d/d}>'
_ = (datetime.now(timezone.utc) - timedelta(days=1)).strftime('%Y.%m.%d')
snap_name_parsed = f"snapshot-{_}"
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'],
testvars.snapshot_test.format(self.args['repository'], snap_name, 1, 30),
)
self.invoke_runner()
snapshot = get_snapshot(self.client, self.args['repository'], '*')
assert 1 == len(snapshot['snapshots'])
assert snap_name_parsed == snapshot['snapshots'][0]['snapshot']
def test_snapshot_ignore_empty_list(self):
self.create_indices(5)
self.create_repository()
snap_name = 'snapshot1'
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'],
testvars.test_682.format(self.args['repository'], snap_name, True, 1, 30),
)
self.invoke_runner()
snapshot = {'snapshots': []}
try:
snapshot = get_snapshot(self.client, self.args['repository'], '*')
except FailedExecution:
pass
assert 0 == len(snapshot['snapshots'])
assert 0 == len(get_indices(self.client))
def test_snapshot_do_not_ignore_empty_list(self):
self.create_indices(5)
self.create_repository()
snap_name = 'snapshot1'
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'],
testvars.test_682.format(self.args['repository'], snap_name, False, 1, 30),
)
self.invoke_runner()
snapshot = {'snapshots': []}
try:
snapshot = get_snapshot(self.client, self.args['repository'], '*')
except FailedExecution:
pass
assert 0 == len(snapshot['snapshots'])
assert 5 == len(get_indices(self.client))
def test_no_repository(self):
self.create_indices(5)
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'],
testvars.snapshot_test.format(' ', 'snap_name', 1, 30),
)
self.invoke_runner()
assert 1 == self.result.exit_code
def test_extra_option(self):
self.create_indices(5)
self.write_config(self.args['configfile'], testvars.client_config.format(HOST))
self.write_config(
self.args['actionfile'], testvars.bad_option_proto_test.format('snapshot')
)
self.invoke_runner()
assert 1 == self.result.exit_code
class TestCLISnapshot(CuratorTestCase):
def test_snapshot(self):
self.create_indices(5)
self.create_repository()
snap_name = 'snapshot1'
args = self.get_runner_args()
args += [
'--config',
self.args['configfile'],
'snapshot',
'--repository',
self.args['repository'],
'--name',
snap_name,
'--wait_interval',
'1',
'--max_wait',
'30',
'--filter_list',
'{"filtertype":"none"}',
]
assert 0 == self.run_subprocess(args, logname='TestCLISnapshot.test_snapshot')
snapshot = get_snapshot(self.client, self.args['repository'], '*')
assert 1 == len(snapshot['snapshots'])
assert snap_name == snapshot['snapshots'][0]['snapshot']
|
