Add DB API for Firmware and Object
Adds the following methods to DB API: * create_firmware_component * update_firmware_component * get_firmware_component * get_firmware_component_list FirmwareComponent * create | save | get FirmwareComponentList * get_by_node id | sync_firmware_components Adds two exceptions: * FirmwareComponentAlreadyExists * FirmwareComponentNotFound Tests for db and objects Changes were required in models, the class name should match the object name we will create Story: 2010659 Task: 47977 Change-Id: Ie1e2a4150d4ee4521290737612780c02506f4a9e
This commit is contained in:
Iury Gregory Melo Ferreira
parent
76a920aed2
commit
dad6724292
@ -874,3 +874,12 @@ class ConcurrentActionLimit(IronicException):
|
||||
class SwiftObjectStillExists(IronicException):
|
||||
_msg_fmt = _("Clean up failed for swift object %(obj)s during deletion"
|
||||
" of node %(node)s.")
|
||||
|
||||
|
||||
class FirmwareComponentAlreadyExists(Conflict):
|
||||
_msg_fmt = _('A Firmware component %(name)s for node %(node)s'
|
||||
' already exists.')
|
||||
|
||||
|
||||
class FirmwareComponentNotFound(NotFound):
|
||||
_msg_fmt = _("Node %(node)s doesn't have Firmware component %(name)s")
|
||||
|
||||
@ -592,6 +592,7 @@ RELEASE_MAPPING = {
|
||||
'TraitList': ['1.0'],
|
||||
'VolumeConnector': ['1.0'],
|
||||
'VolumeTarget': ['1.0'],
|
||||
'FirmwareComponent': ['1.0'],
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
@ -1454,3 +1454,64 @@ class Connection(object, metaclass=abc.ABCMeta):
|
||||
|
||||
:returns: list of dicts containing shard names and count
|
||||
"""
|
||||
|
||||
@abc.abstractclassmethod
|
||||
def create_firmware_component(self, values):
|
||||
"""Create a FirmwareComponent record for a given node.
|
||||
|
||||
:param values: a dictionary with the necessary information to create
|
||||
a FirmwareComponent.
|
||||
|
||||
::
|
||||
|
||||
{
|
||||
'component': String,
|
||||
'initial_version': String,
|
||||
'current_version': String,
|
||||
'last_version_flashed': String
|
||||
}
|
||||
:returns: A FirmwareComponent object.
|
||||
:raises: FirmwareComponentAlreadyExists if any of the component
|
||||
records already exists.
|
||||
"""
|
||||
|
||||
@abc.abstractclassmethod
|
||||
def update_firmware_component(self, node_id, component, values):
|
||||
"""Update a FirmwareComponent record.
|
||||
|
||||
:param node_id: The node id.
|
||||
:param component: The component of the node to update.
|
||||
:param values: A dictionary with the new information about the
|
||||
FirmwareComponent.
|
||||
|
||||
::
|
||||
|
||||
{
|
||||
'current_version': String,
|
||||
'last_version_flashed': String
|
||||
}
|
||||
:returns: A FirmwareComponent object.
|
||||
:raises: FirmwareComponentNotFound the component
|
||||
is not found.
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def get_firmware_component(self, node_id, name):
|
||||
"""Retrieve Firmware Component.
|
||||
|
||||
:param node_id: The node id.
|
||||
:param name: name of Firmware component.
|
||||
:returns: The FirmwareComponent object.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
:raises: FirmwareComponentNotFound if the Firmware component
|
||||
is not found.
|
||||
"""
|
||||
|
||||
@abc.abstractclassmethod
|
||||
def get_firmware_component_list(self, node_id):
|
||||
"""Retrieve a list Firmware Components of a given node.
|
||||
|
||||
:param node_id: The node id.
|
||||
:returns: A list of FirmwareComponent objects.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
"""
|
||||
|
||||
@ -892,6 +892,11 @@ class Connection(api.Connection):
|
||||
models.NodeInventory).filter_by(node_id=node_id)
|
||||
inventory_query.delete()
|
||||
|
||||
# delete all firmware components attached to the node
|
||||
firmware_component_query = session.query(
|
||||
models.FirmwareComponent).filter_by(node_id=node_id)
|
||||
firmware_component_query.delete()
|
||||
|
||||
query.delete()
|
||||
|
||||
def update_node(self, node_id, values):
|
||||
@ -2721,3 +2726,103 @@ class Connection(api.Connection):
|
||||
)
|
||||
|
||||
return shard_list
|
||||
|
||||
def create_firmware_component(self, values):
|
||||
"""Create a FirmwareComponent record for a given node.
|
||||
|
||||
:param values: a dictionary with the necessary information to create
|
||||
a FirmwareComponent.
|
||||
|
||||
::
|
||||
|
||||
{
|
||||
'component': String,
|
||||
'initial_version': String,
|
||||
'current_version': String,
|
||||
'last_version_flashed': String
|
||||
}
|
||||
:returns: A FirmwareComponent object.
|
||||
:raises: FirmwareComponentAlreadyExists if any of the component
|
||||
records already exists.
|
||||
"""
|
||||
|
||||
fw_component = models.FirmwareComponent()
|
||||
fw_component.update(values)
|
||||
|
||||
try:
|
||||
with _session_for_write() as session:
|
||||
session.add(fw_component)
|
||||
session.flush()
|
||||
except db_exc.DBDuplicateEntry:
|
||||
raise exception.FirmwareComponentAlreadyExists(
|
||||
name=values['component'], node=values['node_id'])
|
||||
return fw_component
|
||||
|
||||
def update_firmware_component(self, node_id, component, values):
|
||||
"""Update a FirmwareComponent record.
|
||||
|
||||
:param node_id: The node id.
|
||||
:param component: The component of the node to update.
|
||||
:param values: A dictionary with the new information about the
|
||||
FirmwareComponent.
|
||||
|
||||
::
|
||||
|
||||
{
|
||||
'current_version': String,
|
||||
'last_version_flashed': String
|
||||
}
|
||||
:returns: A FirmwareComponent object.
|
||||
:raises: FirmwareComponentNotFound the component
|
||||
is not found.
|
||||
"""
|
||||
with _session_for_write() as session:
|
||||
query = session.query(models.FirmwareComponent)
|
||||
query = query.filter_by(node_id=node_id, component=component)
|
||||
|
||||
try:
|
||||
ref = query.with_for_update().one()
|
||||
except NoResultFound:
|
||||
raise exception.FirmwareComponentNotFound(
|
||||
node=node_id, name=component)
|
||||
|
||||
ref.update(values)
|
||||
session.flush()
|
||||
return ref
|
||||
|
||||
def get_firmware_component(self, node_id, name):
|
||||
"""Retrieve Firmware Component.
|
||||
|
||||
:param node_id: The node id.
|
||||
:param name: name of Firmware component.
|
||||
:returns: The FirmwareComponent object.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
:raises: FirmwareComponentNotFound if the FirmwareComponent
|
||||
is not found.
|
||||
"""
|
||||
|
||||
with _session_for_read() as session:
|
||||
self._check_node_exists(session, node_id)
|
||||
query = session.query(models.FirmwareComponent).filter_by(
|
||||
node_id=node_id, component=name)
|
||||
try:
|
||||
ref = query.one()
|
||||
except NoResultFound:
|
||||
raise exception.FirmwareComponentNotFound(
|
||||
node=node_id, name=name)
|
||||
return ref
|
||||
|
||||
def get_firmware_component_list(self, node_id):
|
||||
"""Retrieve Firmware Components of a given node.
|
||||
|
||||
:param node_id: The node id.
|
||||
:returns: A list of FirmwareComponent objects.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
"""
|
||||
|
||||
with _session_for_read() as session:
|
||||
self._check_node_exists(session, node_id)
|
||||
result = (session.query(models.FirmwareComponent)
|
||||
.filter_by(node_id=node_id)
|
||||
.all())
|
||||
return result
|
||||
|
||||
@ -499,7 +499,7 @@ class NodeInventory(Base):
|
||||
node_id = Column(Integer, ForeignKey('nodes.id'), nullable=True)
|
||||
|
||||
|
||||
class FirmwareInformation(Base):
|
||||
class FirmwareComponent(Base):
|
||||
"""Represents the firmware information of a bare metal node."""
|
||||
__tablename__ = "firmware_information"
|
||||
__table_args__ = (
|
||||
|
||||
@ -30,6 +30,7 @@ def register_all():
|
||||
__import__('ironic.objects.conductor')
|
||||
__import__('ironic.objects.deploy_template')
|
||||
__import__('ironic.objects.deployment')
|
||||
__import__('ironic.objects.firmware')
|
||||
__import__('ironic.objects.node')
|
||||
__import__('ironic.objects.node_history')
|
||||
__import__('ironic.objects.node_inventory')
|
||||
|
||||
161
ironic/objects/firmware.py
Normal file
161
ironic/objects/firmware.py
Normal file
@ -0,0 +1,161 @@
|
||||
# coding=utf-8
|
||||
#
|
||||
#
|
||||
# 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 oslo_versionedobjects import base as object_base
|
||||
|
||||
from ironic.db import api as dbapi
|
||||
from ironic.objects import base
|
||||
from ironic.objects import fields as object_fields
|
||||
|
||||
|
||||
@base.IronicObjectRegistry.register
|
||||
class FirmwareComponent(base.IronicObject):
|
||||
# Version 1.0: Initial version
|
||||
VERSION = '1.0'
|
||||
|
||||
dbapi = dbapi.get_instance()
|
||||
|
||||
fields = {
|
||||
'id': object_fields.IntegerField(),
|
||||
'node_id': object_fields.IntegerField(nullable=False),
|
||||
'component': object_fields.StringField(nullable=False),
|
||||
'initial_version': object_fields.StringField(nullable=False),
|
||||
'current_version': object_fields.StringField(nullable=True),
|
||||
'last_version_flashed': object_fields.StringField(nullable=True),
|
||||
}
|
||||
|
||||
def create(self, context=None):
|
||||
"""Create a Firmware record in the DB.
|
||||
|
||||
:param context: Security context.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
:raises: FirmwareComponentAlreadyExists if the record already exists.
|
||||
"""
|
||||
values = self.do_version_changes_for_db()
|
||||
# Note(iurygregory): We ensure that when creating we will be setting
|
||||
# initial_version to the current_version we got from the BMC.
|
||||
values['initial_version'] = values['current_version']
|
||||
db_fwcmp = self.dbapi.create_firmware_component(values)
|
||||
self._from_db_object(self._context, self, db_fwcmp)
|
||||
|
||||
def save(self, context=None):
|
||||
"""Save updates to this Firmware Component.
|
||||
|
||||
Updates will be made column by column based on the result of
|
||||
self.what_changed()
|
||||
|
||||
:param context: Security context. NOTE: This should only
|
||||
be used internally by the indirection_api.
|
||||
Unfortunately, RPC requires context as the first
|
||||
argument, even though we don't use it.
|
||||
A context should be set when instantiating the
|
||||
object, e.g.: FirmwareComponent(context)
|
||||
:raises: NodeNotFound if the node id is not found.
|
||||
:raises: FirmwareComponentNotFound if the component is not found.
|
||||
"""
|
||||
|
||||
# NOTE(iurygregory): some fields shouldn't be updated, like
|
||||
# 'initial_version', 'id', 'node_id', 'component'
|
||||
# filter them out or raise an Error?
|
||||
updates = self.do_version_changes_for_db()
|
||||
up_fwcmp = self.dbapi.update_firmware_component(
|
||||
self.node_id, self.component, updates)
|
||||
self._from_db_object(self._context, self, up_fwcmp)
|
||||
|
||||
@classmethod
|
||||
def get(cls, context, node_id, name):
|
||||
"""Get a FirmwareComponent based on its node_id and name.
|
||||
|
||||
:param context: Security context.
|
||||
:param node_id: The node id.
|
||||
:param name: The Firmware Component name.
|
||||
:raises: NodeNotFound if the node id is not found.
|
||||
:raises: FirmwareComponentNotFound if the Firmware Component
|
||||
name is not found.
|
||||
:returns: A :class:'FirmwareComponent' object.
|
||||
"""
|
||||
db_fw_cmp = cls.dbapi.get_firmware_component(node_id, name)
|
||||
fw_cmp = cls._from_db_object(context, cls(), db_fw_cmp)
|
||||
return fw_cmp
|
||||
|
||||
|
||||
@base.IronicObjectRegistry.register
|
||||
class FirmwareComponentList(base.IronicObjectListBase, base.IronicObject):
|
||||
# Version 1.0: Initial version
|
||||
VERSION = '1.0'
|
||||
|
||||
dbapi = dbapi.get_instance()
|
||||
|
||||
fields = {
|
||||
'objects': object_fields.ListOfObjectsField('FirmwareComponent'),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_by_node_id(cls, context, node_id):
|
||||
"""Get FirmwareComponent based on node_id.
|
||||
|
||||
:param context: Security context.
|
||||
:param node_id: The node id.
|
||||
:raises: NodeNotFound if the node is not found.
|
||||
:return: A list of FirmwareComponent objects.
|
||||
"""
|
||||
node_fw_components = cls.dbapi.get_firmware_component_list(node_id)
|
||||
return object_base.obj_make_list(
|
||||
context, cls(), FirmwareComponent, node_fw_components)
|
||||
|
||||
@classmethod
|
||||
def sync_firmware_components(cls, context, node_id, components):
|
||||
"""Returns a list of create/update components.
|
||||
|
||||
This method sync with the 'firmware_information' database table
|
||||
and sorts three lists - create / update / unchanged components.
|
||||
|
||||
:param context: Security context.
|
||||
:param node_id: The node id.
|
||||
:param components: List of FirmwareComponents.
|
||||
:returns: A 3-tuple of lists of Firmware Components to be created,
|
||||
updated and unchanged.
|
||||
"""
|
||||
create_list = []
|
||||
update_list = []
|
||||
unchanged_list = []
|
||||
current_components_dict = {}
|
||||
|
||||
current_components = cls.get_by_node_id(context, node_id)
|
||||
|
||||
for cmp in current_components:
|
||||
current_components_dict[cmp.component] = {
|
||||
'initial_version': cmp.initial_version,
|
||||
'current_version': cmp.current_version,
|
||||
'last_version_flashed': cmp.last_version_flashed,
|
||||
}
|
||||
|
||||
for cmp in components:
|
||||
if cmp['component'] in current_components_dict:
|
||||
values = current_components_dict[cmp['component']]
|
||||
|
||||
cv_changed = cmp['current_version'] \
|
||||
!= values.get('current_version')
|
||||
lvf_changed = cmp['last_version_flashed'] \
|
||||
!= values.get('last_version_flashed')
|
||||
|
||||
if cv_changed or lvf_changed:
|
||||
update_list.append(cmp)
|
||||
else:
|
||||
unchanged_list.append(cmp)
|
||||
else:
|
||||
create_list.append(cmp)
|
||||
|
||||
return (create_list, update_list, unchanged_list)
|
||||
@ -101,7 +101,7 @@ class ReleaseMappingsTestCase(base.TestCase):
|
||||
# NodeBase is also excluded as it is covered by Node.
|
||||
exceptions = set(['NodeTag', 'ConductorHardwareInterfaces',
|
||||
'NodeTrait', 'DeployTemplateStep',
|
||||
'NodeBase', 'FirmwareInformation'])
|
||||
'NodeBase'])
|
||||
model_names -= exceptions
|
||||
# NodeTrait maps to two objects
|
||||
model_names |= set(['Trait', 'TraitList'])
|
||||
|
||||
@ -1323,17 +1323,17 @@ class MigrationCheckersMixin(object):
|
||||
insert_fw_cmp = fw_information.insert().values(fw_data)
|
||||
connection.execute(insert_fw_cmp)
|
||||
fw_cmp_stmt = sqlalchemy.select(
|
||||
models.FirmwareInformation.initial_version
|
||||
models.FirmwareComponent.initial_version
|
||||
).where(
|
||||
models.FirmwareInformation.node_id == data['id'],
|
||||
models.FirmwareInformation.component == fw_data['component']
|
||||
models.FirmwareComponent.node_id == data['id'],
|
||||
models.FirmwareComponent.component == fw_data['component']
|
||||
)
|
||||
fw_component = connection.execute(fw_cmp_stmt).first()
|
||||
self.assertEqual('v1.0.0', fw_component['initial_version'])
|
||||
del_stmt = (
|
||||
sqlalchemy.delete(
|
||||
models.FirmwareInformation
|
||||
).where(models.FirmwareInformation.node_id == data['id'])
|
||||
models.FirmwareComponent
|
||||
).where(models.FirmwareComponent.node_id == data['id'])
|
||||
)
|
||||
connection.execute(del_stmt)
|
||||
|
||||
|
||||
115
ironic/tests/unit/db/test_firmware_component.py
Normal file
115
ironic/tests/unit/db/test_firmware_component.py
Normal file
@ -0,0 +1,115 @@
|
||||
# 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.
|
||||
|
||||
"""Tests for manipulating FirmwareComponent via the DB API"""
|
||||
|
||||
from ironic.common import exception
|
||||
from ironic.tests.unit.db import base
|
||||
from ironic.tests.unit.db import utils as db_utils
|
||||
|
||||
|
||||
class DbFirmwareComponentTestCase(base.DbTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(DbFirmwareComponentTestCase, self).setUp()
|
||||
self.node = db_utils.create_test_node()
|
||||
|
||||
def test_create_firmware_component(self):
|
||||
values = {'node_id': self.node.id}
|
||||
fw_cmp = db_utils.create_test_firmware_component(**values)
|
||||
self.assertCountEqual('bmc', fw_cmp.component)
|
||||
self.assertCountEqual('v1.0.0', fw_cmp.initial_version)
|
||||
|
||||
def test_create_firmware_component_duplicate(self):
|
||||
component = db_utils.get_test_firmware_component_list()[0]
|
||||
component['node_id'] = self.node.id
|
||||
self.dbapi.create_firmware_component(component)
|
||||
self.assertRaises(exception.FirmwareComponentAlreadyExists,
|
||||
self.dbapi.create_firmware_component,
|
||||
component)
|
||||
|
||||
def test_get_firmware_component(self):
|
||||
values = {'node_id': self.node.id}
|
||||
db_utils.create_test_firmware_component(**values)
|
||||
result = self.dbapi.get_firmware_component(self.node.id, 'bmc')
|
||||
self.assertEqual(result.node_id, self.node.id)
|
||||
self.assertEqual(result.component, 'bmc')
|
||||
self.assertEqual(result.initial_version, 'v1.0.0')
|
||||
|
||||
def test_get_firmware_component_node_not_exist(self):
|
||||
self.assertRaises(exception.NodeNotFound,
|
||||
self.dbapi.get_firmware_component,
|
||||
456, 'BIOS')
|
||||
|
||||
def test_get_firmware_component_setting_not_exist(self):
|
||||
db_utils.create_test_firmware_component(node_id=self.node.id)
|
||||
self.assertRaises(exception.FirmwareComponentNotFound,
|
||||
self.dbapi.get_firmware_component,
|
||||
self.node.id, 'bios_name')
|
||||
|
||||
def test_get_firmware_component_list(self):
|
||||
db_utils.create_test_firmware_component(node_id=self.node.id)
|
||||
result = self.dbapi.get_firmware_component_list(
|
||||
node_id=self.node.id)
|
||||
self.assertEqual(result[0]['node_id'], self.node.id)
|
||||
self.assertEqual(result[0]['component'], 'bmc')
|
||||
self.assertEqual(result[0]['initial_version'], 'v1.0.0')
|
||||
self.assertEqual(result[0]['version'], '1.0')
|
||||
self.assertEqual(len(result), 1)
|
||||
|
||||
def test_get_firmware_component_list_node_not_exist(self):
|
||||
self.assertRaises(exception.NodeNotFound,
|
||||
self.dbapi.get_firmware_component_list,
|
||||
4040)
|
||||
|
||||
def test_update_firmware_components(self):
|
||||
components = db_utils.get_test_firmware_component_list()
|
||||
components[0].update({'node_id': self.node.id})
|
||||
components[1].update({'node_id': self.node.id})
|
||||
self.dbapi.create_firmware_component(components[0])
|
||||
self.dbapi.create_firmware_component(components[1])
|
||||
update_components = [
|
||||
{'component': 'bmc', 'initial_version': 'v1.0.0',
|
||||
'current_version': 'v1.3.0', 'last_version_flashed': 'v1.3.0'},
|
||||
{'component': 'BIOS', 'initial_version': 'v1.5.0',
|
||||
'current_version': 'v1.5.5', 'last_version_flashed': 'v1.5.5'}
|
||||
]
|
||||
bmc_values = update_components[0]
|
||||
bios_values = update_components[1]
|
||||
bmc = self.dbapi.update_firmware_component(
|
||||
self.node.id, bmc_values['component'], bmc_values)
|
||||
bios = self.dbapi.update_firmware_component(
|
||||
self.node.id, bios_values['component'], bios_values)
|
||||
self.assertCountEqual('v1.3.0', bmc.current_version)
|
||||
self.assertCountEqual('v1.5.5', bios.current_version)
|
||||
|
||||
def test_update_firmware_component_not_exist(self):
|
||||
values = db_utils.get_test_firmware_component_list()[0]
|
||||
values['node_id'] = self.node.id
|
||||
db_utils.create_test_firmware_component(**values)
|
||||
values['component'] = 'nic'
|
||||
self.assertRaises(exception.FirmwareComponentNotFound,
|
||||
self.dbapi.update_firmware_component,
|
||||
self.node.id, 'nic', values)
|
||||
|
||||
def test_delete_firmware_component_list(self):
|
||||
values = db_utils.get_test_firmware_component_list()[0]
|
||||
values['node_id'] = self.node.id
|
||||
self.assertRaises(exception.FirmwareComponentNotFound,
|
||||
self.dbapi.get_firmware_component,
|
||||
self.node.id, 'bmc')
|
||||
self.dbapi.create_firmware_component(values)
|
||||
|
||||
self.dbapi.destroy_node(self.node.id)
|
||||
self.assertRaises(exception.NodeNotFound,
|
||||
self.dbapi.get_firmware_component,
|
||||
self.node.id, 'bmc')
|
||||
@ -798,6 +798,15 @@ class DbNodeTestCase(base.DbTestCase):
|
||||
self.assertRaises(exception.NodeInventoryNotFound,
|
||||
self.dbapi.get_node_inventory_by_node_id, node.id)
|
||||
|
||||
def test_firmware_component_list_after_destroying_a_node_by_uuid(self):
|
||||
node = utils.create_test_node()
|
||||
|
||||
utils.create_test_firmware_component(node_id=node.id)
|
||||
|
||||
self.dbapi.destroy_node(node.uuid)
|
||||
self.assertRaises(exception.NodeNotFound,
|
||||
self.dbapi.get_firmware_component_list, node.id)
|
||||
|
||||
def test_update_node(self):
|
||||
node = utils.create_test_node()
|
||||
|
||||
|
||||
@ -26,6 +26,7 @@ from ironic.objects import bios
|
||||
from ironic.objects import chassis
|
||||
from ironic.objects import conductor
|
||||
from ironic.objects import deploy_template
|
||||
from ironic.objects import firmware
|
||||
from ironic.objects import node
|
||||
from ironic.objects import node_history
|
||||
from ironic.objects import node_inventory
|
||||
@ -752,3 +753,43 @@ def create_test_inventory(**kw):
|
||||
del inventory['id']
|
||||
dbapi = db_api.get_instance()
|
||||
return dbapi.create_node_inventory(inventory)
|
||||
|
||||
|
||||
def create_test_firmware_component(**kw):
|
||||
"""Create test Firmware Component entry in DB and return object.
|
||||
|
||||
Function to be used to create test FirmwareComponent object in the DB.
|
||||
|
||||
:param kw: kwargs with overriding values for firmware component
|
||||
:returns: Test FirmwareComponent DB object
|
||||
"""
|
||||
fw_cmp_values = get_test_firmware_component(**kw)
|
||||
if 'id' not in kw:
|
||||
del fw_cmp_values['id']
|
||||
dbapi = db_api.get_instance()
|
||||
return dbapi.create_firmware_component(fw_cmp_values)
|
||||
|
||||
|
||||
def get_test_firmware_component(**kw):
|
||||
# NOTE(iurygregory): update version to get from the object in the object
|
||||
# patch.
|
||||
return {
|
||||
'id': kw.get('id', 256),
|
||||
'node_id': kw.get('node_id', 123),
|
||||
'component': kw.get('component', 'bmc'),
|
||||
'initial_version': kw.get('initial_version', 'v1.0.0'),
|
||||
'current_version': kw.get('current_version', 'v1.0.0'),
|
||||
'last_version_flashed': kw.get('last_version_flashed', None),
|
||||
'version': kw.get('version', firmware.FirmwareComponent.VERSION),
|
||||
'created_at': kw.get('created_at'),
|
||||
'updated_at': kw.get('updated_at'),
|
||||
}
|
||||
|
||||
|
||||
def get_test_firmware_component_list():
|
||||
return [
|
||||
{'component': 'bmc', 'initial_version': 'v1.0.0',
|
||||
'current_version': 'v1.0.0', 'last_version_flashed': None},
|
||||
{'component': 'BIOS', 'initial_version': 'v1.5.0',
|
||||
'current_version': 'v1.5.0', 'last_version_flashed': None},
|
||||
]
|
||||
|
||||
166
ironic/tests/unit/objects/test_firmware.py
Normal file
166
ironic/tests/unit/objects/test_firmware.py
Normal file
@ -0,0 +1,166 @@
|
||||
# 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 datetime
|
||||
import types
|
||||
from unittest import mock
|
||||
|
||||
|
||||
from ironic.common import exception
|
||||
from ironic import objects
|
||||
from ironic.tests.unit.db import base as db_base
|
||||
from ironic.tests.unit.db import utils as db_utils
|
||||
from ironic.tests.unit.objects import utils as obj_utils
|
||||
|
||||
|
||||
class TestFirmwareComponentObject(db_base.DbTestCase,
|
||||
obj_utils.SchemasTestMixIn):
|
||||
|
||||
def setUp(self):
|
||||
super(TestFirmwareComponentObject, self).setUp()
|
||||
self.firmware_component_dict = db_utils.get_test_firmware_component()
|
||||
|
||||
def test_get_firmware_component(self):
|
||||
with mock.patch.object(self.dbapi, 'get_firmware_component',
|
||||
autospec=True) as mock_get_component:
|
||||
node_id = self.firmware_component_dict['node_id']
|
||||
name = self.firmware_component_dict['component']
|
||||
mock_get_component.return_value = self.firmware_component_dict
|
||||
|
||||
fw_component = objects.FirmwareComponent.get(
|
||||
self.context, node_id, name)
|
||||
|
||||
mock_get_component.assert_called_once_with(node_id, name)
|
||||
|
||||
self.assertEqual(self.context, fw_component._context)
|
||||
self.assertEqual(self.firmware_component_dict['node_id'],
|
||||
fw_component.node_id)
|
||||
self.assertEqual(self.firmware_component_dict['component'],
|
||||
fw_component.component)
|
||||
self.assertEqual(self.firmware_component_dict['initial_version'],
|
||||
fw_component.initial_version)
|
||||
self.assertEqual(self.firmware_component_dict['current_version'],
|
||||
fw_component.current_version)
|
||||
self.assertEqual(
|
||||
self.firmware_component_dict['last_version_flashed'],
|
||||
fw_component.last_version_flashed)
|
||||
|
||||
def test_get_firmware_component_does_not_exist(self):
|
||||
fw_cmp_name = "does not exist"
|
||||
with mock.patch.object(self.dbapi, 'get_firmware_component',
|
||||
autospec=True) as mock_get_component:
|
||||
not_found = exception.FirmwareComponentNotFound(
|
||||
node=self.firmware_component_dict['node_id'],
|
||||
name=fw_cmp_name)
|
||||
mock_get_component.side_effect = not_found
|
||||
|
||||
self.assertRaises(exception.FirmwareComponentNotFound,
|
||||
objects.FirmwareComponent.get, self.context,
|
||||
self.firmware_component_dict['node_id'],
|
||||
fw_cmp_name)
|
||||
|
||||
def test_get_firmware_component_node_does_not_exist(self):
|
||||
with mock.patch.object(self.dbapi, 'get_firmware_component',
|
||||
autospec=True) as mock_get_component:
|
||||
mock_get_component.side_effect = exception.NodeNotFound(node=404)
|
||||
|
||||
self.assertRaises(exception.NodeNotFound,
|
||||
objects.FirmwareComponent.get, self.context,
|
||||
self.firmware_component_dict['node_id'],
|
||||
self.firmware_component_dict['component'])
|
||||
|
||||
def test_create(self):
|
||||
with mock.patch.object(self.dbapi, 'create_firmware_component',
|
||||
autospec=True) as mock_db_create:
|
||||
mock_db_create.return_value = self.firmware_component_dict
|
||||
new_fw_component = objects.FirmwareComponent(
|
||||
self.context, **self.firmware_component_dict)
|
||||
new_fw_component.create()
|
||||
|
||||
mock_db_create.assert_called_once_with(
|
||||
self.firmware_component_dict)
|
||||
|
||||
def test_save(self):
|
||||
node_id = self.firmware_component_dict['node_id']
|
||||
name = self.firmware_component_dict['component']
|
||||
test_time = datetime.datetime(2000, 1, 1, 0, 0)
|
||||
|
||||
with mock.patch.object(self.dbapi, 'get_firmware_component',
|
||||
autospec=True) as mock_get_fw:
|
||||
mock_get_fw.return_value = self.firmware_component_dict
|
||||
with mock.patch.object(self.dbapi, 'update_firmware_component',
|
||||
autospec=True) as mock_update_fw:
|
||||
|
||||
mock_update_fw.return_value = (
|
||||
db_utils.get_test_firmware_component(
|
||||
current_version="v2.0.1",
|
||||
last_version_flashed="v2.0.1",
|
||||
updated_at=test_time)
|
||||
)
|
||||
|
||||
fw_cmp = objects.FirmwareComponent.get(
|
||||
self.context, node_id, name)
|
||||
fw_cmp.current_version = "v2.0.1"
|
||||
fw_cmp.last_version_flashed = "v2.0.1"
|
||||
fw_cmp.save()
|
||||
|
||||
mock_get_fw.assert_called_once_with(node_id, name)
|
||||
|
||||
mock_update_fw.assert_called_once_with(
|
||||
node_id, fw_cmp.component,
|
||||
{'current_version': 'v2.0.1',
|
||||
'last_version_flashed': 'v2.0.1',
|
||||
'version': objects.FirmwareComponent.VERSION})
|
||||
|
||||
self.assertEqual(self.context, fw_cmp._context)
|
||||
self.assertEqual("v2.0.1", fw_cmp.current_version)
|
||||
self.assertEqual("v2.0.1", fw_cmp.last_version_flashed)
|
||||
res_updated_at = (fw_cmp.updated_at).replace(tzinfo=None)
|
||||
self.assertEqual(test_time, res_updated_at)
|
||||
|
||||
@mock.patch(
|
||||
'ironic.objects.firmware.FirmwareComponentList.get_by_node_id',
|
||||
spec_set=types.FunctionType)
|
||||
def test_sync_firmware_components_create_and_update(self, mock_get):
|
||||
node = obj_utils.create_test_node(self.context)
|
||||
fw_obj = obj_utils.create_test_firmware_component(
|
||||
self.context, node_id=node.id)
|
||||
mock_get.return_value = [fw_obj]
|
||||
components = db_utils.get_test_firmware_component_list()
|
||||
components[0]['current_version'] = 'v2.0.0'
|
||||
components[0]['last_version_flashed'] = 'v2.0.0'
|
||||
create, update, unchanged = (
|
||||
objects.FirmwareComponentList.sync_firmware_components(
|
||||
self.context, node.id, components))
|
||||
self.assertEqual(create, components[1:])
|
||||
self.assertEqual(update, [components[0]])
|
||||
self.assertEqual(unchanged, [])
|
||||
|
||||
@mock.patch(
|
||||
'ironic.objects.firmware.FirmwareComponentList.get_by_node_id',
|
||||
spec_set=types.FunctionType)
|
||||
def test_sync_firmware_components_nochange(self, mock_get):
|
||||
node = obj_utils.create_test_node(self.context)
|
||||
fw_obj_1 = obj_utils.create_test_firmware_component(
|
||||
self.context, node_id=node.id)
|
||||
fw_obj_2 = obj_utils.create_test_firmware_component(
|
||||
self.context, node_id=node.id, component='BIOS',
|
||||
initial_version='v1.5.0', current_version='v1.5.0')
|
||||
mock_get.return_value = [fw_obj_1, fw_obj_2]
|
||||
components = db_utils.get_test_firmware_component_list()
|
||||
|
||||
create, update, unchanged = (
|
||||
objects.FirmwareComponentList.sync_firmware_components(
|
||||
self.context, node.id, components))
|
||||
self.assertEqual(create, [])
|
||||
self.assertEqual(update, [])
|
||||
self.assertEqual(unchanged, components)
|
||||
@ -35,7 +35,7 @@ class TestNodeObject(db_base.DbTestCase, obj_utils.SchemasTestMixIn):
|
||||
super(TestNodeObject, self).setUp()
|
||||
self.ctxt = context.get_admin_context()
|
||||
self.fake_node = db_utils.get_test_node()
|
||||
self.node = obj_utils.get_test_node(self.ctxt, **self.fake_node)
|
||||
self.node = obj_utils.get_test_node(self.context, **self.fake_node)
|
||||
|
||||
def test_as_dict_insecure(self):
|
||||
self.node.driver_info['ipmi_password'] = 'fake'
|
||||
|
||||
@ -722,6 +722,8 @@ expected_object_fingerprints = {
|
||||
'Deployment': '1.0-ff10ae028c5968f1596131d85d7f5f9d',
|
||||
'NodeHistory': '1.0-9b576c6481071e7f7eac97317fa29418',
|
||||
'NodeInventory': '1.0-97692fec24e20ab02022b9db54e8f539',
|
||||
'FirmwareComponent': '1.0-0e0720dab959e20247bbcfd5f28958c5',
|
||||
'FirmwareComponentList': '1.0-33a2e1bb91ad4082f9f63429b77c1244',
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -392,3 +392,35 @@ def create_test_inventory(ctxt, node, **kw):
|
||||
setattr(inv, key, value)
|
||||
inv.create()
|
||||
return inv
|
||||
|
||||
|
||||
def get_test_firmware_component(ctxt, **kw):
|
||||
"""Return a FirmwareComponent object with appropriate attributes.
|
||||
|
||||
NOTE: The object leaves the attributes marked as changed, such
|
||||
that a create() could be used to commit it to the DB.
|
||||
"""
|
||||
kw['object_type'] = 'firmware'
|
||||
get_db_fw_checked = check_keyword_arguments(
|
||||
db_utils.get_test_firmware_component)
|
||||
db_fw_cmp = get_db_fw_checked(**kw)
|
||||
|
||||
if 'id' not in kw:
|
||||
del db_fw_cmp['id']
|
||||
|
||||
fw_cmp = objects.FirmwareComponent(ctxt)
|
||||
for key in db_fw_cmp:
|
||||
setattr(fw_cmp, key, db_fw_cmp[key])
|
||||
|
||||
return fw_cmp
|
||||
|
||||
|
||||
def create_test_firmware_component(ctxt, **kw):
|
||||
"""Create and return a test Firmware Component object.
|
||||
|
||||
Create a Firmware Component in the DB and return a FirmwareComponent
|
||||
object with appropriate attributes.
|
||||
"""
|
||||
fw_cmp = get_test_firmware_component(ctxt, **kw)
|
||||
fw_cmp.create()
|
||||
return fw_cmp
|
||||
|
||||
Loading…
Reference in New Issue
Block a user