Add unit test

inbm-lib/inbm_common_lib/utility.py

@@ -287,4 +287,5 @@ def get_os_version() -> str:
  logger.error(f"VERSION not found in {OS_RELEASE_PATH}.")
  else:
  logger.error(f"{OS_RELEASE_PATH} not exist.")
- return UNKNOWN
+
+ return UNKNOWN

inbm/dispatcher-agent/dispatcher/sota/snapshot.py

@@ -13,9 +13,10 @@
 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.constants import UNKNOWN
 from .constants import MENDER_FILE_PATH
 from .mender_util import read_current_mender_version
-from .update_tool_util import update_tool_version_command, update_tool_commit_command
+from .update_tool_util import update_tool_commit_command
 from .rebooter import Rebooter
 from ..common import dispatcher_state
 from .sota_error import SotaError
@@ -468,6 +469,8 @@ def take_snapshot(self) -> None:
  format(self.sota_cmd))
  try:
  content = get_os_version()
+ if content == UNKNOWN:
+ raise SotaError("Failed to get os version.")
  state: dispatcher_state.DispatcherState
  if dispatcher_state.is_dispatcher_state_file_exists():
  consumed_state = dispatcher_state.consume_dispatcher_state_file(read=True)

inbm/dispatcher-agent/tests/unit/sota/test_downloader.py

@@ -1,14 +1,18 @@
 import unittest
 from typing import Optional
 import os
+import hashlib
 
 from ..common.mock_resources import *
 from dispatcher.dispatcher_exception import DispatcherException
 from dispatcher.packagemanager.memory_repo import MemoryRepo
+from dispatcher.packagemanager.local_repo import DirectoryRepo
 from dispatcher.sota.os_factory import SotaOsFactory
+from dispatcher.sota.downloader import TiberOSDownloader
 from dispatcher.sota.sota import SOTA
 from dispatcher.sota.sota_error import SotaError
 from inbm_lib.xmlhandler import XmlHandler
+from inbm_lib.constants import CACHE
 from unittest.mock import patch
 
 TEST_SCHEMA_LOCATION = os.path.join(os.path.dirname(__file__),
@@ -110,3 +114,39 @@ def _build_mock_repo(num_files=0):
  for i in range(0, num_files):
  mem_repo.add("test" + str(i + 1) + ".rpm", b"0123456789")
  return mem_repo
+
+ @patch('dispatcher.sota.downloader.oras_download')
+ def test_tiberos_download_successful(self, mock_download) -> None:
+ self.release_date = self.username = None
+ password = "mock_password"
+ mock_url = canonicalize_uri("https://registry-rs.internal.ledgepark.intel.com/one-intel-edge/tiberos:latest")
+
+ assert TestDownloader.sota_instance
+ TestDownloader.sota_instance.factory = SotaOsFactory(
+ MockDispatcherBroker.build_mock_dispatcher_broker(), None, []).get_os('TiberOS')
+ factory = TestDownloader.sota_instance.factory
+ assert factory
+ installer = factory.create_downloader()
+ assert isinstance(installer, TiberOSDownloader)
+
+ if not os.path.exists(CACHE):
+ os.makedirs(CACHE)
+
+ file_path = os.path.join(CACHE, 'test')
+ with open(file_path, 'w') as file:
+ file.write('This is a test file.')
+ # Calculate the SHA256 checksum
+ sha256_hash = hashlib.sha256()
+ with open(file_path, 'rb') as file:
+ for chunk in iter(lambda: file.read(4096), b''):
+ sha256_hash.update(chunk)
+ checksum = sha256_hash.hexdigest()
+ installer._signature = checksum
+ try:
+ installer.download(self.mock_disp_broker,
+ mock_url, DirectoryRepo(CACHE),
+ self.username, password, self.release_date)
+ except (SotaError, DispatcherException):
+ self.fail("raised Error unexpectedly!")
+
+ mock_download.assert_called_once()

inbm/dispatcher-agent/tests/unit/sota/test_snapshot.py

@@ -7,6 +7,7 @@
 from dispatcher.dispatcher_exception import DispatcherException
 from dispatcher.sota.sota_error import SotaError
 from dispatcher.sota.os_factory import DebianBasedSnapshot, YoctoSnapshot, SotaOsFactory, TiberOSSnapshot
+from inbm_common_lib.constants import UNKNOWN
 
 
 @ddt
@@ -204,9 +205,9 @@ def test_raise_when_unable_to_write_file(
 
 class TestTiberOSSnapshot(unittest.TestCase):
  @patch('dispatcher.sota.snapshot.dispatcher_state', autospec=True)
- @patch('dispatcher.sota.snapshot.update_tool_version_command', autospec=True)
+ @patch('inbm_common_lib.utility.get_os_version', autospec=True)
  def test_take_snapshot_succeeds(self, mock_version, mock_dispatcher_state) -> None:
- mock_version.return_value = "2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae"
+ mock_version.return_value = "2.0.20240802.0213"
  mock_dispatcher_state.write_dispatcher_state_to_state_file.return_value = True
 
  dispatcher_broker = Mock()
@@ -219,15 +220,14 @@ def test_take_snapshot_succeeds(self, mock_version, mock_dispatcher_state) -> No
  message, = args
  assert "unsuccessful" not in message
 
- @patch("inbm_common_lib.shell_runner.PseudoShellRunner.run", return_value=('', "", 1))
- def test_take_snapshot_ut_version_command_error(self, mock_run) -> None:
+ @patch('dispatcher.sota.snapshot.get_os_version', return_value=UNKNOWN)
+ def test_take_snapshot_unknown_version_error(self, mock_version) -> None:
  dispatcher_broker = Mock()
 
  tiberos_snapshot = TiberOSSnapshot(Mock(), "command", dispatcher_broker, "1", True, True)
  with self.assertRaises(SotaError):
  tiberos_snapshot.take_snapshot()
-
- mock_run.assert_called_once()
+ mock_version.assert_called_once()
 
  @patch("inbm_common_lib.shell_runner.PseudoShellRunner.run", return_value=('', "", 0))
  @patch('dispatcher.sota.snapshot.dispatcher_state', autospec=True)
@@ -266,13 +266,12 @@ 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("inbm_common_lib.shell_runner.PseudoShellRunner.run", return_value=('2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae', "", 0))
+ @patch('dispatcher.sota.snapshot.get_os_version', return_value='2.0.20240802.0213')
  @patch('dispatcher.common.dispatcher_state.consume_dispatcher_state_file',
- return_value={'restart_reason': 'sota', 'tiberos-version': '2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae'})
- def test_update_system_raise_error_when_versions_are_same(self, mock_consume_disp_state, mock_run) -> None:
- rebooter = Mock()
+ return_value={'restart_reason': 'sota', 'tiberos-version': '2.0.20240802.0213'})
+ 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):
  tiberos_snapshot.update_system()
  mock_consume_disp_state.assert_called_once()
- mock_run.assert_called_once()
+ mock_version.assert_called_once()

inbm/dispatcher-agent/tests/unit/sota/test_sota.py

@@ -11,6 +11,7 @@
 from dispatcher.sota.sota import SOTAUtil
 from dispatcher.sota.constants import *
 from inbm_lib.xmlhandler import XmlHandler
+from inbm_lib.constants import OTA_SUCCESS, OTA_PENDING, FAIL, ROLLBACK, GRANULAR_LOG_FILE
 
 
 TEST_SCHEMA_LOCATION = os.path.join(os.path.dirname(__file__),
@@ -270,3 +271,100 @@ def test_is_ota_no_update_available_return_false(self) -> None:
 Processing triggers for systemd (245.4-4ubuntu3.23) ...
  """
  self.assertFalse(TestSota.sota_instance._is_ota_no_update_available(cmd_list))
+
+ @patch('dispatcher.sota.sota.get_os_version', return_value='2.0.20240802.0213')
+ @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("TiberOS", "", 0))
+ def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_version) -> None:
+ TestSota.sota_instance.factory = SotaOsFactory(
+ TestSota.mock_disp_broker,
+ None, []).get_os('TiberOS')
+
+ TestSota.sota_instance._update_logger.detail_status = OTA_SUCCESS
+ TestSota.sota_instance.save_granular_log(check_package=False)
+
+ with open(GRANULAR_LOG_FILE, 'r') as f:
+ granular_log = json.load(f)
+
+ expected_content = {
+ "UpdateLog": [
+ {
+ "StatusDetail.Status": OTA_SUCCESS,
+ "OS Version": '2.0.20240802.0213'
+ }
+ ]
+ }
+
+ self.assertEqual(granular_log, expected_content)
+
+ @patch('dispatcher.sota.sota.get_os_version', return_value='2.0.20240802.0213')
+ @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("TiberOS", "", 0))
+ def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_os_version) -> None:
+ TestSota.sota_instance.factory = SotaOsFactory(
+ TestSota.mock_disp_broker,
+ None, []).get_os('TiberOS')
+
+ TestSota.sota_instance._update_logger.detail_status = OTA_PENDING
+ TestSota.sota_instance.save_granular_log(check_package=False)
+
+ with open(GRANULAR_LOG_FILE, 'r') as f:
+ granular_log = json.load(f)
+
+ expected_content = {
+ "UpdateLog": [
+ {
+ "StatusDetail.Status": OTA_PENDING,
+ "OS Version": '2.0.20240802.0213'
+ }
+ ]
+ }
+
+ self.assertEqual(granular_log, expected_content)
+
+ @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("TiberOS", "", 0))
+ def test_save_granular_in_tiberos_with_fail_log(self, mock_run) -> None:
+ TestSota.sota_instance.factory = SotaOsFactory(
+ TestSota.mock_disp_broker,
+ None, []).get_os('TiberOS')
+
+ TestSota.sota_instance._update_logger.detail_status = FAIL
+ TestSota.sota_instance._update_logger.error = 'Error getting artifact size from https://registry-rs.internal.ledgepark.intel.com/v2/one-intel-edge/tiberos/manifests/latest using token'
+ TestSota.sota_instance.save_granular_log(check_package=False)
+
+ with open(GRANULAR_LOG_FILE, 'r') as f:
+ granular_log = json.load(f)
+
+ expected_content = {
+ "UpdateLog": [
+ {
+ "StatusDetail.Status": FAIL,
+ "FailureReason": 'Error getting artifact size from https://registry-rs.internal.ledgepark.intel.com/v2/one-intel-edge/tiberos/manifests/latest using token'
+ }
+ ]
+ }
+
+ self.assertEqual(granular_log, expected_content)
+
+
+ @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("TiberOS", "", 0))
+ def test_save_granular_in_tiberos_with_rollback_log(self, mock_run) -> None:
+ TestSota.sota_instance.factory = SotaOsFactory(
+ TestSota.mock_disp_broker,
+ None, []).get_os('TiberOS')
+
+ TestSota.sota_instance._update_logger.detail_status = ROLLBACK
+ TestSota.sota_instance._update_logger.error = 'FAILED INSTALL: System has not been properly updated; reverting..'
+ TestSota.sota_instance.save_granular_log(check_package=False)
+
+ with open(GRANULAR_LOG_FILE, 'r') as f:
+ granular_log = json.load(f)
+
+ expected_content = {
+ "UpdateLog": [
+ {
+ "StatusDetail.Status": ROLLBACK,
+ "FailureReason": 'FAILED INSTALL: System has not been properly updated; reverting..'
+ }
+ ]
+ }
+
+ self.assertEqual(granular_log, expected_content)