Enhance list-checkpoints CLI (sonic-net#3746)

* Add datetime to list checkpoints

* fix format

* remove unused import

* fix import

* fix CHECKPOINTS_DIR

* fix getmtime mock

* Add a parameter to include modify time.

* Fix UT

* Fix test

* Fix UT

Author: xincunli-sonic

config/main.py

@@ -1795,12 +1795,14 @@ def delete_checkpoint(ctx, checkpoint_name, verbose):
         ctx.fail(ex)
 
 @config.command('list-checkpoints')
+@click.option('-t', '--time', is_flag=True, default=False,
+              help='Add extra last modified time information for each checkpoint')
 @click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing')
 @click.pass_context
-def list_checkpoints(ctx, verbose):
+def list_checkpoints(ctx, time, verbose):
     """List the config checkpoints available."""
     try:
-        checkpoints_list = GenericUpdater().list_checkpoints(verbose)
+        checkpoints_list = GenericUpdater().list_checkpoints(time, verbose)
         formatted_output = json.dumps(checkpoints_list, indent=4)
         click.echo(formatted_output)
     except Exception as ex:

generic_config_updater/generic_updater.py

@@ -3,6 +3,7 @@
 import os
 import subprocess
 
+from datetime import datetime, timezone
 from enum import Enum
 from .gu_common import HOST_NAMESPACE, GenericConfigUpdaterError, EmptyTableError, ConfigWrapper, \
                     DryRunConfigWrapper, PatchWrapper, genericUpdaterLogging
@@ -233,7 +234,7 @@ def checkpoint(self, checkpoint_name):
 
         self.logger.log_notice("Config checkpoint completed.")
 
-    def list_checkpoints(self):
+    def list_checkpoints(self, includes_time=False):
         self.logger.log_info("Listing checkpoints starting.")
 
         self.logger.log_info(f"Verifying checkpoints directory '{self.checkpoints_dir}' exists.")
@@ -244,14 +245,35 @@ def list_checkpoints(self):
         self.logger.log_info("Getting checkpoints in checkpoints directory.")
         checkpoint_names = self.util.get_checkpoint_names()
 
-        checkpoints_len = len(checkpoint_names)
-        self.logger.log_info(f"Found {checkpoints_len} checkpoint{'s' if checkpoints_len != 1 else ''}{':' if checkpoints_len > 0 else '.'}")
-        for checkpoint_name in checkpoint_names:
-            self.logger.log_info(f"  * {checkpoint_name}")
+        checkpoints = []
+        if includes_time:
+            for checkpoint_name in checkpoint_names:
+                checkpoint_path = os.path.join(self.checkpoints_dir, checkpoint_name + CHECKPOINT_EXT)
+                last_modified = datetime.fromtimestamp(os.path.getmtime(checkpoint_path), tz=timezone.utc).isoformat()
+                checkpoints.append({"name": checkpoint_name, "time": last_modified})
+
+            checkpoints.sort(key=lambda x: x["time"], reverse=True)
+        else:
+            checkpoints = checkpoint_names
+
+        checkpoints_len = len(checkpoints)
+        self.logger.log_info(
+            "Found {} checkpoint{}{}".format(
+                checkpoints_len,
+                's' if checkpoints_len != 1 else '',
+                ':' if checkpoints_len > 0 else '.'
+            )
+        )
+
+        for checkpoint in checkpoints:
+            if includes_time:
+                self.logger.log_info(f"  * {checkpoint['name']} (Last Modified: {checkpoint['time']})")
+            else:
+                self.logger.log_info(f"  * {checkpoint}")
 
         self.logger.log_info("Listing checkpoints completed.")
 
-        return checkpoint_names
+        return checkpoints
 
     def delete_checkpoint(self, checkpoint_name):
         self.logger.log_notice("Deleting checkpoint starting.")
@@ -407,8 +429,8 @@ def rollback(self, checkpoint_name):
     def checkpoint(self, checkpoint_name):
         self.decorated_config_rollbacker.checkpoint(checkpoint_name)
 
-    def list_checkpoints(self):
-        return self.decorated_config_rollbacker.list_checkpoints()
+    def list_checkpoints(self, includes_time):
+        return self.decorated_config_rollbacker.list_checkpoints(includes_time)
 
     def delete_checkpoint(self, checkpoint_name):
         self.decorated_config_rollbacker.delete_checkpoint(checkpoint_name)
@@ -617,6 +639,6 @@ def delete_checkpoint(self, checkpoint_name, verbose):
         config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose)
         config_rollbacker.delete_checkpoint(checkpoint_name)
 
-    def list_checkpoints(self, verbose):
+    def list_checkpoints(self, includes_time, verbose):
         config_rollbacker = self.generic_update_factory.create_config_rollbacker(verbose)
-        return config_rollbacker.list_checkpoints()
+        return config_rollbacker.list_checkpoints(includes_time)

tests/config_test.py

@@ -10,7 +10,8 @@
 import sys
 import unittest
 import ipaddress
-import shutil
+
+from datetime import timezone
 from unittest import mock
 from jsonpatch import JsonPatchConflict
 
@@ -1890,7 +1891,14 @@ def setUp(self):
         self.any_target_config_as_text = json.dumps(self.any_target_config)
         self.any_checkpoint_name = "any_checkpoint_name"
         self.any_checkpoints_list = ["checkpoint1", "checkpoint2", "checkpoint3"]
+        self.any_checkpoints_list_with_time = [
+            {"name": "checkpoint1", "time": datetime.datetime.now(timezone.utc).isoformat()},
+            {"name": "checkpoint2", "time": datetime.datetime.now(timezone.utc).isoformat()},
+            {"name": "checkpoint3", "time": datetime.datetime.now(timezone.utc).isoformat()}
+        ]
         self.any_checkpoints_list_as_text = json.dumps(self.any_checkpoints_list, indent=4)
+        self.any_checkpoints_list_with_time_as_text = json.dumps(self.any_checkpoints_list_with_time, indent=4)
+
 
     @patch('config.main.validate_patch', mock.Mock(return_value=True))
     def test_apply_patch__no_params__get_required_params_error_msg(self):
@@ -2476,15 +2484,15 @@ def test_list_checkpoints__help__gets_help_msg(self):
     def test_list_checkpoints__all_optional_params_non_default__non_default_values_used(self):
         # Arrange
         expected_exit_code = 0
-        expected_output = self.any_checkpoints_list_as_text
-        expected_call_with_non_default_values = mock.call(True)
+        expected_output = self.any_checkpoints_list_with_time_as_text
+        expected_call_with_non_default_values = mock.call(True, True)
         mock_generic_updater = mock.Mock()
-        mock_generic_updater.list_checkpoints.return_value = self.any_checkpoints_list
+        mock_generic_updater.list_checkpoints.return_value = self.any_checkpoints_list_with_time
         with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater):
 
             # Act
             result = self.runner.invoke(config.config.commands["list-checkpoints"],
-                                        ["--verbose"],
+                                        ["--time", "--verbose"],
                                         catch_exceptions=False)
 
         # Assert
@@ -2493,6 +2501,26 @@ def test_list_checkpoints__all_optional_params_non_default__non_default_values_u
         mock_generic_updater.list_checkpoints.assert_called_once()
         mock_generic_updater.list_checkpoints.assert_has_calls([expected_call_with_non_default_values])
 
+    def test_list_checkpoints__time_param_true__time_included_in_output(self):
+        # Arrange
+        expected_exit_code = 0
+        expected_output = self.any_checkpoints_list_with_time_as_text
+        expected_call_with_time_param = mock.call(True, False)
+        mock_generic_updater = mock.Mock()
+        mock_generic_updater.list_checkpoints.return_value = self.any_checkpoints_list_with_time
+        with mock.patch('config.main.GenericUpdater', return_value=mock_generic_updater):
+
+            # Act
+            result = self.runner.invoke(config.config.commands["list-checkpoints"],
+                                        ["--time"],
+                                        catch_exceptions=False)
+
+        # Assert
+        self.assertEqual(expected_exit_code, result.exit_code)
+        self.assertTrue(expected_output in result.output)
+        mock_generic_updater.list_checkpoints.assert_called_once()
+        mock_generic_updater.list_checkpoints.assert_has_calls([expected_call_with_time_param])
+
     def test_list_checkpoints__exception_thrown__error_displayed_error_code_returned(self):
         # Arrange
         unexpected_exit_code = 0
@@ -2512,7 +2540,7 @@ def test_list_checkpoints__exception_thrown__error_displayed_error_code_returned
     def test_list_checkpoints__optional_parameters_passed_correctly(self):
         self.validate_list_checkpoints_optional_parameter(
             ["--verbose"],
-            mock.call(True))
+            mock.call(False, True))
 
     def validate_list_checkpoints_optional_parameter(self, param_args, expected_call):
         # Arrange
@@ -4138,6 +4166,7 @@ def test_rollback_multiasic(self, mock_get_checkpoint_content):
         self.assertEqual(result.exit_code, 0, "Command should succeed")
         self.assertIn("Config rolled back successfully.", result.output)
 
+    @patch('os.path.getmtime', mock.Mock(return_value=1700000000.0))
     @patch('generic_config_updater.generic_updater.Util.checkpoints_dir_exist', mock.Mock(return_value=True))
     @patch('generic_config_updater.generic_updater.Util.get_checkpoint_names',
            mock.Mock(return_value=["checkpointname"]))

tests/generic_config_updater/generic_updater_test.py

@@ -2,6 +2,8 @@
 import os
 import shutil
 import unittest
+
+from datetime import datetime, timezone
 from unittest.mock import MagicMock, Mock, call, patch
 from .gutest_helpers import create_side_effect_dict, Files
 
@@ -206,7 +208,7 @@ def test_list_checkpoints__checkpoints_dir_exist_but_no_files__empty_list(self):
         # 'assertCountEqual' does check same count, same elements ignoring order
         self.assertCountEqual(expected, actual)
 
-    def test_list_checkpoints__checkpoints_dir_has_multiple_files__multiple_files(self):
+    def test_list_checkpoints__checkpoints_dir_has_multiple_files(self):
         # Arrange
         self.create_checkpoints_dir()
         self.add_checkpoint(self.any_checkpoint_name, self.any_config)
@@ -221,7 +223,25 @@ def test_list_checkpoints__checkpoints_dir_has_multiple_files__multiple_files(se
         # 'assertCountEqual' does check same count, same elements ignoring order
         self.assertCountEqual(expected, actual)
 
-    def test_list_checkpoints__checkpoints_names_have_special_characters__multiple_files(self):
+    def test_list_checkpoints__checkpoints_dir_has_multiple_files_with_time(self):
+        # Arrange
+        self.create_checkpoints_dir()
+        mod_time1 = self.add_checkpoint(self.any_checkpoint_name, self.any_config)
+        mod_time2 = self.add_checkpoint(self.any_other_checkpoint_name, self.any_config)
+        rollbacker = self.create_rollbacker()
+        expected = [
+            {"name": self.any_checkpoint_name, "time": mod_time1},
+            {"name": self.any_other_checkpoint_name, "time": mod_time2}
+        ]
+
+        # Act
+        actual = rollbacker.list_checkpoints(includes_time=True)
+
+        # Assert
+        # 'assertCountEqual' does check same count, same elements ignoring order
+        self.assertCountEqual(expected, actual)
+
+    def test_list_checkpoints__checkpoints_names_have_special_characters_multiple_files(self):
         # Arrange
         self.create_checkpoints_dir()
         self.add_checkpoint("check.point1", self.any_config)
@@ -237,6 +257,26 @@ def test_list_checkpoints__checkpoints_names_have_special_characters__multiple_f
         # 'assertCountEqual' does check same count, same elements ignoring order
         self.assertCountEqual(expected, actual)
 
+    def test_list_checkpoints__checkpoints_names_have_special_characters_multiple_files_with_time(self):
+        # Arrange
+        self.create_checkpoints_dir()
+        mod_time1 = self.add_checkpoint("check.point1", self.any_config)
+        mod_time2 = self.add_checkpoint(".checkpoint2", self.any_config)
+        mod_time3 = self.add_checkpoint("checkpoint3.", self.any_config)
+        rollbacker = self.create_rollbacker()
+        expected = [
+            {"name": "check.point1", "time": mod_time1},
+            {"name": ".checkpoint2", "time": mod_time2},
+            {"name": "checkpoint3.", "time": mod_time3}
+        ]
+
+        # Act
+        actual = rollbacker.list_checkpoints(includes_time=True)
+
+        # Assert
+        # 'assertCountEqual' does check same count, same elements ignoring order
+        self.assertCountEqual(expected, actual)
+
     def test_delete_checkpoint__checkpoint_does_not_exist__failure(self):
         # Arrange
         rollbacker = self.create_rollbacker()
@@ -279,18 +319,60 @@ def test_multiple_operations(self):
         rollbacker.delete_checkpoint(self.any_other_checkpoint_name)
         self.assertCountEqual([], rollbacker.list_checkpoints())
 
+    def test_multiple_operations_with_time(self):
+        rollbacker = self.create_rollbacker()
+
+        # 'assertCountEqual' does check same count, same elements ignoring order
+        self.assertCountEqual([], rollbacker.list_checkpoints(includes_time=True))
+
+        rollbacker.checkpoint(self.any_checkpoint_name)
+        self.assertCountEqual(
+            [self.any_checkpoint_name],
+            self.extract_checkpoint_names(rollbacker.list_checkpoints(includes_time=True))
+        )
+        self.assertEqual(self.any_config, self.get_checkpoint(self.any_checkpoint_name))
+
+        rollbacker.rollback(self.any_checkpoint_name)
+        rollbacker.config_replacer.replace.assert_has_calls([call(self.any_config)])
+
+        rollbacker.checkpoint(self.any_other_checkpoint_name)
+        self.assertCountEqual(
+            [self.any_checkpoint_name, self.any_other_checkpoint_name],
+            self.extract_checkpoint_names(rollbacker.list_checkpoints(includes_time=True))
+        )
+        self.assertEqual(self.any_config, self.get_checkpoint(self.any_other_checkpoint_name))
+
+        rollbacker.delete_checkpoint(self.any_checkpoint_name)
+        self.assertCountEqual(
+            [self.any_other_checkpoint_name],
+            self.extract_checkpoint_names(rollbacker.list_checkpoints(includes_time=True))
+        )
+
+        rollbacker.delete_checkpoint(self.any_other_checkpoint_name)
+        self.assertCountEqual([], rollbacker.list_checkpoints(includes_time=True))
+
+    def extract_checkpoint_names(self, checkpoint_list):
+        """Extract checkpoint names from the list of dictionaries."""
+        return [checkpoint["name"] for checkpoint in checkpoint_list]
+
     def clean_up(self):
         if os.path.isdir(self.checkpoints_dir):
             shutil.rmtree(self.checkpoints_dir)
 
     def create_checkpoints_dir(self):
         os.makedirs(self.checkpoints_dir)
 
-    def add_checkpoint(self, name, json_content):
+    def add_checkpoint(self, name, json_content, mod_time=None):
         path=os.path.join(self.checkpoints_dir, f"{name}{self.checkpoint_ext}")
         with open(path, "w") as fh:
             fh.write(json.dumps(json_content))
 
+        if mod_time is None:
+            mod_time = datetime.now(timezone.utc).timestamp()
+        os.utime(path, (mod_time, mod_time))
+
+        return datetime.fromtimestamp(mod_time, tz=timezone.utc).isoformat()
+
     def get_checkpoint(self, name):
         path=os.path.join(self.checkpoints_dir, f"{name}{self.checkpoint_ext}")
         with open(path) as fh:
@@ -519,8 +601,14 @@ def setUp(self):
         self.any_checkpoint_name = "anycheckpoint"
         self.any_other_checkpoint_name = "anyothercheckpoint"
         self.any_checkpoints_list = [self.any_checkpoint_name, self.any_other_checkpoint_name]
+        self.any_checkpoints_list_with_time = [
+            {"name": self.any_checkpoint_name, "time": datetime.now(timezone.utc).isoformat()},
+            {"name": self.any_other_checkpoint_name, "time": datetime.now(timezone.utc).isoformat()}
+        ]
+
         self.any_config_format = gu.ConfigFormat.SONICYANG
         self.any_verbose = True
+        self.list_checkpoints_includes_time = True
         self.any_dry_run = True
         self.any_ignore_non_yang_tables = True
         self.any_ignore_paths = ["", "/ACL_TABLE"]
@@ -651,7 +739,26 @@ def test_list_checkpoints__creates_rollbacker_and_list_checkpoints(self):
         expected = self.any_checkpoints_list
 
         # Act
-        actual = generic_updater.list_checkpoints(self.any_verbose)
+        actual = generic_updater.list_checkpoints(False, self.any_verbose)
+
+        # Assert
+        self.assertCountEqual(expected, actual)
+
+    def test_list_checkpoints__creates_rollbacker_and_list_checkpoints_with_time(self):
+        # Arrange
+        config_rollbacker = Mock()
+        config_rollbacker.list_checkpoints.return_value = self.any_checkpoints_list_with_time
+
+        factory = Mock()
+        factory.create_config_rollbacker.side_effect = \
+            create_side_effect_dict({(str(self.any_verbose),): config_rollbacker})
+
+        generic_updater = gu.GenericUpdater(factory)
+
+        expected = self.any_checkpoints_list_with_time
+
+        # Act
+        actual = generic_updater.list_checkpoints(self.list_checkpoints_includes_time, self.any_verbose)
 
         # Assert
         self.assertCountEqual(expected, actual)
@@ -710,7 +817,7 @@ def test_list_checkpoints__calls_decorated_rollbacker(self):
         expected = self.any_checkpoints_list
 
         # Act
-        actual = self.decorator.list_checkpoints()
+        actual = self.decorator.list_checkpoints(includes_time=False)
 
         # Assert
         self.decorated_config_rollbacker.list_checkpoints.assert_called_once()