starlingx/distcloud
Code Issues Proposed changes

Merge "Objectify firmware audit data"

Browse Source
This commit is contained in:
Zuul 2021-03-15 14:15:56 +00:00 committed by Gerrit Code Review
commit 4c3d7fb9c8
2 changed files with 64 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
distributedcloud/dcmanager/audit/firmware_audit.py
View File

@ -34,6 +34,31 @@ from dcmanager.common import consts
LOG = logging.getLogger(__name__)
class FirmwareAuditData(object):
def __init__(self, uuid, applied, pci_vendor,
pci_device, applied_labels):
self.uuid = uuid
self.applied = applied
self.pci_vendor = pci_vendor
self.pci_device = pci_device
self.applied_labels = applied_labels
def to_dict(self):
return {
'uuid': self.uuid,
'applied': self.applied,
'pci_vendor': self.pci_vendor,
'pci_device': self.pci_device,
'applied_labels': self.applied_labels,
}
@classmethod
def from_dict(cls, values):
if values is None:
return None
return cls(**values)
class FirmwareAudit(object):
"""Manages tasks related to firmware audits."""
@ -77,7 +102,11 @@ class FirmwareAudit(object):
# Filter images which have been applied on RegionOne
for image in local_device_images:
if image.applied:
filtered_images.append(image)
filtered_images.append(FirmwareAuditData(image.uuid,
image.applied,
image.pci_vendor,
image.pci_device,
image.applied_labels))
LOG.debug("RegionOne applied_images: %s" % filtered_images)
except Exception:
LOG.exception('Cannot retrieve device images for RegionOne, '
@ -224,7 +253,10 @@ class FirmwareAudit(object):
# Check that all device images applied in RegionOne
# are applied and installed on this subcloud
# The audit_data for region one is a dictionary
for image in audit_data:
# audit_data will be a dict from passing through RPC, so objectify
image = FirmwareAuditData.from_dict(image)
proceed = self._check_subcloud_device_has_image(image,
enabled_host_device_list,
subcloud_device_image_states,

31
distributedcloud/dcmanager/tests/unit/audit/test_firmware_audit_manager.py
View File

@ -376,6 +376,13 @@ class TestFirmwareAudit(base.DCManagerTestCase):
self.mock_audit_worker_api.return_value = self.fake_audit_worker_api
self.addCleanup(p.stop)
def _rpc_convert(self, object_list):
# Convert to dict like what would happen calling via RPC
dict_results = []
for result in object_list:
dict_results.append(result.to_dict())
return dict_results
def test_init(self):
fm = firmware_audit.FirmwareAudit(self.ctxt,
self.fake_dcmanager_api)
@ -406,6 +413,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -439,6 +449,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -472,6 +485,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -505,6 +521,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -538,6 +557,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -571,6 +593,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -604,6 +629,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [
@ -637,6 +665,9 @@ class TestFirmwareAudit(base.DCManagerTestCase):
patch_audit_data, firmware_audit_data, kubernetes_audit_data = \
am._get_audit_data(True, True, True)
# Convert to dict like what would happen calling via RPC
firmware_audit_data = self._rpc_convert(firmware_audit_data)
for name in ['subcloud1', 'subcloud2']:
fm.subcloud_firmware_audit(name, firmware_audit_data)
expected_calls = [