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
|
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
from unittest import mock
from heat.common import template_format
from heat.engine import scheduler
from heat.tests import common
from heat.tests import utils
sg_template = {
"heat_template_version": "2013-05-23",
"resources": {
"ServerGroup": {
"type": "OS::Nova::ServerGroup",
"properties": {
"name": "test",
"policies": ["anti-affinity"],
"rules": {
"max_server_per_host": 8}
}
}
}
}
class FakeGroup(object):
def __init__(self, name):
self.id = name
self.name = name
class NovaServerGroupTest(common.HeatTestCase):
def setUp(self):
super(NovaServerGroupTest, self).setUp()
def _init_template(self, sg_template):
template = template_format.parse(json.dumps(sg_template))
self.stack = utils.parse_stack(template)
self.sg = self.stack['ServerGroup']
# create mock clients and objects
nova = mock.MagicMock()
self.sg.client = mock.MagicMock(return_value=nova)
mock_plugin = mock.MagicMock()
self.patchobject(mock_plugin,
'is_version_supported',
return_value=True)
self.patchobject(self.sg, 'client_plugin',
return_value=mock_plugin)
self.sg_mgr = nova.server_groups
def _create_sg(self, name):
if name:
sg = sg_template['resources']['ServerGroup']
sg['properties']['name'] = name
self._init_template(sg_template)
self.sg_mgr.create.return_value = FakeGroup(name)
else:
try:
sg = sg_template['resources']['ServerGroup']
del sg['properties']['name']
except Exception:
pass
self._init_template(sg_template)
name = 'test'
n = name
def fake_create(name, policy, rules):
self.assertGreater(len(name), 1)
return FakeGroup(n)
self.sg_mgr.create = fake_create
scheduler.TaskRunner(self.sg.create)()
self.assertEqual((self.sg.CREATE, self.sg.COMPLETE),
self.sg.state)
self.assertEqual(name, self.sg.resource_id)
def test_sg_create(self):
self._create_sg('test')
expected_args = ()
expected_kwargs = {'name': 'test',
'policy': "anti-affinity",
'rules': {
'max_server_per_host': 8}
}
self.sg_mgr.create.assert_called_once_with(*expected_args,
**expected_kwargs)
def test_sg_create_no_name(self):
self._create_sg(None)
def test_sg_show_resource(self):
self._create_sg('test')
self.sg.client = mock.MagicMock()
s_groups = mock.MagicMock()
sg = mock.MagicMock()
sg.to_dict.return_value = {'server_gr': 'info'}
s_groups.get.return_value = sg
self.sg.client().server_groups = s_groups
self.assertEqual({'server_gr': 'info'}, self.sg.FnGetAtt('show'))
s_groups.get.assert_called_once_with('test')
|
