[NEXMANAGE-906] Truncate granular log instead of removing it

The dispatcher agent is unable to remove the granular log file in a read-only system. This PR modifies the dispatcher agent so that it truncates the file instead of removing it. Signed-off-by: yengliong <[email protected]>
intel · Oct 22, 2024 · 014b984 · 014b984
1 parent f7f0afd
commit 014b984
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 10 deletions.
diff --git a/inbm/Changelog.md b/inbm/Changelog.md
@@ -4,6 +4,8 @@ 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-906) Truncate the granular log instead of removing it
 
 ## 4.2.6.1 - 2024-10-18
 ### Added

diff --git a/inbm/dispatcher-agent/dispatcher/sota/granular_log_handler.py b/inbm/dispatcher-agent/dispatcher/sota/granular_log_handler.py
@@ -7,7 +7,7 @@
 import os
 import threading
 
-from inbm_common_lib.utility import remove_file, get_os_version
+from inbm_common_lib.utility import get_os_version
 from inbm_lib.detect_os import detect_os, LinuxDistType
 from inbm_lib.constants import OTA_PENDING, FAIL, OTA_SUCCESS, ROLLBACK, GRANULAR_LOG_FILE
 
@@ -33,7 +33,8 @@ def save_granular_log(self, update_logger: UpdateLogger, check_package: bool = T
             if current_os == LinuxDistType.tiber.name or current_os == LinuxDistType.Mariner.name:
                 # Delete the previous log if exist.
                 if os.path.exists(GRANULAR_LOG_FILE):
-                    remove_file(GRANULAR_LOG_FILE)
+                    with open(GRANULAR_LOG_FILE, "r+") as file:
+                        file.truncate(0)
 
                 if update_logger.detail_status == FAIL or update_logger.detail_status == ROLLBACK:
                     log = {

diff --git a/inbm/dispatcher-agent/tests/unit/sota/test_granular_log_handler.py b/inbm/dispatcher-agent/tests/unit/sota/test_granular_log_handler.py
@@ -11,13 +11,13 @@
 from inbm_lib.constants import OTA_SUCCESS, OTA_PENDING, FAIL, ROLLBACK
 
 class TestGranularLogHandler(testtools.TestCase):
-    @patch('dispatcher.sota.granular_log_handler.remove_file')
+    @patch('typing.IO.truncate')
     @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('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_remove_file) -> None:
+    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:
         update_logger = UpdateLogger("SOTA", "metadata")
         update_logger.detail_status = OTA_SUCCESS
 
@@ -36,13 +36,13 @@ def test_save_granular_in_tiberos_with_success_log(self, mock_run, mock_get_os_v
         mock_dump.assert_called_with(expected_content, m_open(), indent=4)
 
 
-    @patch('dispatcher.sota.granular_log_handler.remove_file')
+    @patch('typing.IO.truncate')
     @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('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_remove_file) -> None:
+    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:
         update_logger = UpdateLogger("SOTA", "metadata")
         update_logger.detail_status = OTA_PENDING
 
@@ -60,12 +60,12 @@ def test_save_granular_in_tiberos_with_pending_log(self, mock_run, mock_get_os_v
 
         mock_dump.assert_called_with(expected_content, m_open(), indent=4)
 
-    @patch('dispatcher.sota.granular_log_handler.remove_file')
+    @patch('typing.IO.truncate')
     @patch('os.path.exists', return_value=False)
     @patch('json.dump')
     @patch('json.load', return_value={"UpdateLog":[]})
     @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
-    def test_save_granular_in_tiberos_with_fail_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
+    def test_save_granular_in_tiberos_with_fail_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
         update_logger = UpdateLogger("SOTA", "metadata")
         update_logger.detail_status = FAIL
         update_logger.error = 'Error getting artifact size from https://registry-rs.internal.ledgepark.intel.com/v2/one-intel-edge/tiberos/manifests/latest using token'
@@ -85,12 +85,12 @@ def test_save_granular_in_tiberos_with_fail_log(self, mock_run, mock_load, mock_
         mock_dump.assert_called_with(expected_content, m_open(), indent=4)
 
 
-    @patch('dispatcher.sota.granular_log_handler.remove_file')
+    @patch('typing.IO.truncate')
     @patch('os.path.exists', return_value=False)
     @patch('json.dump')
     @patch('json.load', return_value={"UpdateLog":[]})
     @patch('inbm_common_lib.shell_runner.PseudoShellRunner.run', return_value=("tiber", "", 0))
-    def test_save_granular_in_tiberos_with_rollback_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_remove_file) -> None:
+    def test_save_granular_in_tiberos_with_rollback_log(self, mock_run, mock_load, mock_dump, mock_exists, mock_truncate) -> None:
         update_logger = UpdateLogger("SOTA", "metadata")
         update_logger.detail_status = ROLLBACK
         update_logger.error = 'FAILED INSTALL: System has not been properly updated; reverting..'
@@ -106,4 +106,31 @@ def test_save_granular_in_tiberos_with_rollback_log(self, mock_run, mock_load, m
             ]
         }
 
+        mock_dump.assert_called_with(expected_content, m_open(), indent=4)
+
+
+    @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('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:
+        update_logger = UpdateLogger("SOTA", "metadata")
+        update_logger.detail_status = OTA_SUCCESS
+
+        with patch('builtins.open', mock_open()) as m_open:
+            mock_file = m_open.return_value.__enter__.return_value
+            mock_file.truncate.return_value = None
+            GranularLogHandler().save_granular_log(update_logger=update_logger, check_package=False)
+            mock_file.truncate.assert_called_once()
+
+        expected_content = {
+            "UpdateLog": [
+                {
+                    "StatusDetail.Status": OTA_SUCCESS,
+                    "Version": '2.0.20240802.0213'
+                }
+            ]
+        }
+
         mock_dump.assert_called_with(expected_content, m_open(), indent=4)