[sonic-config-engine] Replace os.system, replace yaml.load, remove subprocess with shell=True (#12607)

Signed-off-by: maipbui <[email protected]>
#### Why I did it
Missing import statement in PR #12533
#### How I did it
Revert [PR 12646](#12616)
Add import statement
1. https://github.com/sonic-net/sonic-buildimage/blob/31f7afa92e8d1e353ef2f3b004711dd18c0d94f6/src/sonic-config-engine/tests/test_j2files_t2_chassis_fe.py#L8
2. https://github.com/sonic-net/sonic-buildimage/blob/31f7afa92e8d1e353ef2f3b004711dd18c0d94f6/src/sonic-config-engine/tests/test_j2files.py#L8
3. https://github.com/sonic-net/sonic-buildimage/blob/31f7afa92e8d1e353ef2f3b004711dd18c0d94f6/src/sonic-config-engine/tests/test_multinpu_cfggen.py#L11
#### How to verify it
Pass UT

Author: maipbui

src/sonic-config-engine/sonic-cfggen

@@ -351,7 +351,7 @@ def main():
             if yaml.__version__ >= "5.1":
                 additional_data = yaml.full_load(stream)
             else:
-                additional_data = yaml.load(stream)
+                additional_data = yaml.safe_load(stream)
             deep_update(data, FormatConverter.to_deserialized(additional_data))
 
     if args.additional_data is not None:

src/sonic-config-engine/tests/common_utils.py

@@ -5,7 +5,6 @@
 import sys
 import subprocess
 import argparse
-import shlex
 
 PY3x = sys.version_info >= (3, 0)
 PYvX_DIR = "py3" if PY3x else "py2"
@@ -47,7 +46,7 @@ def __init__(self, path=YANG_MODELS_DIR):
             self.yang_parser = sonic_yang.SonicYang(path)
             self.yang_parser.loadYangModel()
             self.test_dir = os.path.dirname(os.path.realpath(__file__))
-            self.script_file = PYTHON_INTERPRETTER + ' ' + os.path.join(self.test_dir, '..', 'sonic-cfggen')
+            self.script_file = [PYTHON_INTERPRETTER, os.path.join(self.test_dir, '..', 'sonic-cfggen')]
 
     def validate(self, argument):
         """
@@ -62,22 +61,22 @@ def validate(self, argument):
             parser.add_argument("-p", "--port-config", help="port config file, used with -m or -k", nargs='?', const=None)
             parser.add_argument("-S", "--hwsku-config", help="hwsku config file, used with -p and -m or -k", nargs='?', const=None)
             parser.add_argument("-j", "--json", help="additional json file input, used with -p, -S and -m or -k", nargs='?', const=None)
-            args, unknown = parser.parse_known_args(shlex.split(argument))
+            args, unknown = parser.parse_known_args(argument)
 
             print('\n    Validating yang schema')
-            cmd = self.script_file + ' -m ' + args.minigraph
+            cmd = self.script_file + ['-m', args.minigraph]
             if args.hwsku is not None:
-                cmd += ' -k ' + args.hwsku
+                cmd += ['-k', args.hwsku]
             if args.hwsku_config is not None:
-                cmd += ' -S ' + args.hwsku_config
+                cmd += ['-S', args.hwsku_config]
             if args.port_config is not None:
-                cmd += ' -p ' + args.port_config
+                cmd += ['-p', args.port_config]
             if args.namespace is not None:
-                cmd += ' -n ' + args.namespace
+                cmd += ['-n', args.namespace]
             if args.json is not None:
-                cmd += ' -j ' + args.json
-            cmd += ' --print-data'
-            output = subprocess.check_output(cmd, shell=True).decode()
+                cmd += ['-j', args.json]
+            cmd += ['--print-data']
+            output = subprocess.check_output(cmd).decode()
             try:
                 self.yang_parser.loadData(configdbJson=json.loads(output))
                 self.yang_parser.validate_data_tree()

src/sonic-config-engine/tests/test_cfggen.py

@@ -4,6 +4,7 @@
 import os
 
 import tests.common_utils as utils
+from sonic_py_common.general import getstatusoutput_noshell
 
 
 #TODO: Remove this fixuture once SONiC moves to python3.x
@@ -21,20 +22,18 @@ class TestCfgGen(object):
     @pytest.fixture(autouse=True)
     def setup_teardown(self):
         self.test_dir = os.path.dirname(os.path.realpath(__file__))
-        self.script_file = utils.PYTHON_INTERPRETTER + ' ' + os.path.join(
-            self.test_dir, '..', 'sonic-cfggen')
+        self.script_file = [utils.PYTHON_INTERPRETTER, os.path.join(
+            self.test_dir, '..', 'sonic-cfggen')]
         self.sample_yang_file = os.path.join(self.test_dir,
                                              'test_yang_data.json')
 
     def run_script(self, arg, check_stderr=False):
-        print('\n    Running sonic-cfggen ' + arg)
+        print('\n    Running sonic-cfggen ', arg)
         if check_stderr:
-            output = subprocess.check_output(self.script_file + ' ' + arg,
-                                             stderr=subprocess.STDOUT,
-                                             shell=True)
+            output = subprocess.check_output(self.script_file + arg,
+                                             stderr=subprocess.STDOUT)
         else:
-            output = subprocess.check_output(self.script_file + ' ' + arg,
-                                             shell=True)
+            output = subprocess.check_output(self.script_file + arg)
 
         if utils.PY3x:
             output = output.decode()
@@ -48,32 +47,31 @@ def run_script(self, arg, check_stderr=False):
         return output
 
     def run_diff(self, file1, file2):
-        return subprocess.check_output('diff -u {} {} || true'.format(
-            file1, file2),
-                                       shell=True)
+        _, output = getstatusoutput_noshell(['diff', '-u', file1, file2])
+        return output
 
     def run_script_with_yang_arg(self, arg, check_stderr=False):
-        args = "-Y {} {}".format(self.sample_yang_file, arg)
+        args = ["-Y", self.sample_yang_file] + arg
         return self.run_script(arg=args, check_stderr=check_stderr)
 
     def test_print_data(self):
-        arg = "--print-data"
+        arg = ["--print-data"]
         output = self.run_script_with_yang_arg(arg)
         assert len(output.strip()) > 0
 
 
     def test_jinja_expression(self, expected_router_type='LeafRouter'):
-        arg = " -v \"DEVICE_METADATA[\'localhost\'][\'type\']\" "
+        arg = ["-v", "DEVICE_METADATA[\'localhost\'][\'type\']"]
         output = self.run_script_with_yang_arg(arg)
         assert output.strip() == expected_router_type
 
     def test_hwsku(self):
-        arg = "-v \"DEVICE_METADATA[\'localhost\'][\'hwsku\']\" "
+        arg = ["-v", "DEVICE_METADATA[\'localhost\'][\'hwsku\']"]
         output = self.run_script_with_yang_arg(arg)
         assert output.strip() == "Force10-S6000"
 
     def test_device_metadata(self):
-        arg = "--var-json \"DEVICE_METADATA\" "
+        arg = ["--var-json", "DEVICE_METADATA"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert (output['localhost'] ==  {\
             'bgp_asn': '65100',
@@ -87,7 +85,7 @@ def test_device_metadata(self):
 
 
     def test_port_table(self):
-        arg = "--var-json \"PORT\""
+        arg = ["--var-json", "PORT"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == \
             {'Ethernet0': {'admin_status': 'up', 'alias': 'eth0', 'description': 'Ethernet0', 'fec': 'rs', 'lanes': '65, 66', 'mtu': '9100', 'pfc_asym': 'on', 'speed': '40000'},
@@ -101,7 +99,7 @@ def test_port_table(self):
             })
 
     def test_portchannel_table(self):
-        arg = "--var-json \"PORTCHANNEL\""
+        arg = ["--var-json", "PORTCHANNEL"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == \
             {'PortChannel1001': {'admin_status': 'up',
@@ -116,7 +114,7 @@ def test_portchannel_table(self):
                       'mtu': '9100'}})
 
     def test_portchannel_member_table(self):
-        arg = "--var-json \"PORTCHANNEL_MEMBER\""
+        arg = ["--var-json", "PORTCHANNEL_MEMBER"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output ==\
             {   "PortChannel1001|Ethernet0": {},
@@ -126,7 +124,7 @@ def test_portchannel_member_table(self):
             })
 
     def test_interface_table(self):
-        arg = "--var-json \"INTERFACE\""
+        arg = ["--var-json", "INTERFACE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output =={\
             "Ethernet8": {},
@@ -150,15 +148,15 @@ def test_interface_table(self):
         })
 
     def test_portchannel_interface_table(self):
-        arg = "--var-json \"PORTCHANNEL_INTERFACE\""
+        arg = ["--var-json", "PORTCHANNEL_INTERFACE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output =={\
             "PortChannel1001|10.0.0.1/31": {},
             "PortChannel1002|10.0.0.60/31": {}
             })
 
     def test_loopback_table(self):
-        arg = "--var-json \"LOOPBACK_INTERFACE\""
+        arg = ["--var-json", "LOOPBACK_INTERFACE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "Loopback0": {},
@@ -173,7 +171,7 @@ def test_loopback_table(self):
         })
 
     def test_acl_table(self):
-        arg = "--var-json \"ACL_TABLE\""
+        arg = ["--var-json", "ACL_TABLE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             'DATAACL':   {'policy_desc': 'DATAACL',    'ports': ['PortChannel1001','PortChannel1002'], 'stage': 'ingress', 'type': 'L3'},
@@ -183,7 +181,7 @@ def test_acl_table(self):
             'SSH_ONLY':  {'policy_desc': 'SSH_ONLY',    'services': ['SSH'],         'stage': 'ingress', 'type': 'CTRLPLANE'}})
 
     def test_acl_rule(self):
-        arg = "--var-json \"ACL_RULE\""
+        arg = ["--var-json", "ACL_RULE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "DATAACL|Rule1": {
@@ -201,7 +199,7 @@ def test_acl_rule(self):
         })
 
     def test_vlan_table(self):
-        arg = "--var-json \"VLAN\""
+        arg = ["--var-json", "VLAN"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "Vlan100": {
@@ -218,7 +216,7 @@ def test_vlan_table(self):
         })
 
     def test_vlan_interface(self):
-        arg = "--var-json \"VLAN_INTERFACE\""
+        arg = ["--var-json", "VLAN_INTERFACE"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "Vlan100": {},
@@ -233,7 +231,7 @@ def test_vlan_interface(self):
         })
 
     def test_vlan_member(self):
-        arg = "--var-json \"VLAN_MEMBER\""
+        arg = ["--var-json", "VLAN_MEMBER"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "Vlan100|Ethernet24": {
@@ -245,7 +243,7 @@ def test_vlan_member(self):
         })
 
     def test_vlan_crm(self):
-        arg = "--var-json \"CRM\""
+        arg = ["--var-json", "CRM"]
         output = json.loads(self.run_script_with_yang_arg(arg))
         assert(output == {\
             "Config": {

src/sonic-config-engine/tests/test_cfggen_from_yang.py

@@ -9,13 +9,14 @@ class TestPfxFilter(TestCase):
     def test_comprehensive(self):
         # Generate output
         data_dir = "tests/data/pfx_filter"
-        cmd = "{} ./sonic-cfggen -j {}/param_1.json -t {}/tmpl_1.txt.j2 > /tmp/result_1.txt".format(
-            utils.PYTHON_INTERPRETTER, data_dir, data_dir
-        )
-        subprocess.check_output(cmd, shell=True)
+        output_file = "/tmp/result_1.txt"
+        cmd = [utils.PYTHON_INTERPRETTER, "./sonic-cfggen", "-j", "{}/param_1.json".format(data_dir), "-t", "{}/tmpl_1.txt.j2".format(data_dir)]
+        output = subprocess.check_output(cmd, universal_newlines=True)
+        with open(output_file, 'w') as f:
+            f.write(output)
         # Compare outputs
-        cmd = "diff -u tests/data/pfx_filter/result_1.txt /tmp/result_1.txt"
+        cmd = ["diff", "-u", "tests/data/pfx_filter/result_1.txt", "/tmp/result_1.txt"]
         try:
-            res = subprocess.check_output(cmd, shell=True)
+            res = subprocess.check_output(cmd)
         except subprocess.CalledProcessError as e:
             assert False, "Wrong output. return code: %d, Diff: %s" % (e.returncode, e.output)

src/sonic-config-engine/tests/test_cfggen_pfx_filter.py

@@ -3,7 +3,7 @@
 import os
 import subprocess
 import sys
-
+import ast
 import tests.common_utils as utils
 
 from unittest import TestCase
@@ -21,17 +21,17 @@ class TestCfgGenPlatformJson(TestCase):
 
     def setUp(self):
         self.test_dir = os.path.dirname(os.path.realpath(__file__))
-        self.script_file = utils.PYTHON_INTERPRETTER + ' ' + os.path.join(self.test_dir, '..', 'sonic-cfggen')
+        self.script_file = [utils.PYTHON_INTERPRETTER, os.path.join(self.test_dir, '..', 'sonic-cfggen')]
         self.platform_sample_graph = os.path.join(self.test_dir, 'platform-sample-graph.xml')
         self.platform_json = os.path.join(self.test_dir, 'sample_platform.json')
         self.hwsku_json = os.path.join(self.test_dir, 'sample_hwsku.json')
 
     def run_script(self, argument, check_stderr=False):
-        print('\n    Running sonic-cfggen ' + argument)
+        print('\n    Running sonic-cfggen ', argument)
         if check_stderr:
-            output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True)
+            output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT)
         else:
-            output = subprocess.check_output(self.script_file + ' ' + argument, shell=True)
+            output = subprocess.check_output(self.script_file + argument)
 
         if utils.PY3x:
             output = output.decode()
@@ -44,18 +44,18 @@ def run_script(self, argument, check_stderr=False):
         return output
 
     def test_dummy_run(self):
-        argument = ''
+        argument = []
         output = self.run_script(argument)
         self.assertEqual(output, '')
 
     def test_print_data(self):
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" --print-data'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '--print-data']
         output = self.run_script(argument)
         self.assertTrue(len(output.strip()) > 0)
 
     # Check whether all interfaces present or not as per platform.json
     def test_platform_json_interfaces_keys(self):
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT.keys()|list"'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT.keys()|list"]
         output = self.run_script(argument)
         self.maxDiff = None
 
@@ -71,24 +71,24 @@ def test_platform_json_interfaces_keys(self):
             'Ethernet139', 'Ethernet140', 'Ethernet141', 'Ethernet142', 'Ethernet144'
             ]
 
-        self.assertEqual(sorted(eval(output.strip())), sorted(expected))
+        self.assertEqual(sorted(ast.literal_eval(output.strip())), sorted(expected))
 
     # Check specific Interface with it's proper configuration as per platform.json
     def test_platform_json_specific_ethernet_interfaces(self):
 
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json  + '" -v "PORT[\'Ethernet8\']"'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet8\']"]
         output = self.run_script(argument)
         self.maxDiff = None
         expected = "{'index': '3', 'lanes': '8', 'description': 'Eth3/1', 'mtu': '9100', 'alias': 'Eth3/1', 'pfc_asym': 'off', 'speed': '25000', 'tpid': '0x8100'}"
         self.assertEqual(utils.to_dict(output.strip()), utils.to_dict(expected))
 
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json  + '" -v "PORT[\'Ethernet112\']"'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet112\']"]
         output = self.run_script(argument)
         self.maxDiff = None
         expected = "{'index': '29', 'lanes': '112', 'description': 'Eth29/1', 'mtu': '9100', 'alias': 'Eth29/1', 'pfc_asym': 'off', 'speed': '25000', 'tpid': '0x8100'}"
         self.assertEqual(utils.to_dict(output.strip()), utils.to_dict(expected))
 
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json  + '" -v "PORT[\'Ethernet4\']"'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet4\']"]
         output = self.run_script(argument)
         self.maxDiff = None
         expected = "{'index': '2', 'lanes': '4,5', 'description': 'Eth2/1', 'admin_status': 'up', 'mtu': '9100', 'alias': 'Eth2/1', 'pfc_asym': 'off', 'speed': '50000', 'tpid': '0x8100'}"
@@ -97,7 +97,7 @@ def test_platform_json_specific_ethernet_interfaces(self):
 
     # Check all Interface with it's proper configuration as per platform.json
     def test_platform_json_all_ethernet_interfaces(self):
-        argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json  + '" -v "PORT"'
+        argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT"]
         output = self.run_script(argument)
         self.maxDiff = None