feat: validate crm high and low threshold input

crm/main.py

@@ -48,6 +48,15 @@ def get_thresholds_list(self):
 
     def get_resources_list(self):
         return list(self.resources)
+
+    def get_crm_config(self):
+        configdb = self.cfgdb
+        if configdb is None:
+            namespaces = multi_asic.get_namespace_list()
+            configdb = ConfigDBConnector(namespace=namespaces[0])
+            configdb.connect()
+
+        return configdb.get_entry('CRM', 'Config')
 
     @multi_asic_util.run_on_multi_asic
     def config(self, attr, val):
@@ -63,15 +72,7 @@ def show_summary(self):
         CRM Handler to display general information.
         """
 
-        configdb = self.cfgdb
-        if configdb is None:
-            # Get the namespace list
-            namespaces = multi_asic.get_namespace_list()
-
-            configdb = ConfigDBConnector(namespace=namespaces[0])
-            configdb.connect()
-
-        crm_info = configdb.get_entry('CRM', 'Config')
+        crm_info = self.get_crm_config()
 
         if crm_info:
             try:
@@ -86,15 +87,8 @@ def show_thresholds(self, resource):
         """
         CRM Handler to display thresholds information.
         """
-        configdb = self.cfgdb
-        if configdb is None:
-            # Get the namespace list
-            namespaces = multi_asic.get_namespace_list()
-
-            configdb = ConfigDBConnector(namespace=namespaces[0])
-            configdb.connect()
 
-        crm_info = configdb.get_entry('CRM', 'Config')
+        crm_info = self.get_crm_config()
 
         header = ("Resource Name", "Threshold Type", "Low Threshold", "High Threshold")
         data = []
@@ -418,27 +412,71 @@ def type(ctx, value):
 @click.pass_context
 def low(ctx, value):
     """CRM low threshold configuration"""
-    attr = ''
 
+    type_attr = ''
     if ctx.obj["crm"].addr_family != None:
-        attr += ctx.obj["crm"].addr_family + '_'
+        type_attr += ctx.obj["crm"].addr_family + '_'
+    type_attr += ctx.obj["crm"].res_type + '_' + 'threshold_type'
 
-    attr += ctx.obj["crm"].res_type + '_' + 'low_threshold'
+    high_attr = ''
+    if ctx.obj["crm"].addr_family != None:
+        high_attr += ctx.obj["crm"].addr_family + '_'
+    high_attr += ctx.obj["crm"].res_type + '_' + 'high_threshold'
+
+    crm_info = ctx.obj["crm"].get_crm_config()
+    threshold_type = crm_info.get(type_attr, 'percentage')
+    if threshold_type == 'percentage':
+        if value < 0 or value > 100:
+            raise click.ClickException(
+                'Low threshold value must be between 0 and 100 for '
+                'percentage type.')
+
+        high_value = crm_info.get(high_attr, None)
+        if high_value is not None and value >= int(high_value):
+            raise click.ClickException(
+                'Low threshold value must be less than high threshold '
+                f'value: {high_value} for percentage type.')
 
+    attr = ''
+    if ctx.obj["crm"].addr_family != None:
+        attr += ctx.obj["crm"].addr_family + '_'
+    attr += ctx.obj["crm"].res_type + '_' + 'low_threshold'
     ctx.obj["crm"].config(attr, value)
 
 @click.command()
 @click.argument('value', type=click.INT)
 @click.pass_context
 def high(ctx, value):
     """CRM high threshold configuration"""
-    attr = ''
 
+    type_attr = ''
     if ctx.obj["crm"].addr_family != None:
-        attr += ctx.obj["crm"].addr_family + '_'
+        type_attr += ctx.obj["crm"].addr_family + '_'
+    type_attr += ctx.obj["crm"].res_type + '_' + 'threshold_type'
 
-    attr += ctx.obj["crm"].res_type + '_' + 'high_threshold'
+    low_attr = ''
+    if ctx.obj["crm"].addr_family != None:
+        low_attr += ctx.obj["crm"].addr_family + '_'
+    low_attr += ctx.obj["crm"].res_type + '_' + 'low_threshold'
+
+    crm_info = ctx.obj["crm"].get_crm_config()
+    threshold_type = crm_info.get(type_attr, 'percentage')
+    if threshold_type == 'percentage':
+        if value < 0 or value > 100:
+            raise click.ClickException(
+                'High threshold value must be between 0 and 100 for '
+                'percentage type.')
+
+        low_value = crm_info.get(low_attr, None)
+        if low_value is not None and value <= int(low_value):
+            raise click.ClickException(
+                'High threshold value must be greater than low threshold '
+                f'value: {low_value} for percentage type.')
 
+    attr = ''
+    if ctx.obj["crm"].addr_family != None:
+        attr += ctx.obj["crm"].addr_family + '_'
+    attr += ctx.obj["crm"].res_type + '_' + 'high_threshold'
     ctx.obj["crm"].config(attr, value)
 
 route.add_command(type)

tests/crm_test.py

@@ -1922,6 +1922,108 @@ def test_crm_multi_asic_show_resources_srv6_nexthop(self):
         assert result.exit_code == 0
         assert result.output == crm_multi_asic_show_resources_srv6_nexthop
 
+    def test_crm_config_thresholds_percentage_over_100(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv4', 'route', 'high', '150'],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 1
+        assert (
+            "Error: High threshold value must be between 0 and 100 for percentage type."
+            in result.output
+        )
+
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv6', 'neighbor', 'low', '101'],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 1
+        assert (
+            "Error: Low threshold value must be between 0 and 100 for percentage type."
+            in result.output
+        )
+
+    def test_crm_config_thresholds_low_greater_than_high(self):
+        runner = CliRunner()
+        db = Db()
+
+        high_value = '90'
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv4', 'route', 'high', high_value],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
+
+        low_value = '95'
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv4', 'route', 'low', low_value],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 1
+        assert (
+            "Error: Low threshold value must be less than high threshold value: "
+            f"{high_value} for percentage type." in result.output
+        )
+
+    def test_crm_config_thresholds_high_less_than_low(self):
+        runner = CliRunner()
+        db = Db()
+        low_value = '50'
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv6', 'route', 'low', low_value],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
+
+        high_value = '40'
+        result = runner.invoke(
+            crm.cli,
+            ['config', 'thresholds', 'ipv6', 'route', 'high', high_value],
+            obj=db,
+        )
+        print(sys.stderr, result.output)
+        assert result.exit_code == 1
+        assert (
+            "Error: High threshold value must be greater than low threshold value: "
+            f"{low_value} for percentage type." in result.output
+        )
+
+    def test_crm_config_thresholds_valid_values(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(
+            crm.cli, ['config', 'thresholds', 'ipv4', 'route', 'high', '100'], obj=db)
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
+
+        result = runner.invoke(
+            crm.cli, ['config', 'thresholds', 'ipv4', 'route', 'low', '50'], obj=db)
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
+
+        result = runner.invoke(
+            crm.cli, ['config', 'thresholds', 'ipv6', 'neighbor', 'low', '30'], obj=db)
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
+
+        result = runner.invoke(
+            crm.cli, ['config', 'thresholds', 'ipv6', 'neighbor', 'high', '80'], obj=db)
+        print(sys.stderr, result.output)
+        assert result.exit_code == 0
 
     @classmethod
     def teardown_class(cls):