add yang validation service

Author: Ubuntu

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
@@ -1753,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):
@@ -1769,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
@@ -1789,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
+
+