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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
|
# Copyright 2016 Intel Corp.
#
# 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.
from unittest import mock
from oslo_utils import versionutils
from ironic.api.controllers.v1 import versions as api_versions
from ironic.common import release_mappings
from ironic.conductor import rpcapi
from ironic.db.sqlalchemy import models
from ironic.objects import base as obj_base
from ironic.tests import base
def _check_versions_compatibility(conf_version, actual_version):
"""Checks the configured version against the actual version.
Returns True if the configured version is <= the actual version;
otherwise returns False.
:param conf_version: configured version, a string with dots
:param actual_version: actual version, a string with dots
:returns: True if the configured version is <= the actual version;
False otherwise.
"""
conf_cap = versionutils.convert_version_to_tuple(conf_version)
actual_cap = versionutils.convert_version_to_tuple(actual_version)
return conf_cap <= actual_cap
NUMERIC_RELEASES = sorted(
map(versionutils.convert_version_to_tuple,
set(release_mappings.RELEASE_MAPPING)
# Update the exceptions whenever needed
- {'master', 'zed', 'yoga'}),
reverse=True)
class ReleaseMappingsTestCase(base.TestCase):
"""Tests the dict release_mappings.RELEASE_MAPPING.
Tests whether the dict release_mappings.RELEASE_MAPPING is correct,
valid and consistent.
"""
def test_structure(self):
for value in release_mappings.RELEASE_MAPPING.values():
self.assertIsInstance(value, dict)
self.assertEqual({'api', 'rpc', 'objects'}, set(value))
self.assertIsInstance(value['api'], str)
(major, minor) = value['api'].split('.')
self.assertEqual(1, int(major))
self.assertLessEqual(int(minor), api_versions.MINOR_MAX_VERSION)
self.assertIsInstance(value['rpc'], str)
self.assertIsInstance(value['objects'], dict)
for obj_value in value['objects'].values():
self.assertIsInstance(obj_value, list)
for ver in obj_value:
self.assertIsInstance(ver, str)
tuple_ver = versionutils.convert_version_to_tuple(ver)
self.assertEqual(2, len(tuple_ver))
def test_object_names_are_registered(self):
registered_objects = set(obj_base.IronicObjectRegistry.obj_classes())
for mapping in release_mappings.RELEASE_MAPPING.values():
objects = set(mapping['objects'])
self.assertTrue(objects.issubset(registered_objects))
def test_current_rpc_version(self):
self.assertEqual(rpcapi.ConductorAPI.RPC_API_VERSION,
release_mappings.RELEASE_MAPPING['master']['rpc'])
def test_current_object_versions(self):
registered_objects = obj_base.IronicObjectRegistry.obj_classes()
obj_versions = release_mappings.get_object_versions(
releases=['master'])
for obj, vers in obj_versions.items():
# vers is a set of versions, not ordered
self.assertIn(registered_objects[obj][0].VERSION, vers)
def test_contains_all_db_objects(self):
self.assertIn('master', release_mappings.RELEASE_MAPPING)
use_models = models.Base.__subclasses__()
use_models.append(models.Node)
model_names = set((s.__name__ for s in use_models))
# NOTE(xek): As a rule, all models which can be changed between
# releases or are sent through RPC should have their counterpart
# versioned objects. Do not add an exception for such objects,
# initialize them with the version 1.0 instead.
# NodeBase is also excluded as it is covered by Node.
exceptions = set(['NodeTag', 'ConductorHardwareInterfaces',
'NodeTrait', 'DeployTemplateStep',
'NodeBase'])
model_names -= exceptions
# NodeTrait maps to two objects
model_names |= set(['Trait', 'TraitList'])
# Deployment is purely virtual.
model_names.add('Deployment')
object_names = set(
release_mappings.RELEASE_MAPPING['master']['objects'])
self.assertEqual(model_names, object_names)
def test_rpc_and_objects_versions_supported(self):
registered_objects = obj_base.IronicObjectRegistry.obj_classes()
for versions in release_mappings.RELEASE_MAPPING.values():
self.assertTrue(_check_versions_compatibility(
versions['rpc'], rpcapi.ConductorAPI.RPC_API_VERSION))
for obj_name, obj_vers in versions['objects'].items():
for ver in obj_vers:
self.assertTrue(_check_versions_compatibility(
ver, registered_objects[obj_name][0].VERSION))
def test_no_gaps_in_release_versions(self):
for i, ver in enumerate(NUMERIC_RELEASES[:-1]):
prev = NUMERIC_RELEASES[i + 1]
if ver != (prev[0] + 1, 0) and ver != (prev[0], prev[1] + 1):
self.fail("Versions %s and %s are not sequential"
% (prev, ver))
def test_no_gaps_in_object_versions(self):
oldest_release = '%d.%d' % NUMERIC_RELEASES[-1]
oldest_versions = release_mappings.RELEASE_MAPPING[
oldest_release]['objects']
all_versions = release_mappings.get_object_versions()
for obj, versions in all_versions.items():
try:
# NOTE(dtantsur): assuming all versions are 1.x, fix this test
# if this assumption ever changes.
min_version = min(map(versionutils.convert_version_to_tuple,
oldest_versions[obj]))[1]
except (KeyError, ValueError):
# If the object was introduced after the oldest known release,
# then 1.0 must be known.
expected = {"1.%d" % i for i in range(len(versions))}
else:
# The object was introduced before or in the oldest known
# release, start counting from this release.
expected = {"1.%d" % i
for i in range(min_version,
min_version + len(versions))}
self.assertEqual(expected, versions,
"There are gaps in versions for %s" % obj)
class GetObjectVersionsTestCase(base.TestCase):
TEST_MAPPING = {
'7.0': {
'api': '1.30',
'rpc': '1.40',
'objects': {
'Node': ['1.21'],
'Conductor': ['1.2'],
'Port': ['1.6'],
'Portgroup': ['1.3'],
}
},
'8.0': {
'api': '1.30',
'rpc': '1.40',
'objects': {
'Node': ['1.22'],
'Conductor': ['1.2'],
'Chassis': ['1.3'],
'Port': ['1.6'],
'Portgroup': ['1.5', '1.4'],
}
},
'master': {
'api': '1.34',
'rpc': '1.40',
'objects': {
'Node': ['1.23'],
'Conductor': ['1.2'],
'Chassis': ['1.3'],
'Port': ['1.7'],
'Portgroup': ['1.5'],
}
},
}
TEST_MAPPING['ocata'] = TEST_MAPPING['7.0']
def test_get_object_versions(self):
with mock.patch.dict(release_mappings.RELEASE_MAPPING,
self.TEST_MAPPING, clear=True):
actual_versions = release_mappings.get_object_versions()
expected_versions = {
'Node': set(['1.21', '1.22', '1.23']),
'Conductor': set(['1.2']),
'Chassis': set(['1.3']),
'Port': set(['1.6', '1.7']),
'Portgroup': set(['1.3', '1.4', '1.5']),
}
self.assertEqual(expected_versions, actual_versions)
def test_get_object_versions_releases(self):
with mock.patch.dict(release_mappings.RELEASE_MAPPING,
self.TEST_MAPPING, clear=True):
actual_versions = release_mappings.get_object_versions(
releases=['ocata'])
expected_versions = {
'Node': set(['1.21']),
'Conductor': set(['1.2']),
'Port': set(['1.6']),
'Portgroup': set(['1.3']),
}
self.assertEqual(expected_versions, actual_versions)
def test_get_object_versions_objects(self):
with mock.patch.dict(release_mappings.RELEASE_MAPPING,
self.TEST_MAPPING, clear=True):
actual_versions = release_mappings.get_object_versions(
objects=['Portgroup', 'Chassis'])
expected_versions = {
'Portgroup': set(['1.3', '1.4', '1.5']),
'Chassis': set(['1.3']),
}
self.assertEqual(expected_versions, actual_versions)
def test_get_object_versions_releases_objects(self):
with mock.patch.dict(release_mappings.RELEASE_MAPPING,
self.TEST_MAPPING, clear=True):
actual_versions = release_mappings.get_object_versions(
releases=['7.0'], objects=['Portgroup', 'Chassis'])
expected_versions = {
'Portgroup': set(['1.3']),
}
self.assertEqual(expected_versions, actual_versions)
|
