Add CLI to configure YANG config validation

config/main.py

@@ -30,6 +30,7 @@
 from utilities_common.general import load_db_config
 
 from .utils import log
+from .yang_validation_service import YangValidationService
 
 from . import aaa
 from . import chassis_modules
@@ -1714,6 +1715,21 @@ def synchronous_mode(sync_mode):
     else:
         raise click.BadParameter("Error: Invalid argument %s, expect either enable or disable" % sync_mode)
 
+#
+# 'yang_config_validation' command ('config yang_config_validation ...')
+# 
+@config.command('yang_config_validation')
+@click.argument('yang_config_validation', metavar='<enable|disable>', required=True)
+def yang_config_validation(yang_config_validation):
+    # Enable or disable YANG validation on updates to ConfigDB
+    if yang_config_validation == 'enable' or yang_config_validation == 'disable':
+        config_db = ConfigDBConnector()
+        config_db.connect()
+        config_db.mod_entry('DEVICE_METADATA', 'localhost', {"yang_config_validation": yang_config_validation})
+        click.echo("""Wrote %s yang config validation into CONFIG_DB""" % yang_config_validation)
+    else:
+        raise click.BadParameter("Error: Invalid argument %s, expect either enable or disable" % yang_config_validation)
+
 #
 # 'portchannel' group ('config portchannel ...')
 #
@@ -1738,10 +1754,6 @@ def portchannel(ctx, namespace):
 @click.pass_context
 def add_portchannel(ctx, portchannel_name, min_links, fallback):
     """Add port channel"""
-    if is_portchannel_name_valid(portchannel_name) != True:
-        ctx.fail("{} is invalid!, name should have prefix '{}' and suffix '{}'"
-                 .format(portchannel_name, CFG_PORTCHANNEL_PREFIX, CFG_PORTCHANNEL_NO))
-
     db = ctx.obj['db']
 
     if is_portchannel_present_in_db(db, portchannel_name):
@@ -1754,17 +1766,18 @@ def add_portchannel(ctx, portchannel_name, min_links, fallback):
         fvs['min_links'] = str(min_links)
     if fallback != 'false':
         fvs['fallback'] = 'true'
-    db.set_entry('PORTCHANNEL', portchannel_name, fvs)
+    yvs = YangValidationService()
+    if not yvs.validate_set_entry('PORTCHANNEL', portchannel_name, fvs):
+        ctx.fail("Invalid configuration based on PortChannel YANG model")
+    else:
+        db.set_entry('PORTCHANNEL', portchannel_name, fvs)
+
 
 @portchannel.command('del')
 @click.argument('portchannel_name', metavar='<portchannel_name>', required=True)
 @click.pass_context
 def remove_portchannel(ctx, portchannel_name):
     """Remove port channel"""
-    if is_portchannel_name_valid(portchannel_name) != True:
-        ctx.fail("{} is invalid!, name should have prefix '{}' and suffix '{}'"
-                 .format(portchannel_name, CFG_PORTCHANNEL_PREFIX, CFG_PORTCHANNEL_NO))
-
     db = ctx.obj['db']
 
     # Dont proceed if the port channel does not exist
@@ -1774,7 +1787,11 @@ def remove_portchannel(ctx, portchannel_name):
     if len([(k, v) for k, v in db.get_table('PORTCHANNEL_MEMBER') if k == portchannel_name]) != 0:
         click.echo("Error: Portchannel {} contains members. Remove members before deleting Portchannel!".format(portchannel_name))
     else:
-        db.set_entry('PORTCHANNEL', portchannel_name, None)
+        yvs = YangValidationService()
+        if not yvs.validate_set_entry('PORTCHANNEL', portchannel_name, None):
+            ctx.fail("Invalid configuration based on PortChannel YANG model")
+        else:
+            db.set_entry('PORTCHANNEL', portchannel_name, None)
 
 @portchannel.group(cls=clicommon.AbbreviationGroup, name='member')
 @click.pass_context

config/yang_validation_service.py

@@ -0,0 +1,51 @@
+import json
+import sonic_yang
+import subprocess
+import copy
+
+YANG_DIR = "/usr/local/yang-models"
+
+class YangValidationService:
+    def __init__(self, yang_dir = YANG_DIR):
+        self.yang_dir = YANG_DIR
+        self.sonic_yang_with_loaded_models = None
+
+    def get_config_db_as_json(self):
+        cmd = "show runningconfiguration all"
+        result = subprocess.Popen(cmd, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        text, err = result.communicate()
+        return_code = result.returncode
+        if return_code:
+            raise RuntimeError("Failed to get running config, Return Code: {}, Error: {}".format(return_code, err))
+        return json.loads(text)
+
+    def create_sonic_yang_with_loaded_models(self):
+        if self.sonic_yang_with_loaded_models is None:
+            loaded_models_sy = sonic_yang.SonicYang(self.yang_dir)
+            loaded_models_sy.loadYangModel()
+            self.sonic_yang_with_loaded_models = loaded_models_sy
+
+        return copy.copy(self.sonic_yang_with_loaded_models)
+
+    def validate_config_db_config(self, config_json):
+        sy = self.create_sonic_yang_with_loaded_models()
+        try:
+            tmp_config_json = copy.deepcopy(config_json)
+            sy.loadData(tmp_config_json)
+            sy.validate_data_tree()
+            return True
+        except sonic_yang.SonicYangException as ex:
+            return False
+        return False
+
+    def validate_set_entry(self, table, key, data):
+        config_json = self.get_config_db_as_json()
+        if not self.validate_config_db_config(config_json):
+            return False
+        if data is not None:
+            config_json[table][key] = data
+        if not self.validate_config_db_config(config_json):
+            return False
+        return True
+
+

tests/yang_config_validation_test.py

@@ -0,0 +1,33 @@
+from click.testing import CliRunner
+import config.main as config
+
+class TestYangConfigValidation(object):
+    @classmethod
+    def setup_class(cls):
+        print("SETUP")
+
+    def __check_result(self, result_msg, mode):
+        if mode == "enable" or mode == "disable":
+            expected_msg = """Wrote %s yang config validation into CONFIG_DB""" % mode
+        else:
+            expected_msg = "Error: Invalid argument %s, expect either enable or disable" % mode
+
+        return expected_msg in result_msg
+
+    def test_yang_config_validation(self):
+        runner = CliRunner()
+
+        result = runner.invoke(config.config.commands["yang_config_validation"], ["enable"])
+        print(result.output)
+        assert result.exit_code == 0
+        assert self.__check_result(result.output, "enable")
+
+        result = runner.invoke(config.config.commands["yang_config_validation"], ["disable"])
+        print(result.output)
+        assert result.exit_code == 0
+        assert self.__check_result(result.output, "disable")
+
+        result = runner.invoke(config.config.commands["yang_config_validation"], ["invalid-input"])
+        print(result.output)
+        assert result.exit_code != 0
+        assert self.__check_result(result.output, "invalid-input")