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
|
from testtools import TestCase
from mock import Mock
from reddwarfclient import management
from reddwarfclient import base
"""
Unit tests for management.py
"""
class RootHistoryTest(TestCase):
def setUp(self):
super(RootHistoryTest, self).setUp()
self.orig__init = management.RootHistory.__init__
management.RootHistory.__init__ = Mock(return_value=None)
def tearDown(self):
super(RootHistoryTest, self).tearDown()
management.RootHistory.__init__ = self.orig__init
def test___repr__(self):
root_history = management.RootHistory()
root_history.id = "1"
root_history.created = "ct"
root_history.user = "tu"
self.assertEqual('<Root History: Instance 1 enabled at ct by tu>',
root_history.__repr__())
class ManagementTest(TestCase):
def setUp(self):
super(ManagementTest, self).setUp()
self.orig__init = management.Management.__init__
management.Management.__init__ = Mock(return_value=None)
self.management = management.Management()
self.management.api = Mock()
self.management.api.client = Mock()
self.orig_hist__init = management.RootHistory.__init__
self.orig_base_getid = base.getid
base.getid = Mock(return_value="instance1")
def tearDown(self):
super(ManagementTest, self).tearDown()
management.Management.__init__ = self.orig__init
management.RootHistory.__init__ = self.orig_hist__init
base.getid = self.orig_base_getid
def test__list(self):
self.management.api.client.get = Mock(return_value=('resp', None))
self.assertRaises(Exception, self.management._list, "url", None)
body = Mock()
body.get = Mock(return_value=[{'href': 'http://test.net/test_file',
'rel': 'next'}])
body.__getitem__ = Mock(return_value='instance1')
self.management.resource_class = Mock(return_value="instance-1")
self.management.api.client.get = Mock(return_value=('resp', body))
_expected = [{'href': 'http://test.net/test_file', 'rel': 'next'}]
self.assertEqual(_expected, self.management._list("url", None).links)
def test_show(self):
def side_effect_func(path, instance):
return path, instance
self.management._get = Mock(side_effect=side_effect_func)
p, i = self.management.show(1)
self.assertEqual(('/mgmt/instances/instance1', 'instance'), (p, i))
def test_index(self):
def side_effect_func(url, name, limit, marker):
return url
self.management._list = Mock(side_effect=side_effect_func)
self.assertEqual('/mgmt/instances?deleted=true',
self.management.index(deleted=True))
self.assertEqual('/mgmt/instances?deleted=false',
self.management.index(deleted=False))
def test_root_enabled_history(self):
self.management.api.client.get = Mock(return_value=('resp', None))
self.assertRaises(Exception,
self.management.root_enabled_history, "instance")
body = {'root_history': 'rh'}
self.management.api.client.get = Mock(return_value=('resp', body))
management.RootHistory.__init__ = Mock(return_value=None)
rh = self.management.root_enabled_history("instance")
self.assertTrue(isinstance(rh, management.RootHistory))
def test__action(self):
resp = Mock()
self.management.api.client.post = Mock(return_value=(resp, 'body'))
resp.status = 200
self.management._action(1, 'body')
self.assertEqual(1, self.management.api.client.post.call_count)
resp.status = 400
self.assertRaises(ValueError, self.management._action, 1, 'body')
self.assertEqual(2, self.management.api.client.post.call_count)
def _mock_action(self):
self.body_ = ""
def side_effect_func(instance_id, body):
self.body_ = body
self.management._action = Mock(side_effect=side_effect_func)
def test_stop(self):
self._mock_action()
self.management.stop(1)
self.assertEqual(1, self.management._action.call_count)
self.assertEqual({'stop': {}}, self.body_)
def test_reboot(self):
self._mock_action()
self.management.reboot(1)
self.assertEqual(1, self.management._action.call_count)
self.assertEqual({'reboot': {}}, self.body_)
def test_migrate(self):
self._mock_action()
self.management.migrate(1)
self.assertEqual(1, self.management._action.call_count)
self.assertEqual({'migrate': {}}, self.body_)
def test_update(self):
self._mock_action()
self.management.update(1)
self.assertEqual(1, self.management._action.call_count)
self.assertEqual({'update': {}}, self.body_)
def test_reset_task_status(self):
self._mock_action()
self.management.reset_task_status(1)
self.assertEqual(1, self.management._action.call_count)
self.assertEqual({'reset-task-status': {}}, self.body_)
|
