Working on coverage

sonic-chassisd/scripts/chassisd

@@ -925,7 +925,7 @@ class SmartSwitchModuleUpdater(ModuleUpdater):
                 self.chassis_state_db.delete(key)
 
         # Fetch the list of reboot cause history files
-        history_path = f"/host/reboot-cause/module/{module.lower()}/history/*.txt"
+        history_path = f"/host/reboot-cause/module/{module.lower()}/history/*_reboot_cause.txt"
         reboot_cause_files = glob.glob(history_path)
 
         if not reboot_cause_files:

sonic-chassisd/tests/mock_platform.py

@@ -177,54 +177,6 @@ def get_dataplane_state(self):
     def get_controlplane_state(self):
         raise NotImplementedError
 
-    def _is_first_boot(self, module):
-        """Checks if the reboot-cause file indicates a first boot."""
-        file_path = os.path.join("path_to_reboot_cause_dir", module.lower(), "reboot-cause.txt")
-        try:
-            with open(file_path, 'r') as f:
-                content = f.read().strip()
-                return content == "First boot"
-        except FileNotFoundError:
-            return False
-
-    def persist_dpu_reboot_cause(self, reboot_cause, module):
-        """Mock implementation to persist reboot cause."""
-        file_path = os.path.join("path_to_reboot_cause_dir", module.lower(), "reboot-cause.txt")
-        with open(file_path, 'w') as f:
-            f.write(reboot_cause)
-
-    def _rotate_files(self, module):
-        """Mock implementation to simulate file rotation."""
-        os.rename(f"{module}_reboot-cause.txt", f"{module}_reboot-cause.old.txt")
-        os.remove(f"{module}_reboot-cause.older.txt")
-
-    def update_dpu_reboot_cause_to_db(self, module):
-        """Mock implementation to update reboot cause in a database."""
-        pass
-
-    def _get_current_time_str(self):
-        """Mock implementation to get the current time string."""
-        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-
-    def _get_history_path(self, module, file_name):
-        """Mock implementation to get the history file path."""
-        return os.path.join("path_to_history_dir", module.lower(), file_name)
-
-    def persist_dpu_reboot_time(self, module):
-        """Mock implementation to persist the current reboot time."""
-        file_path = os.path.join("path_to_reboot_time_dir", module.lower(), "reboot-time.txt")
-        with open(file_path, 'w') as f:
-            f.write(self._get_current_time_str())
-
-    def retrieve_dpu_reboot_time(self, module):
-        """Mock implementation to retrieve the persisted reboot time."""
-        file_path = os.path.join("path_to_reboot_time_dir", module.lower(), "reboot-time.txt")
-        try:
-            with open(file_path, 'r') as f:
-                return f.read().strip()
-        except FileNotFoundError:
-            return None
-
 class MockDpuChassis:
 
     def get_dpu_id(self):

sonic-chassisd/tests/test_chassisd.py

@@ -114,6 +114,44 @@ def test_smartswitch_moduleupdater_check_valid_fields():
     assert status == fvs[CHASSIS_MODULE_INFO_OPERSTATUS_FIELD]
     assert serial == fvs[CHASSIS_MODULE_INFO_SERIAL_FIELD]
 
+def test_smartswitch_moduleupdater_status_transition_to_offline():
+    # Mock the chassis and module
+    chassis = MockSmartSwitchChassis()
+    index = 0
+    name = "DPU0"
+    desc = "DPU Module 0"
+    slot = 0
+    serial = "DPU0-0000"
+    module_type = ModuleBase.MODULE_TYPE_DPU
+    module = MockModule(index, name, desc, module_type, slot, serial)
+
+    # Set initial state to ONLINE
+    initial_status = ModuleBase.MODULE_STATUS_ONLINE
+    module.set_oper_status(initial_status)
+    chassis.module_list.append(module)
+
+    # Create the updater and update module_db
+    module_updater = SmartSwitchModuleUpdater(SYSLOG_IDENTIFIER, chassis)
+    module_updater.module_db_update()
+
+    # Update module's operational status to OFFLINE
+    offline_status = ModuleBase.MODULE_STATUS_OFFLINE
+    module.set_oper_status(offline_status)
+    module_updater.module_db_update()
+
+    # Verify the "offline" logic was executed
+    fvs = module_updater.module_table.get(name)
+    if isinstance(fvs, list):
+        fvs = dict(fvs[-1])
+    assert desc == fvs[CHASSIS_MODULE_INFO_DESC_FIELD]
+    assert NOT_AVAILABLE == fvs[CHASSIS_MODULE_INFO_SLOT_FIELD]
+    assert offline_status == fvs[CHASSIS_MODULE_INFO_OPERSTATUS_FIELD]
+    assert serial == fvs[CHASSIS_MODULE_INFO_SERIAL_FIELD]
+
+    # Ensure the reboot time is persisted for the module
+    # Mock or verify persist_dpu_reboot_time() was called with the correct key
+    assert module_updater.retrieve_dpu_reboot_time(name) is not None
+
 def test_smartswitch_moduleupdater_check_invalid_name():
     chassis = MockSmartSwitchChassis()
     index = 0