starlingx/update
Code Issues Proposed changes
update/software/software/tests/test_software_controller.py
junfeng-li b36ee79426 Replace software upgrade sysinv data 
This commit is to replace legacy sysinv endpoints for platform
upgrade with new USM endpoints.

New endpoints:
get_software_upgrade: get from/to versions and deploy state
for upgrade.

get_software_host_upgrade: get current/target versions and
deploy state for the given host.

Test Plan:

PASS: call the endpoints using curl before deploy start
PASS: call the endpoints using curl after deploy start.

Task: 49766
Story: 2010676
Change-Id: I574da8f60e1cf9fc046f5c6a727f7e17fe8c55f7
Signed-off-by: junfeng-li <junfeng.li@windriver.com>
2024-04-11 15:20:25 +00:00

353 lines
17 KiB
Python
Raw Blame History

#
# SPDX-License-Identifier: Apache-2.0
#
# Copyright (c) 2023-2024 Wind River Systems, Inc.
#
from software.software_controller import PatchController
from software.software_controller import ReleaseValidationFailure
import unittest
from unittest.mock import MagicMock
from unittest.mock import mock_open
from unittest.mock import patch
from software import constants
class TestSoftwareController(unittest.TestCase):
def setUp(self):
self.upgrade_files = {
constants.ISO_EXTENSION: "test.iso",
constants.SIG_EXTENSION: "test.sig"
}
def tearDown(self):
pass
@patch('software.software_controller.PatchController.__init__', return_value=None)
@patch('software.software_controller.verify_files')
@patch('software.software_controller.mount_iso_load')
@patch('software.software_controller.shutil.copyfile')
@patch('software.software_controller.os.chmod')
@patch('software.software_controller.read_upgrade_support_versions')
@patch('software.software_controller.subprocess.run')
@patch('software.software_controller.shutil.copytree')
@patch('software.software_controller.parse_release_metadata')
@patch('software.software_controller.unmount_iso_load')
@patch('software.software_controller.PatchController.major_release_upload_check')
def test_process_upload_upgrade_files(self,
mock_major_release_upload_check,
mock_unmount_iso_load,
mock_parse_release_metadata,
mock_copytree, # pylint: disable=unused-argument
mock_run,
mock_read_upgrade_support_versions,
mock_chmod, # pylint: disable=unused-argument
mock_copyfile, # pylint: disable=unused-argument
mock_mount_iso_load,
mock_verify_files,
mock_init): # pylint: disable=unused-argument
controller = PatchController()
controller.release_data = MagicMock()
# Mock the return values of the mocked functions
mock_major_release_upload_check.return_value = True
mock_verify_files.return_value = True
mock_mount_iso_load.return_value = '/test/iso'
mock_read_upgrade_support_versions.return_value = (
'2.0.0', [{'version': '1.0.0'}, ])
mock_run.return_value.returncode = 0
mock_run.return_value.stdout = 'Load import successful'
mock_parse_release_metadata.return_value = {"id": 1, "sw_version": "2.0.0"}
# Call the function being tested
with patch('software.software_controller.SW_VERSION', '1.0.0'):
info, warning, error, release_meta_info = controller._process_upload_upgrade_files(self.upgrade_files, # pylint: disable=protected-access
controller.release_data)
# Verify that the expected functions were called with the expected arguments
mock_verify_files.assert_called_once_with([self.upgrade_files[constants.ISO_EXTENSION]],
self.upgrade_files[constants.SIG_EXTENSION])
mock_mount_iso_load.assert_called_once_with(
self.upgrade_files[constants.ISO_EXTENSION], constants.TMP_DIR)
mock_read_upgrade_support_versions.assert_called_once_with('/test/iso')
self.assertEqual(mock_run.call_args[0][0], ["%s/rel-%s/bin/%s" % (
constants.SOFTWARE_STORAGE_DIR,
release_meta_info["test.iso"]["sw_version"],
"usm_load_import"),
"--from-release=1.0.0", "--to-release=2.0.0", "--iso-dir=/test/iso"])
mock_unmount_iso_load.assert_called_once_with('/test/iso')
# Verify that the expected messages were returned
self.assertEqual(
info,
'iso and signature files upload completed\nImporting iso is in progress\nLoad import successful')
self.assertEqual(warning, '')
self.assertEqual(error, '')
self.assertEqual(
release_meta_info,
{"test.iso": {"id": 1, "sw_version": "2.0.0"},
"test.sig": {"id": None, "sw_version": None}})
@patch('software.software_controller.PatchController.__init__', return_value=None)
@patch('software.software_controller.verify_files')
@patch('software.software_controller.mount_iso_load')
@patch('software.software_controller.unmount_iso_load')
@patch('software.software_controller.PatchController.major_release_upload_check')
def test_process_upload_upgrade_files_invalid_signature(self,
mock_major_release_upload_check,
mock_unmount_iso_load, # pylint: disable=unused-argument
mock_mount_iso_load,
mock_verify_files,
mock_init): # pylint: disable=unused-argument
controller = PatchController()
controller.release_data = MagicMock()
# Mock the return values of the mocked functions
mock_verify_files.return_value = False
mock_mount_iso_load.return_value = '/test/iso'
mock_major_release_upload_check.return_value = True
# Call the function being tested
with patch('software.software_controller.SW_VERSION', '1.0'):
info, warning, error, _ = controller._process_upload_upgrade_files(self.upgrade_files, # pylint: disable=protected-access
controller.release_data)
# Verify that the expected messages were returned
self.assertEqual(info, '')
self.assertEqual(warning, '')
self.assertEqual(error, 'Upgrade file signature verification failed\n')
@patch('software.software_controller.PatchController.__init__', return_value=None)
@patch('software.software_controller.verify_files',
side_effect=ReleaseValidationFailure('Invalid signature file'))
@patch('software.software_controller.PatchController.major_release_upload_check')
def test_process_upload_upgrade_files_validation_error(self,
mock_major_release_upload_check,
mock_verify_files,
mock_init): # pylint: disable=unused-argument
controller = PatchController()
controller.release_data = MagicMock()
mock_verify_files.return_value = False
mock_major_release_upload_check.return_value = True
# Call the function being tested
info, warning, error, _ = controller._process_upload_upgrade_files(self.upgrade_files, # pylint: disable=protected-access
controller.release_data)
# Verify that the expected messages were returned
self.assertEqual(info, '')
self.assertEqual(warning, '')
self.assertEqual(error, 'Upgrade file signature verification failed\n')
@patch('software.software_controller.os.path.isfile')
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_in_sync_controller_api_files_identical(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load,
mock_isfile):
controller = PatchController()
# Test when both files exist and are identical
mock_isfile.side_effect = [True, True]
mock_json_load.side_effect = [{'key': 'value'}, {'key': 'value'}]
result = controller.in_sync_controller_api()
self.assertEqual(result, {"in_sync": True})
@patch('software.software_controller.os.path.isfile')
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_in_sync_controller_api_files_not_identical(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load,
mock_isfile):
controller = PatchController()
# Test when both files exist and are not identical
mock_isfile.side_effect = [True, True]
mock_json_load.side_effect = [{'key': 'value1'}, {'key': 'value2'}]
result = controller.in_sync_controller_api()
self.assertEqual(result, {"in_sync": False})
@patch('software.software_controller.os.path.isfile')
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_in_sync_controller_api_files_not_exist(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
mock_isfile):
controller = PatchController()
# Test when neither file exists
mock_isfile.side_effect = [False, False]
result = controller.in_sync_controller_api()
self.assertEqual(result, {"in_sync": True})
@patch('software.software_controller.os.path.isfile')
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_in_sync_controller_api_one_file_exist(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
mock_isfile):
controller = PatchController()
# Test when only one file exists
mock_isfile.side_effect = [True, False]
result = controller.in_sync_controller_api()
self.assertEqual(result, {"in_sync": False})
mock_isfile.side_effect = [False, True]
result = controller.in_sync_controller_api()
self.assertEqual(result, {"in_sync": False})
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_software_host_upgrade_deployed(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
controller._get_software_upgrade = MagicMock(return_value={ # pylint: disable=protected-access
"from_release": "1.0.0",
"to_release": "2.0.0"
})
controller.db_api_instance.get_deploy_host = MagicMock(return_value=[
{"hostname": "host1", "state": constants.DEPLOYED},
{"hostname": "host2", "state": constants.DEPLOYING}
])
# Test when the host is deployed
result = controller.get_one_software_host_upgrade("host1")
self.assertEqual(result, [{
"hostname": "host1",
"current_sw_version": "2.0.0",
"target_sw_version": "2.0.0",
"host_state": constants.DEPLOYED
}])
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_software_host_upgrade_deploying(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
controller._get_software_upgrade = MagicMock(return_value={ # pylint: disable=protected-access
"from_release": "1.0.0",
"to_release": "2.0.0"
})
controller.db_api_instance.get_deploy_host = MagicMock(return_value=[
{"hostname": "host1", "state": constants.DEPLOYED},
{"hostname": "host2", "state": constants.DEPLOYING}
])
# Test when the host is deploying
result = controller.get_one_software_host_upgrade("host2")
self.assertEqual(result, [{
"hostname": "host2",
"current_sw_version": "1.0.0",
"target_sw_version": "2.0.0",
"host_state": constants.DEPLOYING
}])
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_all_software_host_upgrade_deploying(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
controller._get_software_upgrade = MagicMock(return_value={ # pylint: disable=protected-access
"from_release": "1.0.0",
"to_release": "2.0.0"
})
controller.db_api_instance.get_deploy_host = MagicMock(return_value=[
{"hostname": "host1", "state": constants.DEPLOYED},
{"hostname": "host2", "state": constants.DEPLOYING}
])
# Test when the host is deploying
result = controller.get_all_software_host_upgrade()
self.assertEqual(result, [{
"hostname": "host1",
"current_sw_version": "2.0.0",
"target_sw_version": "2.0.0",
"host_state": constants.DEPLOYED
}, {
"hostname": "host2",
"current_sw_version": "1.0.0",
"target_sw_version": "2.0.0",
"host_state": constants.DEPLOYING
}])
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_software_host_upgrade_none_state(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
# Test when the deploy or deploy_hosts is None
controller._get_software_upgrade = MagicMock(return_value=None) # pylint: disable=protected-access
controller.db_api_instance.get_deploy_host.return_value = None
result = controller.get_one_software_host_upgrade("host1")
self.assertIsNone(result)
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_software_upgrade_get_deploy_all(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
# Create a mock instance of the db_api
db_api_instance_mock = MagicMock()
controller.db_api_instance = db_api_instance_mock
# Create a mock return value for the get_deploy_all method
deploy_all_mock = [{"from_release": "1.0.0", "to_release": "2.0.0", "state": "start"}]
db_api_instance_mock.get_deploy_all.return_value = deploy_all_mock
# Call the method being tested
result = controller._get_software_upgrade() # pylint: disable=protected-access
# Verify that the expected methods were called
db_api_instance_mock.get_deploy_all.assert_called_once()
# Verify the expected result
expected_result = {
"from_release": "1.0",
"to_release": "2.0",
"state": "start"
}
self.assertEqual(result, expected_result)
@patch('software.software_controller.json.load')
@patch('software.software_controller.open', new_callable=mock_open)
def test_get_software_upgrade_get_deploy_all_none(self,
mock_dummy, # pylint: disable=unused-argument
mock_json_load, # pylint: disable=unused-argument
):
controller = PatchController()
# Create a mock instance of the db_api
db_api_instance_mock = MagicMock()
controller.db_api_instance = db_api_instance_mock
# Create a mock return value for the get_deploy_all method
db_api_instance_mock.get_deploy_all.return_value = None
# Call the method being tested
result = controller._get_software_upgrade() # pylint: disable=protected-access
# Verify that the expected methods were called
db_api_instance_mock.get_deploy_all.assert_called_once()
self.assertEqual(result, None)
View Git Blame Copy Permalink