starlingx
/
update
Code Issues Proposed changes

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>
This commit is contained in:
junfeng-li 2024-03-22 14:43:56 +00:00
parent 75de9f16ff
commit b36ee79426
3 changed files with 231 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
software/software/api/controllers/v1/software.py
View File

@ -239,3 +239,16 @@ class SoftwareAPIController(object):
@expose(method='GET', template='json')
def in_sync_controller(self):
return sc.in_sync_controller_api()
@expose(method='GET', template='json')
def software_upgrade(self):
return sc.get_software_upgrade()
@expose(method='GET', template='json')
def software_host_upgrade(self, *args):
args_list = list(args)
if not args_list:
return sc.get_all_software_host_upgrade()
hostname = args_list[0]
return sc.get_one_software_host_upgrade(hostname)

70
software/software/software_controller.py
View File

@ -2915,7 +2915,6 @@ class PatchController(PatchService):
return None
deploy = deploy[0]
deploy_host_list = []
for host in deploy_hosts:
state = host.get("state")
@ -2949,6 +2948,75 @@ class PatchController(PatchService):
func(*args, **kwargs)
self._update_state_to_peer()
def _get_software_upgrade(self):
"""
Get the current software upgrade from/to versions and state
:return: dict of from_release, to_release and state
"""
all_deploy = self.db_api_instance.get_deploy_all()
if not all_deploy:
return None
deploy = all_deploy[0]
from_maj_min_release = utils.get_major_release_version(deploy.get("from_release"))
to_maj_min_release = utils.get_major_release_version(deploy.get("to_release"))
state = deploy.get("state")
return {
"from_release": from_maj_min_release,
"to_release": to_maj_min_release,
"state": state
}
def get_software_upgrade(self):
return self._get_software_upgrade()
def get_all_software_host_upgrade(self):
"""
Get all software host upgrade from/to versions and state
:return: list of dict of hostname, current_sw_version, target_sw_version and host_state
"""
deploy = self._get_software_upgrade()
deploy_hosts = self.db_api_instance.get_deploy_host()
if deploy is None or deploy_hosts is None:
return None
from_maj_min_release = deploy.get("from_release")
to_maj_min_release = deploy.get("to_release")
all_host_upgrades = []
for deploy_host in deploy_hosts:
all_host_upgrades.append({
"hostname": deploy_host.get("hostname"),
"current_sw_version": to_maj_min_release if deploy_host.get(
"state") == constants.DEPLOYED else from_maj_min_release,
"target_sw_version": to_maj_min_release,
"host_state": deploy_host.get("state")
})
return all_host_upgrades
def get_one_software_host_upgrade(self, hostname):
"""
Get the given software host upgrade from/to versions and state
:param hostname: hostname
:return: array of dict of hostname, current_sw_version, target_sw_version and host_state
"""
all_host_upgrades = self.get_all_software_host_upgrade()
if not all_host_upgrades:
return None
for host_upgrade in all_host_upgrades:
if host_upgrade.get("hostname") == hostname:
return [host_upgrade]
return None
class PatchControllerApiThread(threading.Thread):
def __init__(self):

149
software/software/tests/test_software_controller.py
View File

@ -201,3 +201,152 @@ class TestSoftwareController(unittest.TestCase):
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)