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
|
"""test_action_snapshot"""
# pylint: disable=missing-function-docstring, missing-class-docstring, line-too-long, protected-access, attribute-defined-outside-init
from unittest import TestCase
from unittest.mock import Mock
from curator.actions import Snapshot
from curator.exceptions import ActionError, CuratorException, FailedExecution, FailedSnapshot, MissingArgument, SnapshotInProgress
from curator import IndexList
# Get test variables and constants from a single source
from . import testvars
class TestActionSnapshot(TestCase):
VERSION = {'version': {'number': '8.0.0'} }
def builder(self):
self.client = Mock()
self.client.info.return_value = self.VERSION
self.client.cat.indices.return_value = testvars.state_one
self.client.indices.get_settings.return_value = testvars.settings_one
self.client.indices.stats.return_value = testvars.stats_one
self.client.indices.exists_alias.return_value = False
self.client.snapshot.get_repository.return_value = testvars.test_repo
self.client.snapshot.get.return_value = testvars.snapshots
self.client.tasks.get.return_value = testvars.no_snap_tasks
self.ilo = IndexList(self.client)
def test_init_raise_bad_index_list(self):
self.assertRaises(TypeError, Snapshot, 'invalid')
def test_init_no_repo_arg_exception(self):
self.builder()
self.assertRaises(MissingArgument, Snapshot, self.ilo)
def test_init_no_repo_exception(self):
self.builder()
self.client.snapshot.get_repository.return_value = {'repo':{'foo':'bar'}}
self.assertRaises(ActionError, Snapshot, self.ilo, repository='notfound')
def test_init_no_name_exception(self):
self.builder()
self.assertRaises(MissingArgument, Snapshot, self.ilo, repository=testvars.repo_name)
def test_init_success(self):
self.builder()
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertEqual(testvars.repo_name, sso.repository)
self.assertIsNone(sso.state)
def test_get_state_success(self):
self.builder()
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
sso.get_state()
self.assertEqual('SUCCESS', sso.state)
def test_get_state_fail(self):
self.builder()
self.client.snapshot.get.return_value = {'snapshots':[]}
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertRaises(CuratorException, sso.get_state)
def test_report_state_success(self):
self.builder()
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
sso.report_state()
self.assertEqual('SUCCESS', sso.state)
def test_report_state_other(self):
self.builder()
self.client.snapshot.get.return_value = testvars.highly_unlikely
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertRaises(FailedSnapshot, sso.report_state)
def test_do_dry_run(self):
self.builder()
self.client.snapshot.create.return_value = None
self.client.snapshot.status.return_value = testvars.nosnap_running
self.client.snapshot.verify_repository.return_value = testvars.verified_nodes
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertIsNone(sso.do_dry_run())
def test_do_action_success(self):
self.builder()
self.client.snapshot.create.return_value = testvars.generic_task
self.client.tasks.get.return_value = testvars.completed_task
self.client.snapshot.status.return_value = testvars.nosnap_running
self.client.snapshot.verify_repository.return_value = testvars.verified_nodes
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertIsNone(sso.do_action())
def test_do_action_raise_snap_in_progress(self):
self.builder()
self.client.snapshot.create.return_value = None
self.client.snapshot.status.return_value = testvars.snap_running
self.client.snapshot.verify_repository.return_value = testvars.verified_nodes
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertRaises(SnapshotInProgress, sso.do_action)
def test_do_action_no_wait_for_completion(self):
self.builder()
self.client.snapshot.create.return_value = testvars.generic_task
self.client.snapshot.status.return_value = testvars.nosnap_running
self.client.snapshot.verify_repository.return_value = testvars.verified_nodes
sso = Snapshot(self.ilo, repository=testvars.repo_name,
name=testvars.snap_name, wait_for_completion=False)
self.assertIsNone(sso.do_action())
def test_do_action_raise_on_failure(self):
self.builder()
self.client.snapshot.create.return_value = None
self.client.snapshot.create.side_effect = testvars.fake_fail
self.client.snapshot.status.return_value = testvars.nosnap_running
self.client.snapshot.verify_repository.return_value = testvars.verified_nodes
sso = Snapshot(self.ilo, repository=testvars.repo_name, name=testvars.snap_name)
self.assertRaises(FailedExecution, sso.do_action)
|
