Skip to content

Commit

Permalink
[NEXMANAGE-949] Check the IMAGE_BUILD_DATE instead of VERSION for Tib…
Browse files Browse the repository at this point in the history
…erOS SOTA verification (#589)

After SOTA system reboot, the VERSION in /etc/os-release remains the same as before.

However, if we check the image build date at /etc/image-id, it's showing the correct image that we want to flash.

Based on the testing, the dispatcher agent should checks the IMAGE_BUILD_DATE instead of VERSION.

Signed-off-by: yengliong <[email protected]>
  • Loading branch information
yengliong93 authored Oct 30, 2024
1 parent 39b6c9c commit c051447
Show file tree
Hide file tree
Showing 10 changed files with 53 additions and 47 deletions.
4 changes: 2 additions & 2 deletions inbm-lib/inbm_common_lib/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,5 +51,5 @@
# Default signature version
DEFAULT_HASH_ALGORITHM = 384

# Os release path
OS_RELEASE_PATH = '/etc/os-release'
# Image id path
IMAGE_ID_PATH = '/etc/image-id'
32 changes: 16 additions & 16 deletions inbm-lib/inbm_common_lib/utility.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from pathlib import Path
from typing import Iterable, Optional, Union, Dict, Tuple

from inbm_common_lib.constants import VALID_MAGIC_FILE_TYPE_PREFIXES, TEMP_EXT_FOLDER, OS_RELEASE_PATH, UNKNOWN
from inbm_common_lib.constants import VALID_MAGIC_FILE_TYPE_PREFIXES, TEMP_EXT_FOLDER, IMAGE_ID_PATH, UNKNOWN
from inbm_common_lib.shell_runner import PseudoShellRunner

from .constants import URL_NULL_CHAR
Expand Down Expand Up @@ -273,34 +273,34 @@ def validate_file_type(path: list[str]) -> None:
remove_file_list(extracted_file_list)


def get_os_version() -> str:
"""Get os version from os release file.
def get_image_build_date() -> str:
"""Get image build date from image id file.
@return value of the VERSION
@return value of the IMAGE_BUILD_DATE
"""
try:
if os.path.exists(OS_RELEASE_PATH):
with open(OS_RELEASE_PATH, 'r') as version_file:
if os.path.exists(IMAGE_ID_PATH):
with open(IMAGE_ID_PATH, 'r') as version_file:
content = version_file.read()

content_dict = parse_os_release(content)
version = content_dict.get("VERSION")
if version:
return version
logger.error(f"VERSION not found in {OS_RELEASE_PATH}.")
content_dict = parse_image_id(content)
build_date = content_dict.get("IMAGE_BUILD_DATE")
if build_date:
return build_date
logger.error(f"IMAGE_BUILD_DATE not found in {IMAGE_ID_PATH}.")
else:
logger.error(f"{OS_RELEASE_PATH} not exist.")
logger.error(f"{IMAGE_ID_PATH} not exist.")

return UNKNOWN
except OSError as err:
raise OSError(f"Error while reading the os version: {err}")
raise OSError(f"Error while reading the image build date: {err}")


def parse_os_release(file_content: str) -> Dict[str, str]:
def parse_image_id(file_content: str) -> Dict[str, str]:
"""
Parses the content of an os-release file and returns a dictionary of key-value pairs.
Parses the content of an image-id file and returns a dictionary of key-value pairs.
:param file_content: The content of the os-release file as a string.
:param file_content: The content of the image id file as a string.
:return: A dictionary containing key-value pairs from the file.
"""
result: Dict[str, str] = {}
Expand Down
18 changes: 10 additions & 8 deletions inbm-lib/tests/unit/inbm_common_lib/test_utility.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

from inbm_common_lib.exceptions import UrlSecurityException
from inbm_common_lib.utility import clean_input, get_canonical_representation_of_path, canonicalize_uri, \
validate_file_type, remove_file, copy_file, move_file, create_file_with_contents, get_os_version
validate_file_type, remove_file, copy_file, move_file, create_file_with_contents, get_image_build_date
from inbm_common_lib.constants import UNKNOWN


Expand Down Expand Up @@ -112,18 +112,20 @@ def test_create_file_with_contents_successfully(self) -> None:
except IOError as e:
self.fail(f"Unexpected exception raised during test: {e}")

@patch('builtins.open', new_callable=mock_open, read_data='VERSION="2.0.20240802.0213"')
def test_get_os_version_successfully(self, mock_open: Mock) -> None:
@patch('builtins.open', new_callable=mock_open, read_data='IMAGE_BUILD_DATE="20241026100955"')
@patch('os.path.exists', return_value=True)
def test_get_image_build_date_successfully(self, mock_exist: Mock, mock_open: Mock) -> None:
try:
self.assertEqual(get_os_version(), "2.0.20240802.0213")
self.assertEqual(get_image_build_date(), "20241026100955")
except IOError as e:
self.fail(f"Unexpected exception raised during test: {e}")
mock_open.assert_called_once_with('/etc/os-release', 'r')
mock_open.assert_called_once_with('/etc/image-id', 'r')

@patch('builtins.open', new_callable=mock_open, read_data='')
def test_get_os_version_with_no_version_found(self, mock_open: Mock) -> None:
@patch('os.path.exists', return_value=True)
def test_get_image_build_date_with_no_version_found(self, mock_exist: Mock, mock_open: Mock) -> None:
try:
self.assertEqual(get_os_version(), UNKNOWN)
self.assertEqual(get_image_build_date(), UNKNOWN)
except IOError as e:
self.fail(f"Unexpected exception raised during test: {e}")
mock_open.assert_called_once_with('/etc/os-release', 'r')
mock_open.assert_called_once_with('/etc/image-id', 'r')
3 changes: 3 additions & 0 deletions inbm/Changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).

## NEXT - YYYY-MM-DD
### Changed
- (NEXMANAGE-949) Check the IMAGE_BUILD_DATE instead of VERSION for TiberOS SOTA verification

### Added
- (NEXMANAGE-950) Add Set Power State Capability to cloudadapter from UDM

Expand Down
4 changes: 2 additions & 2 deletions inbm/dispatcher-agent/dispatcher/sota/granular_log_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import os
import threading

from inbm_common_lib.utility import get_os_version
from inbm_common_lib.utility import get_image_build_date
from inbm_lib.detect_os import detect_os, LinuxDistType
from inbm_lib.constants import OTA_PENDING, FAIL, OTA_SUCCESS, ROLLBACK, GRANULAR_LOG_FILE

Expand Down Expand Up @@ -44,7 +44,7 @@ def save_granular_log(self, update_logger: UpdateLogger, check_package: bool = T
elif update_logger.detail_status == OTA_SUCCESS or update_logger.detail_status == OTA_PENDING:
log = {
"StatusDetail.Status": update_logger.detail_status,
"Version": get_os_version()
"Version": get_image_build_date()
}
# In TiberOS, no package level information needed.
update_logger.save_granular_log_file(log=log, check_package=False)
Expand Down
8 changes: 4 additions & 4 deletions inbm/dispatcher-agent/dispatcher/sota/snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from inbm_lib.trtl import Trtl
from typing import Any, Dict, Optional
from inbm_common_lib.shell_runner import PseudoShellRunner
from inbm_common_lib.utility import get_os_version
from inbm_common_lib.utility import get_image_build_date
from inbm_common_lib.constants import UNKNOWN
from .constants import MENDER_FILE_PATH
from .mender_util import read_current_mender_version
Expand Down Expand Up @@ -468,9 +468,9 @@ def take_snapshot(self) -> None:
"SOTA attempting to create a dispatcher state file before SOTA {}...".
format(self.sota_cmd))
try:
content = get_os_version()
content = get_image_build_date()
if content == UNKNOWN:
raise SotaError("Failed to get os version.")
raise SotaError("Failed to get image build date.")
state: dispatcher_state.DispatcherState
if dispatcher_state.is_dispatcher_state_file_exists():
consumed_state = dispatcher_state.consume_dispatcher_state_file(readonly=True)
Expand Down Expand Up @@ -539,7 +539,7 @@ def update_system(self) -> None:
state = dispatcher_state.consume_dispatcher_state_file()
if state is not None and 'tiberos-version' in state:
logger.debug("got tiberos-version from state: " + str(state['tiberos-version']))
version = get_os_version()
version = get_image_build_date()
current_tiberos_version = version
previous_tiberos_version = state['tiberos-version']

Expand Down
2 changes: 1 addition & 1 deletion inbm/dispatcher-agent/dispatcher/sota/sota.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

from dispatcher.sota.granular_log_handler import GranularLogHandler
from inbm_common_lib.exceptions import UrlSecurityException
from inbm_common_lib.utility import canonicalize_uri, remove_file, get_os_version
from inbm_common_lib.utility import canonicalize_uri, remove_file
from inbm_common_lib.request_message_constants import SOTA_FAILURE
from inbm_common_lib.constants import REMOTE_SOURCE, LOCAL_SOURCE
from inbm_lib.validate_package_list import parse_and_validate_package_list
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
/etc/opt/csl/csl-node/csl-manager r,
/etc/opt/csl/csl-node/long-lived-token r,
/etc/os-release r,
/etc/image-id r,
/etc/apt/sources.list rw,
/etc/apt/sources.list.bak rw,
/etc/apt/sources.list.d/* rw,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@ class TestGranularLogHandler(testtools.TestCase):
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('dispatcher.sota.granular_log_handler.get_os_version', return_value='2.0.20240802.0213')
@patch('dispatcher.sota.granular_log_handler.get_image_build_date', return_value='20241026100955')
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_version, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_image_build_date, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = OTA_SUCCESS

Expand All @@ -28,7 +28,7 @@ def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_v
"UpdateLog": [
{
"StatusDetail.Status": OTA_SUCCESS,
"Version": '2.0.20240802.0213'
"Version": '20241026100955'
}
]
}
Expand All @@ -40,9 +40,9 @@ def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_v
@patch('os.path.exists', return_value=False)
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('dispatcher.sota.granular_log_handler.get_os_version', return_value='2.0.20240802.0213')
@patch('dispatcher.sota.granular_log_handler.get_image_build_date', return_value='20241026100955')
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_os_version, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_image_build_date, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = OTA_PENDING

Expand All @@ -53,7 +53,7 @@ def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_os_v
"UpdateLog": [
{
"StatusDetail.Status": OTA_PENDING,
"Version": '2.0.20240802.0213'
"Version": '20241026100955'
}
]
}
Expand Down Expand Up @@ -112,9 +112,9 @@ def test_save_granular_in_tiberos_with_rollback_log(self, mock_run, mock_load, m
@patch('os.path.exists', side_effect=[True, False])
@patch('json.dump')
@patch('json.load', return_value={"UpdateLog":[]})
@patch('dispatcher.sota.granular_log_handler.get_os_version', return_value='2.0.20240802.0213')
@patch('dispatcher.sota.granular_log_handler.get_image_build_date', return_value='20241026100955')
@patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
def test_save_granular_in_tiberos_with_truncate_file_being_called(self, mock_run, mock_get_os_version, mock_load, mock_dump, mock_exists) -> None:
def test_save_granular_in_tiberos_with_truncate_file_being_called(self, mock_run, mock_get_image_build_date, mock_load, mock_dump, mock_exists) -> None:
update_logger = UpdateLogger("SOTA", "metadata")
update_logger.detail_status = OTA_SUCCESS

Expand All @@ -128,7 +128,7 @@ def test_save_granular_in_tiberos_with_truncate_file_being_called(self, mock_run
"UpdateLog": [
{
"StatusDetail.Status": OTA_SUCCESS,
"Version": '2.0.20240802.0213'
"Version": '20241026100955'
}
]
}
Expand Down
10 changes: 5 additions & 5 deletions inbm/dispatcher-agent/tests/unit/sota/test_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,9 +205,9 @@ def test_raise_when_unable_to_write_file(

class TestTiberOSSnapshot(unittest.TestCase):
@patch('dispatcher.sota.snapshot.dispatcher_state', autospec=True)
@patch('inbm_common_lib.utility.get_os_version', autospec=True)
@patch('dispatcher.sota.snapshot.get_image_build_date', autospec=True)
def test_take_snapshot_succeeds(self, mock_version, mock_dispatcher_state) -> None:
mock_version.return_value = "2.0.20240802.0213"
mock_version.return_value = "20241026100955"
mock_dispatcher_state.write_dispatcher_state_to_state_file.return_value = True

dispatcher_broker = Mock()
Expand All @@ -220,7 +220,7 @@ def test_take_snapshot_succeeds(self, mock_version, mock_dispatcher_state) -> No
message, = args
assert "unsuccessful" not in message

@patch('dispatcher.sota.snapshot.get_os_version', return_value=UNKNOWN)
@patch('dispatcher.sota.snapshot.get_image_build_date', return_value=UNKNOWN)
def test_take_snapshot_unknown_version_error(self, mock_version) -> None:
dispatcher_broker = Mock()

Expand Down Expand Up @@ -266,9 +266,9 @@ def test_revert_success(self, mock_dispatcher_state) -> None:
assert mock_dispatcher_state.clear_dispatcher_state.call_count == 1
assert rebooter.reboot.call_count == 1

@patch('dispatcher.sota.snapshot.get_os_version', return_value='2.0.20240802.0213')
@patch('dispatcher.sota.snapshot.get_image_build_date', return_value='20241026100955')
@patch('dispatcher.common.dispatcher_state.consume_dispatcher_state_file',
return_value={'restart_reason': 'sota', 'tiberos-version': '2.0.20240802.0213'})
return_value={'restart_reason': 'sota', 'tiberos-version': '20241026100955'})
def test_update_system_raise_error_when_versions_are_same(self, mock_consume_disp_state, mock_version) -> None:
tiberos_snapshot = TiberOSSnapshot(Mock(), "command", Mock(), "1", True, True)
with self.assertRaises(SotaError):
Expand Down

0 comments on commit c051447

Please sign in to comment.