diff --git a/src/sonic-config-engine/sonic-cfggen b/src/sonic-config-engine/sonic-cfggen index d5358f633d..287640d8a1 100755 --- a/src/sonic-config-engine/sonic-cfggen +++ b/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: diff --git a/src/sonic-config-engine/tests/common_utils.py b/src/sonic-config-engine/tests/common_utils.py index 72325ecbc1..d2be32c854 100644 --- a/src/sonic-config-engine/tests/common_utils.py +++ b/src/sonic-config-engine/tests/common_utils.py @@ -5,7 +5,6 @@ import re 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 @@ class YangWrapper(object): 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 @@ class YangWrapper(object): 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() diff --git a/src/sonic-config-engine/tests/test_cfggen.py b/src/sonic-config-engine/tests/test_cfggen.py index 6279a7bfbd..d1d5b898b4 100644 --- a/src/sonic-config-engine/tests/test_cfggen.py +++ b/src/sonic-config-engine/tests/test_cfggen.py @@ -1,7 +1,6 @@ import json import subprocess import os - import tests.common_utils as utils from unittest import TestCase @@ -16,7 +15,7 @@ class TestCfgGen(TestCase): def setUp(self): self.yang = utils.YangWrapper() 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_graph = os.path.join(self.test_dir, 'sample_graph.xml') self.sample_graph_t0 = os.path.join(self.test_dir, 't0-sample-graph.xml') self.sample_graph_simple = os.path.join(self.test_dir, 'simple-sample-graph.xml') @@ -52,13 +51,12 @@ class TestCfgGen(TestCase): pass def run_script(self, argument, check_stderr=False, verbose=False): - print('\n Running sonic-cfggen ' + argument) + print('\n Running sonic-cfggen ' + ' '.join(argument)) self.assertTrue(self.yang.validate(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() @@ -73,52 +71,52 @@ class TestCfgGen(TestCase): return output def test_dummy_run(self): - argument = '' + argument = [] output = self.run_script(argument) self.assertEqual(output, '') def test_device_desc(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'hwsku\']" -M "' + self.sample_device_desc + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'hwsku\']", "-M", self.sample_device_desc] output = self.run_script(argument) self.assertEqual(output.strip(), 'ACS-MSN2700') def test_device_desc_mgmt_ip(self): - argument = '-v "(MGMT_INTERFACE.keys()|list)[0]" -M "' + self.sample_device_desc + '"' + argument = ['-v', "(MGMT_INTERFACE.keys()|list)[0]", '-M', self.sample_device_desc] output = self.run_script(argument) self.assertEqual(output.strip(), "('eth0', '10.0.1.5/28')") def test_minigraph_hostname(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'hostname\']" -m "' + self.sample_graph + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'hostname\']", '-m', self.sample_graph, "-p", self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'OCPSCH01040DDLF') def test_minigraph_sku(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'hwsku\']" -m "' + self.sample_graph + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'hwsku\']", '-m', self.sample_graph, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'Force10-Z9100') def test_minigraph_region(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'region\']" -m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'region\']", '-m', self.sample_graph_metadata, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'usfoo') def test_minigraph_cloudtype(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'cloudtype\']" -m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'cloudtype\']", '-m', self.sample_graph_metadata, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'Public') def test_minigraph_resourcetype(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'resource_type\']" -m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'resource_type\']", '-m', self.sample_graph_metadata, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'resource_type_x') def test_minigraph_downstream_subrole(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'downstream_subrole\']" -m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'downstream_subrole\']", '-m', self.sample_graph_metadata, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'downstream_subrole_y') def test_print_data(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" --print-data' + argument = ['-m', self.sample_graph, '-p', self.port_config, '--print-data'] output = self.run_script(argument) self.assertTrue(len(output.strip()) > 0) @@ -127,29 +125,29 @@ class TestCfgGen(TestCase): graph = self.sample_graph if port_config is None: port_config = self.port_config - argument = '-m "' + graph + '" -p "' + port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'type\']"' + argument = ['-m', graph, '-p', port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'type\']"] output = self.run_script(argument) self.assertEqual(output.strip(), expected_router_type) def test_additional_json_data(self): - argument = '-a \'{"key1":"value1"}\' -v key1' + argument = ['-a', '{"key1":"value1"}', '-v', 'key1'] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1') def test_additional_json_data_level1_key(self): - argument = '-a \'{"k1":{"k11":"v11","k12":"v12"}, "k2":{"k22":"v22"}}\' --var-json k1' + argument = ['-a', '{"k1":{"k11":"v11","k12":"v12"}, "k2":{"k22":"v22"}}', '--var-json', 'k1'] output = self.run_script(argument) self.assertEqual(utils.to_dict(output.strip()), utils.to_dict('{\n "k11": "v11", \n "k12": "v12"\n}')) def test_additional_json_data_level2_key(self): - argument = '-a \'{"k1":{"k11":"v11","k12":"v12"},"k2":{"k22":"v22"}}\' --var-json k1 -K k11' + argument = ['-a', '{"k1":{"k11":"v11","k12":"v12"},"k2":{"k22":"v22"}}', '--var-json', 'k1', '-K', 'k11'] output = self.run_script(argument) self.assertEqual(utils.to_dict(output.strip()), utils.to_dict('{\n "k11": "v11"\n}')) def test_var_json_data(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) tag_mode = kwargs.get('tag_mode', 'untagged') - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" --var-json VLAN_MEMBER' + argument = ['-m', graph_file, '-p', self.port_config, '--var-json', 'VLAN_MEMBER'] output = self.run_script(argument) if tag_mode == "tagged": self.assertEqual( @@ -175,20 +173,20 @@ class TestCfgGen(TestCase): ) def test_read_yaml(self): - argument = '-v yml_item -y ' + os.path.join(self.test_dir, 'test.yml') + argument = ['-v', 'yml_item', '-y', os.path.join(self.test_dir, 'test.yml')] output = self.run_script(argument) self.assertEqual(output.strip(), '[\'value1\', \'value2\']') def test_render_template(self): - argument = '-y ' + os.path.join(self.test_dir, 'test.yml') + ' -t ' + os.path.join(self.test_dir, 'test.j2') + argument = ['-y', os.path.join(self.test_dir, 'test.yml'), '-t', os.path.join(self.test_dir, 'test.j2')] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1\nvalue2') def test_template_batch_mode(self): - argument = '-y ' + os.path.join(self.test_dir, 'test.yml') - argument += ' -a \'{"key1":"value"}\'' - argument += ' -t ' + os.path.join(self.test_dir, 'test.j2') + ',' + self.output_file - argument += ' -t ' + os.path.join(self.test_dir, 'test2.j2') + ',' + self.output2_file + argument = ['-y', os.path.join(self.test_dir, 'test.yml')] + argument += ['-a', '{"key1":"value"}'] + argument += ['-t', os.path.join(self.test_dir, 'test.j2') + ',' + self.output_file] + argument += ['-t', os.path.join(self.test_dir, 'test2.j2') + ',' + self.output2_file] output = self.run_script(argument) assert(os.path.exists(self.output_file)) assert(os.path.exists(self.output2_file)) @@ -199,10 +197,10 @@ class TestCfgGen(TestCase): def test_template_json_batch_mode(self): data = {"key1_1":"value1_1", "key1_2":"value1_2", "key2_1":"value2_1", "key2_2":"value2_2"} - argument = " -a '{0}'".format(repr(data).replace('\'', '"')) - argument += ' -t ' + os.path.join(self.test_dir, 'sample-template-1.json.j2') + ",config-db" - argument += ' -t ' + os.path.join(self.test_dir, 'sample-template-2.json.j2') + ",config-db" - argument += ' --print-data' + argument = ["-a", '{0}'.format(repr(data).replace('\'', '"'))] + argument += ['-t', os.path.join(self.test_dir, 'sample-template-1.json.j2') + ",config-db"] + argument += ['-t', os.path.join(self.test_dir, 'sample-template-2.json.j2') + ",config-db"] + argument += ['--print-data'] output = self.run_script(argument) output_data = json.loads(output) for key, value in data.items(): @@ -212,7 +210,7 @@ class TestCfgGen(TestCase): # it is not at all intuitive what that ordering should be. Could make it # more robust by adding better parsing logic. def test_minigraph_acl(self): - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v ACL_TABLE' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', 'ACL_TABLE'] output = self.run_script(argument, True, True) self.assertEqual( utils.to_dict(output.strip().replace("Warning: Ignoring Control Plane ACL NTP_ACL without type\n", '')), @@ -236,7 +234,7 @@ class TestCfgGen(TestCase): # self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '2.2.2.2'}}") def test_minigraph_mgmt_ports(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v MGMT_PORT' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v','MGMT_PORT'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -244,13 +242,13 @@ class TestCfgGen(TestCase): ) def test_minigraph_interfaces(self): - argument = '-m "' + self.sample_graph_simple + '" -p "' + self.port_config + '" -v "INTERFACE.keys()|list"' + argument = ['-m', self.sample_graph_simple, '-p', self.port_config, '-v', "INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual(output.strip(), "[('Ethernet0', '10.0.0.58/31'), 'Ethernet0', ('Ethernet0', 'FC00::75/126')]") def test_minigraph_vlans(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v VLAN' + argument = ['-m', graph_file, '-p', self.port_config, '-v', 'VLAN'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -266,7 +264,7 @@ class TestCfgGen(TestCase): def test_minigraph_vlan_members(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) tag_mode = kwargs.get('tag_mode', 'untagged') - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v VLAN_MEMBER' + argument = ['-m', graph_file, '-p', self.port_config, '-v', 'VLAN_MEMBER'] output = self.run_script(argument) if tag_mode == "tagged": self.assertEqual( @@ -293,12 +291,12 @@ class TestCfgGen(TestCase): def test_minigraph_vlan_interfaces(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "VLAN_INTERFACE.keys()|list"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "VLAN_INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual(output.strip(), "[('Vlan1000', '192.168.0.1/27'), 'Vlan1000']") def test_minigraph_ecmp_fg_nhg(self): - argument = '-m "' + self.ecmp_graph + '" -p "' + self.mlnx_port_config + '" -v FG_NHG' + argument = ['-m', self.ecmp_graph, '-p', self.mlnx_port_config, '-v', 'FG_NHG'] output = self.run_script(argument) print(output.strip()) self.assertEqual(utils.to_dict(output.strip()), @@ -308,7 +306,7 @@ class TestCfgGen(TestCase): )) def test_minigraph_ecmp_members(self): - argument = '-m "' + self.ecmp_graph + '" -p "' + self.mlnx_port_config + '" -v "FG_NHG_MEMBER.keys()|list|sort"' + argument = ['-m', self.ecmp_graph, '-p', self.mlnx_port_config, '-v', "FG_NHG_MEMBER.keys()|list|sort"] output = self.run_script(argument) self.assertEqual(output.strip(), "['200.200.200.1', '200.200.200.10', '200.200.200.2', '200.200.200.3', '200.200.200.4', '200.200.200.5'," " '200.200.200.6', '200.200.200.7', '200.200.200.8', '200.200.200.9', '200:200:200:200::1', '200:200:200:200::10'," @@ -316,7 +314,7 @@ class TestCfgGen(TestCase): " '200:200:200:200::7', '200:200:200:200::8', '200:200:200:200::9']") def test_minigraph_ecmp_neighbors(self): - argument = '-m "' + self.ecmp_graph + '" -p "' + self.mlnx_port_config + '" -v "NEIGH.keys()|list|sort"' + argument = ['-m', self.ecmp_graph, '-p', self.mlnx_port_config, '-v', "NEIGH.keys()|list|sort"] output = self.run_script(argument) self.assertEqual(output.strip(), "['Vlan31|200.200.200.1', 'Vlan31|200.200.200.10', 'Vlan31|200.200.200.2', 'Vlan31|200.200.200.3'," " 'Vlan31|200.200.200.4', 'Vlan31|200.200.200.5', 'Vlan31|200.200.200.6', 'Vlan31|200.200.200.7'," @@ -326,7 +324,7 @@ class TestCfgGen(TestCase): def test_minigraph_portchannels(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v PORTCHANNEL' + argument = ['-m', graph_file, '-p', self.port_config, '-v', 'PORTCHANNEL'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -334,14 +332,14 @@ class TestCfgGen(TestCase): ) def test_minigraph_portchannel_with_more_member(self): - argument = '-m "' + self.sample_graph_pc_test + '" -p "' + self.port_config + '" -v PORTCHANNEL' + argument = ['-m', self.sample_graph_pc_test, '-p', self.port_config, '-v', 'PORTCHANNEL'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), utils.to_dict("{'PortChannel01': {'admin_status': 'up', 'min_links': '3', 'members': ['Ethernet112', 'Ethernet116', 'Ethernet120', 'Ethernet124'], 'mtu': '9100', 'tpid': '0x8100', 'lacp_key': 'auto'}}")) def test_minigraph_portchannel_members(self): - argument = '-m "' + self.sample_graph_pc_test + '" -p "' + self.port_config + '" -v "PORTCHANNEL_MEMBER.keys()|list"' + argument = ['-m', self.sample_graph_pc_test, '-p', self.port_config, '-v', "PORTCHANNEL_MEMBER.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -349,7 +347,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_portchannel_interfaces(self): - argument = '-m "' + self.sample_graph_simple + '" -p "' + self.port_config + '" -v "PORTCHANNEL_INTERFACE.keys()|list"' + argument = ['-m', self.sample_graph_simple, '-p', self.port_config, '-v', "PORTCHANNEL_INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -357,7 +355,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_neighbors(self): - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v "DEVICE_NEIGHBOR[\'Ethernet124\']"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', "DEVICE_NEIGHBOR[\'Ethernet124\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -368,7 +366,7 @@ class TestCfgGen(TestCase): # it is not at all intuitive what that ordering should be. Could make it # more robust by adding better parsing logic. def test_minigraph_extra_neighbors(self): - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v DEVICE_NEIGHBOR' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', 'DEVICE_NEIGHBOR'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -382,7 +380,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_port_description(self): - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v "PORT[\'Ethernet124\']"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', "PORT[\'Ethernet124\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -391,7 +389,7 @@ class TestCfgGen(TestCase): def test_minigraph_port_fec_disabled(self): # Test for FECDisabled - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v "PORT[\'Ethernet4\']"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', "PORT[\'Ethernet4\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -400,7 +398,7 @@ class TestCfgGen(TestCase): def test_minigraph_port_autonegotiation(self): # Test with a port_config.ini file which doesn't have an 'autoneg' column - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v "PORT"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', "PORT"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -441,7 +439,7 @@ class TestCfgGen(TestCase): ) # Test with a port_config.ini file which has an 'autoneg' column - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config_autoneg + '" -v "PORT"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config_autoneg, '-v', "PORT"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -482,7 +480,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_port_rs(self): - argument = '-m "' + self.sample_graph_t0 + '" -p "' + self.port_config + '" -v "PORT[\'Ethernet124\']"' + argument = ['-m', self.sample_graph_t0, '-p', self.port_config, '-v', "PORT[\'Ethernet124\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -490,7 +488,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_bgp(self): - argument = '-m "' + self.sample_graph_bgp_speaker + '" -p "' + self.port_config + '" -v "BGP_NEIGHBOR[\'10.0.0.59\']"' + argument = ['-m', self.sample_graph_bgp_speaker, '-p', self.port_config, '-v', "BGP_NEIGHBOR[\'10.0.0.59\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -498,7 +496,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_peers_with_range(self): - argument = "-m " + self.sample_graph_bgp_speaker + " -p " + self.port_config + " -v \"BGP_PEER_RANGE.values()|list\"" + argument = ["-m", self.sample_graph_bgp_speaker, "-p", self.port_config, "-v", "BGP_PEER_RANGE.values()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -506,24 +504,24 @@ class TestCfgGen(TestCase): ) def test_minigraph_deployment_id(self): - argument = '-m "' + self.sample_graph_bgp_speaker + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'deployment_id\']"' + argument = ['-m', self.sample_graph_bgp_speaker, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'deployment_id\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "1") def test_minigraph_deployment_id_null(self): - argument = '-m "' + self.sample_graph_deployment_id + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\']"' + argument = ['-m', self.sample_graph_deployment_id, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\']"] output = self.run_script(argument) self.assertNotIn('deployment_id', output.strip()) def test_minigraph_ethernet_interfaces(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "PORT[\'Ethernet8\']"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "PORT[\'Ethernet8\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), utils.to_dict("{'lanes': '37,38,39,40', 'description': 'Interface description', 'pfc_asym': 'off', 'mtu': '9100', 'alias': 'fortyGigE0/8', 'admin_status': 'up', 'speed': '1000', 'tpid': '0x8100'}") ) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "PORT[\'Ethernet12\']"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "PORT[\'Ethernet12\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -531,7 +529,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_neighbor_interfaces(self): - argument = '-m "' + self.sample_graph_simple_case + '" -p "' + self.port_config + '" -v "PORT"' + argument = ['-m', self.sample_graph_simple_case, '-p', self.port_config, '-v', "PORT"] output = self.run_script(argument) self.assertEqual( @@ -574,7 +572,7 @@ class TestCfgGen(TestCase): def test_minigraph_neighbor_interfaces_config_db(self): # test to check if PORT table is retrieved from config_db - argument = '-m "' + self.sample_graph_simple_case + '" -p "' + self.port_config + '" -v "PORT"' + argument = ['-m', self.sample_graph_simple_case, '-p', self.port_config, '-v', "PORT"] output = self.run_script(argument) self.assertEqual( @@ -617,7 +615,7 @@ class TestCfgGen(TestCase): def test_minigraph_extra_ethernet_interfaces(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "PORT"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "PORT"] output = self.run_script(argument) self.assertEqual( @@ -665,7 +663,7 @@ class TestCfgGen(TestCase): # self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '10.0.100.1'}}") def test_metadata_tacacs(self): - argument = '-m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '" -v "TACPLUS_SERVER"' + argument = ['-m', self.sample_graph_metadata, '-p', self.port_config, '-v', "TACPLUS_SERVER"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -673,24 +671,24 @@ class TestCfgGen(TestCase): ) def test_metadata_ntp(self): - argument = '-m "' + self.sample_graph_metadata + '" -p "' + self.port_config + '" -v "NTP_SERVER"' + argument = ['-m', self.sample_graph_metadata, '-p', self.port_config, '-v', "NTP_SERVER"] output = self.run_script(argument) self.assertEqual(utils.to_dict(output.strip()), utils.to_dict("{'10.0.10.1': {}, '10.0.10.2': {}}")) def test_minigraph_vnet(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "VNET"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "VNET"] output = self.run_script(argument) self.assertEqual(output.strip(), "") def test_minigraph_vxlan(self, **kwargs): graph_file = kwargs.get('graph_file', self.sample_graph_simple) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "VXLAN_TUNNEL"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "VXLAN_TUNNEL"] output = self.run_script(argument) self.assertEqual(output.strip(), "") def test_minigraph_bgp_mon(self): - argument = '-m "' + self.sample_graph_simple + '" -p "' + self.port_config + '" -v "BGP_MONITORS"' + argument = ['-m', self.sample_graph_simple, '-p', self.port_config, '-v', "BGP_MONITORS"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -698,7 +696,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_bgp_voq_chassis_peer(self): - argument = '-m "' + self.sample_graph_simple + '" -p "' + self.port_config + '" -v "BGP_VOQ_CHASSIS_NEIGHBOR[\'10.2.0.21\']"' + argument = ['-m', self.sample_graph_simple, '-p', self.port_config, '-v', "BGP_VOQ_CHASSIS_NEIGHBOR[\'10.2.0.21\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -706,7 +704,7 @@ class TestCfgGen(TestCase): ) # make sure VoQChassisInternal value of false is honored - argument = '-m "' + self.sample_graph_simple + '" -p "' + self.port_config + '" -v "BGP_VOQ_CHASSIS_NEIGHBOR[\'10.0.0.57\']"' + argument = ['-m', self.sample_graph_simple, '-p', self.port_config, '-v', "BGP_VOQ_CHASSIS_NEIGHBOR[\'10.0.0.57\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "") @@ -726,14 +724,14 @@ class TestCfgGen(TestCase): try: print('\n Change device type to %s' % (BACKEND_LEAF_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_LEAF_ROUTER, self.sample_backend_graph), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_LEAF_ROUTER), self.sample_backend_graph], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_LEAF_ROUTER, self.sample_backend_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_LEAF_ROUTER), self.sample_backend_graph]) self.test_jinja_expression(self.sample_backend_graph, self.port_config, BACKEND_LEAF_ROUTER) # ACL_TABLE should contain EVERFLOW related entries - argument = '-m "' + self.sample_backend_graph + '" -p "' + self.port_config + '" -v "ACL_TABLE"' + argument = ['-m', self.sample_backend_graph, '-p', self.port_config, '-v', "ACL_TABLE"] output = self.run_script(argument) sample_output = utils.to_dict(output.strip()).keys() assert 'DATAACL' not in sample_output, sample_output @@ -742,9 +740,9 @@ class TestCfgGen(TestCase): finally: print('\n Change device type back to %s' % (TOR_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, TOR_ROUTER, self.sample_backend_graph), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, TOR_ROUTER), self.sample_backend_graph], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, TOR_ROUTER, self.sample_backend_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, TOR_ROUTER), self.sample_backend_graph]) self.test_jinja_expression(self.sample_backend_graph, self.port_config, TOR_ROUTER) @@ -752,23 +750,23 @@ class TestCfgGen(TestCase): try: print('\n Change device type to %s' % (BACKEND_LEAF_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (LEAF_ROUTER, BACKEND_LEAF_ROUTER, self.sample_graph), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (LEAF_ROUTER, BACKEND_LEAF_ROUTER), self.sample_graph], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (LEAF_ROUTER, BACKEND_LEAF_ROUTER, self.sample_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (LEAF_ROUTER, BACKEND_LEAF_ROUTER), self.sample_graph]) self.test_jinja_expression(self.sample_graph, self.port_config, BACKEND_LEAF_ROUTER) self.verify_no_vlan_member() finally: print('\n Change device type back to %s' % (LEAF_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, LEAF_ROUTER, self.sample_graph), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, LEAF_ROUTER), self.sample_graph], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, LEAF_ROUTER, self.sample_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, LEAF_ROUTER), self.sample_graph]) self.test_jinja_expression(self.sample_graph, self.port_config, LEAF_ROUTER) def verify_no_vlan_member(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VLAN_MEMBER"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "VLAN_MEMBER"] output = self.run_script(argument) self.assertEqual(output.strip(), "{}") @@ -786,33 +784,33 @@ class TestCfgGen(TestCase): try: print('\n Change device type to %s' % (BACKEND_TOR_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file]) self.test_jinja_expression(graph_file, self.port_config, BACKEND_TOR_ROUTER) # INTERFACE table does not exist - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "INTERFACE"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "INTERFACE"] output = self.run_script(argument) self.assertEqual(output.strip(), "") # PORTCHANNEL_INTERFACE table does not exist - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "PORTCHANNEL_INTERFACE"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "PORTCHANNEL_INTERFACE"] output = self.run_script(argument) self.assertEqual(output.strip(), "") # SLB and BGP Monitor table does not exist - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "BGP_PEER_RANGE"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "BGP_PEER_RANGE"] output = self.run_script(argument) self.assertEqual(output.strip(), "{}") - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "BGP_MONITORS"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "BGP_MONITORS"] output = self.run_script(argument) self.assertEqual(output.strip(), "{}") # ACL_TABLE should not contain EVERFLOW related entries - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "ACL_TABLE"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "ACL_TABLE"] output = self.run_script(argument) sample_output = utils.to_dict(output.strip()).keys() assert 'DATAACL' in sample_output, sample_output @@ -828,7 +826,7 @@ class TestCfgGen(TestCase): self.test_minigraph_vxlan(graph_file=graph_file) # VLAN_SUB_INTERFACE - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v VLAN_SUB_INTERFACE' + argument = ['-m', graph_file, '-p', self.port_config, '-v', 'VLAN_SUB_INTERFACE'] output = self.run_script(argument) print(output.strip()) # not a usecase to parse SubInterfaces under PortChannel @@ -861,24 +859,24 @@ class TestCfgGen(TestCase): finally: print('\n Change device type back to %s' % (TOR_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file]) self.test_jinja_expression(graph_file, self.port_config, TOR_ROUTER) def test_show_run_acl(self): - argument = '-a \'{"key1":"value"}\' --var-json ACL_RULE' + argument = ['-a', '{"key1":"value"}', '--var-json', 'ACL_RULE'] output = self.run_script(argument) self.assertEqual(output, '') def test_show_run_interfaces(self): - argument = '-a \'{"key1":"value"}\' --var-json INTERFACE' + argument = ['-a', '{"key1":"value"}', '--var-json', 'INTERFACE'] output = self.run_script(argument) self.assertEqual(output, '') def test_minigraph_voq_metadata(self): - argument = "-j {} -m {} -p {} --var-json DEVICE_METADATA".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "DEVICE_METADATA"] output = json.loads(self.run_script(argument)) self.assertEqual(output['localhost']['asic_name'], 'Asic0') self.assertEqual(output['localhost']['switch_id'], '0') @@ -886,7 +884,7 @@ class TestCfgGen(TestCase): self.assertEqual(output['localhost']['max_cores'], '16') def test_minigraph_voq_system_ports(self): - argument = "-j {} -m {} -p {} --var-json SYSTEM_PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "SYSTEM_PORT"] self.assertDictEqual( json.loads(self.run_script(argument)), { @@ -905,7 +903,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_voq_port_macsec_enabled(self): - argument = '-j "' + self.macsec_profile + '" -m "' + self.sample_graph_voq + '" -p "' + self.voq_port_config + '" -v "PORT[\'Ethernet0\']"' + argument = ['-j', self.macsec_profile, '-m', self.sample_graph_voq, '-p', self.voq_port_config, '-v', "PORT[\'Ethernet0\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -913,7 +911,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_voq_inband_interface_port(self): - argument = "-j {} -m {} -p {} --var-json VOQ_INBAND_INTERFACE".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "VOQ_INBAND_INTERFACE"] output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -926,7 +924,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_voq_inband_port(self): - argument = "-j {} -m {} -p {} --var-json PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "PORT"] output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -944,7 +942,7 @@ class TestCfgGen(TestCase): }) def test_minigraph_voq_recirc_ports(self): - argument = "-j {} -m {} -p {} --var-json PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "PORT"] output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -962,7 +960,7 @@ class TestCfgGen(TestCase): }) def test_minigraph_dhcp(self): - argument = '-m "' + self.sample_graph_simple_case + '" -p "' + self.port_config + '" -v DHCP_RELAY' + argument = ['-m', self.sample_graph_simple_case, '-p', self.port_config, '-v', 'DHCP_RELAY'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -973,7 +971,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_bgp_packet_chassis_peer(self): - argument = '-m "' + self.packet_chassis_graph + '" -p "' + self.packet_chassis_port_ini + '" -n "' + "asic1" + '" -v "BGP_INTERNAL_NEIGHBOR[\'8.0.0.1\']"' + argument = ['-m', self.packet_chassis_graph, '-p', self.packet_chassis_port_ini, '-n', "asic1", '-v', "BGP_INTERNAL_NEIGHBOR[\'8.0.0.1\']"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -981,14 +979,14 @@ class TestCfgGen(TestCase): ) def test_minigraph_bgp_packet_chassis_static_route(self): - argument = '-m "' + self.packet_chassis_graph + '" -p "' + self.packet_chassis_port_ini + '" -v "STATIC_ROUTE"' + argument = ['-m', self.packet_chassis_graph, '-p', self.packet_chassis_port_ini, '-v', "STATIC_ROUTE"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), utils.to_dict("{'8.0.0.1/32': {'nexthop': '192.168.1.2,192.168.2.2', 'ifname': 'PortChannel40,PortChannel50', 'advertise':'false'}}") ) - argument = '-m "' + self.packet_chassis_graph + '" -p "' + self.packet_chassis_port_ini + '" -n "' + "asic1" + '" -v "STATIC_ROUTE"' + argument = ['-m', self.packet_chassis_graph, '-p', self.packet_chassis_port_ini, '-n', "asic1", '-v', "STATIC_ROUTE"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -996,7 +994,7 @@ class TestCfgGen(TestCase): ) def test_minigraph_bgp_packet_chassis_vlan_subintf(self): - argument = '-m "' + self.packet_chassis_graph + '" -p "' + self.packet_chassis_port_ini + '" -n "' + "asic1" + '" -v "VLAN_SUB_INTERFACE"' + argument = ['-m', self.packet_chassis_graph, '-p', self.packet_chassis_port_ini, '-n', "asic1", '-v', "VLAN_SUB_INTERFACE"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -1004,14 +1002,14 @@ class TestCfgGen(TestCase): ) def test_minigraph_voq_400g_zr_port_config(self): - argument = "-j {} -m {} -p {} -v \"PORT[\'Ethernet4\']\"".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) + argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "-v" "PORT[\'Ethernet4\']"] output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertEqual(output_dict['tx_power'], '-10') self.assertEqual(output_dict['laser_freq'], 195875) def test_minigraph_packet_chassis_400g_zr_port_config(self): - argument = "-m {} -p {} -n asic1 -v \"PORT[\'Ethernet13\']\"".format(self.packet_chassis_graph, self.packet_chassis_port_ini) + argument = ["-m", self.packet_chassis_graph, "-p", self.packet_chassis_port_ini, "-n", "asic1", "-v", "PORT[\'Ethernet13\']"] output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertEqual(output_dict['tx_power'], '7.5') diff --git a/src/sonic-config-engine/tests/test_cfggen_from_yang.py b/src/sonic-config-engine/tests/test_cfggen_from_yang.py index 801340ea47..ddc77c8b03 100644 --- a/src/sonic-config-engine/tests/test_cfggen_from_yang.py +++ b/src/sonic-config-engine/tests/test_cfggen_from_yang.py @@ -4,6 +4,7 @@ import subprocess 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 @@ class TestCfgGen(object): 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 @@ class TestCfgGen(object): 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 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): '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 @@ class TestCfgGen(object): }) 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,7 +148,7 @@ class TestCfgGen(object): }) 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": {}, @@ -158,7 +156,7 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): '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 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): }) 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 @@ class TestCfgGen(object): }) 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": { diff --git a/src/sonic-config-engine/tests/test_cfggen_pfx_filter.py b/src/sonic-config-engine/tests/test_cfggen_pfx_filter.py index 1ac2b7f7f5..b3cad3aa21 100644 --- a/src/sonic-config-engine/tests/test_cfggen_pfx_filter.py +++ b/src/sonic-config-engine/tests/test_cfggen_pfx_filter.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) diff --git a/src/sonic-config-engine/tests/test_cfggen_platformJson.py b/src/sonic-config-engine/tests/test_cfggen_platformJson.py index 0af361718b..5d39fd2f36 100644 --- a/src/sonic-config-engine/tests/test_cfggen_platformJson.py +++ b/src/sonic-config-engine/tests/test_cfggen_platformJson.py @@ -3,7 +3,7 @@ import json 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 @@ class TestCfgGenPlatformJson(TestCase): 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 @@ class TestCfgGenPlatformJson(TestCase): '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 @@ class TestCfgGenPlatformJson(TestCase): # 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 diff --git a/src/sonic-config-engine/tests/test_cfggen_t2_chassis_fe.py b/src/sonic-config-engine/tests/test_cfggen_t2_chassis_fe.py index a3d6d02a7f..9bf3fc8a3e 100644 --- a/src/sonic-config-engine/tests/test_cfggen_t2_chassis_fe.py +++ b/src/sonic-config-engine/tests/test_cfggen_t2_chassis_fe.py @@ -1,6 +1,5 @@ import os import subprocess - import tests.common_utils as utils from unittest import TestCase @@ -10,18 +9,18 @@ class TestCfgGenT2ChassisFe(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.sample_graph_t2_chassis_fe = os.path.join(self.test_dir, 't2-chassis-fe-graph.xml') self.sample_graph_t2_chassis_fe_vni = os.path.join(self.test_dir, 't2-chassis-fe-graph-vni.xml') self.sample_graph_t2_chassis_fe_pc = os.path.join(self.test_dir, 't2-chassis-fe-graph-pc.xml') self.t2_chassis_fe_port_config = os.path.join(self.test_dir, 't2-chassis-fe-port-config.ini') def run_script(self, argument, check_stderr=False): - print('\n Running sonic-cfggen ' + argument) + print('\n Running sonic-cfggen ' + ' '.join(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() @@ -34,12 +33,12 @@ class TestCfgGenT2ChassisFe(TestCase): return output def test_minigraph_t2_chassis_fe_type(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe + '" -p "' + self.t2_chassis_fe_port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'type\']"' + argument = ['-m', self.sample_graph_t2_chassis_fe, '-p', self.t2_chassis_fe_port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'type\']"] output = self.run_script(argument) self.assertEqual(output.strip(), 'SpineChassisFrontendRouter') def test_minigraph_t2_chassis_fe_interfaces(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe + '" -p "' + self.t2_chassis_fe_port_config + '" -v "INTERFACE"' + argument = ['-m', self.sample_graph_t2_chassis_fe, '-p', self.t2_chassis_fe_port_config, '-v', "INTERFACE"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -53,7 +52,7 @@ class TestCfgGenT2ChassisFe(TestCase): ) ) def test_minigraph_t2_chassis_fe_pc_interfaces(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe_pc + '" -p "' + self.t2_chassis_fe_port_config + '" -v "PORTCHANNEL_INTERFACE"' + argument = ['-m', self.sample_graph_t2_chassis_fe_pc, '-p', self.t2_chassis_fe_port_config, '-v', "PORTCHANNEL_INTERFACE"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -70,17 +69,17 @@ class TestCfgGenT2ChassisFe(TestCase): # Test a minigraph file where VNI is not specified # Default VNI is 8000 def test_minigraph_t2_chassis_fe_vnet_default(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe + '" -p "' + self.t2_chassis_fe_port_config + '" -v "VNET"' + argument = ['-m', self.sample_graph_t2_chassis_fe, '-p', self.t2_chassis_fe_port_config, '-v', "VNET"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'VnetFE': {'vxlan_tunnel': 'TunnelInt', 'vni': 8000}}") # Test a minigraph file where VNI is specified def test_minigraph_t2_chassis_fe_vnet(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe_vni + '" -p "' + self.t2_chassis_fe_port_config + '" -v "VNET"' + argument = ['-m', self.sample_graph_t2_chassis_fe_vni, '-p', self.t2_chassis_fe_port_config, '-v', "VNET"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'VnetFE': {'vxlan_tunnel': 'TunnelInt', 'vni': 9000}}") def test_minigraph_t2_chassis_fe_vxlan(self): - argument = '-m "' + self.sample_graph_t2_chassis_fe + '" -p "' + self.t2_chassis_fe_port_config + '" -v "VXLAN_TUNNEL"' + argument = ['-m', self.sample_graph_t2_chassis_fe, '-p', self.t2_chassis_fe_port_config, '-v', "VXLAN_TUNNEL"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'TunnelInt': {'src_ip': '4.0.0.0'}}") diff --git a/src/sonic-config-engine/tests/test_frr.py b/src/sonic-config-engine/tests/test_frr.py index 3934f8c7d7..8f31b4ac79 100644 --- a/src/sonic-config-engine/tests/test_frr.py +++ b/src/sonic-config-engine/tests/test_frr.py @@ -3,13 +3,13 @@ import os import subprocess import tests.common_utils as utils - +from sonic_py_common.general import getstatusoutput_noshell from unittest import TestCase class TestCfgGen(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.t0_minigraph = os.path.join(self.test_dir, 't0-sample-graph.xml') self.t0_port_config = os.path.join(self.test_dir, 't0-sample-port-config.ini') self.output_file = os.path.join(self.test_dir, 'output') @@ -21,15 +21,19 @@ class TestCfgGen(TestCase): pass - def run_script(self, argument, check_stderr=False): + def run_script(self, argument, check_stderr=False, output_file=None): # 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() + if output_file: + with open(output_file, 'w') as f: + f.write(output) linecount = output.strip().count('\n') if linecount <= 0: @@ -39,20 +43,15 @@ class TestCfgGen(TestCase): return output def run_diff(self, file1, file2): - output = subprocess.check_output('diff -u {} {} || true'.format(file1, file2), shell=True) - - if utils.PY3x: - output = output.decode() - + _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) return output def run_case(self, template, target): template_dir = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-fpm-frr', "frr") conf_template = os.path.join(template_dir, template) constants = os.path.join(self.test_dir, '..', '..', '..', 'files', 'image_config', 'constants', 'constants.yml') - cmd_args = self.t0_minigraph, self.t0_port_config, constants, conf_template, template_dir, self.output_file - cmd = "-m %s -p %s -y %s -t %s -T %s > %s" % cmd_args - self.run_script(cmd) + cmd = ['-m', self.t0_minigraph, '-p', self.t0_port_config, '-y', constants, '-t', conf_template, '-T', template_dir] + self.run_script(cmd, output_file=self.output_file) original_filename = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, target) r = filecmp.cmp(original_filename, self.output_file) @@ -69,4 +68,3 @@ class TestCfgGen(TestCase): def test_zebra_frr(self): self.assertTrue(*self.run_case('zebra/zebra.conf.j2', 'zebra_frr.conf')) - diff --git a/src/sonic-config-engine/tests/test_j2files.py b/src/sonic-config-engine/tests/test_j2files.py index 3ac219468e..40af24151c 100644 --- a/src/sonic-config-engine/tests/test_j2files.py +++ b/src/sonic-config-engine/tests/test_j2files.py @@ -5,12 +5,13 @@ import subprocess from unittest import TestCase import tests.common_utils as utils +from sonic_py_common.general import getstatusoutput_noshell, getstatusoutput_noshell_pipe class TestJ2Files(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.simple_minigraph = os.path.join(self.test_dir, 'simple-sample-graph.xml') self.port_data = os.path.join(self.test_dir, 'sample-port-data.json') self.ztp = os.path.join(self.test_dir, "sample-ztp.json") @@ -40,17 +41,21 @@ class TestJ2Files(TestCase): self.output_file = os.path.join(self.test_dir, 'output') os.environ["CFGGEN_UNIT_TESTING"] = "2" - def run_script(self, argument): - print('CMD: sonic-cfggen ' + argument) - output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) + def run_script(self, argument, output_file=None): + print('CMD: sonic-cfggen ', argument) + output = subprocess.check_output(self.script_file + argument) if utils.PY3x: output = output.decode() + if output_file: + with open(output_file, 'w') as f: + f.write(output) 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 create_machine_conf(self, platform, vendor): file_exist = True @@ -59,23 +64,24 @@ class TestJ2Files(TestCase): 'dell': 'onie', 'mellanox': 'onie' } - echo_cmd = "echo '{}_platform={}' | sudo tee -a /host/machine.conf > /dev/null".format(mode[vendor], platform) + echo_cmd1 = ["echo", '{}_platform={}'.format(mode[vendor], platform)] + echo_cmd2 = ["sudo", "tee", "-a", "/host/machine.conf"] if not os.path.exists('/host/machine.conf'): file_exist = False if not os.path.isdir('/host'): dir_exist = False - os.system('sudo mkdir /host') - os.system('sudo touch /host/machine.conf') - os.system(echo_cmd) + subprocess.call(['sudo', 'mkdir', '/host']) + subprocess.call(['sudo', 'touch', '/host/machine.conf']) + getstatusoutput_noshell_pipe(echo_cmd1, echo_cmd2) return file_exist, dir_exist def remove_machine_conf(self, file_exist, dir_exist): if not file_exist: - os.system('sudo rm -f /host/machine.conf') + subprocess.call(['sudo', 'rm', '-f', '/host/machine.conf']) if not dir_exist: - os.system('sudo rmdir /host') + subprocess.call(['sudo', 'rmdir', '/host']) def modify_cable_len(self, base_file, file_dir): input_file = os.path.join(file_dir, base_file) @@ -95,76 +101,76 @@ class TestJ2Files(TestCase): interfaces_template = os.path.join(self.test_dir, '..', '..', '..', 'files', 'image_config', 'interfaces', 'interfaces.j2') # ZTP enabled - argument = '-m ' + self.t0_minigraph_nomgmt + ' -p ' + self.t0_port_config_tiny + ' -j ' + self.ztp + ' -j ' + self.port_data + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + '> ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_nomgmt, '-p', self.t0_port_config_tiny, '-j', self.ztp, '-j', self.port_data, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces_nomgmt_ztp'), self.output_file)) - argument = '-m ' + self.t0_minigraph_nomgmt + ' -p ' + self.t0_port_config_tiny + ' -j ' + self.ztp_inband + ' -j ' + self.port_data + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + '> ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_nomgmt, '-p', self.t0_port_config_tiny, '-j', self.ztp_inband, '-j', self.port_data, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces_nomgmt_ztp_inband'), self.output_file)) - argument = '-m ' + self.t0_minigraph_nomgmt + ' -p ' + self.t0_port_config_tiny + ' -j ' + self.ztp_ip + ' -j ' + self.port_data + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + '> ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_nomgmt, '-p', self.t0_port_config_tiny, '-j', self.ztp_ip, '-j', self.port_data, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces_nomgmt_ztp_ip'), self.output_file)) - argument = '-m ' + self.t0_minigraph_nomgmt + ' -p ' + self.t0_port_config_tiny + ' -j ' + self.ztp_inband_ip + ' -j ' + self.port_data + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + '> ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_nomgmt, '-p', self.t0_port_config_tiny, '-j', self.ztp_inband_ip, '-j', self.port_data, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces_nomgmt_ztp_inband_ip'), self.output_file)) # ZTP disabled, MGMT_INTERFACE defined - argument = '-m ' + self.t0_minigraph + ' -p ' + self.t0_port_config + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph, '-p', self.t0_port_config, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces'), self.output_file)) - argument = '-m ' + self.t0_mvrf_minigraph + ' -p ' + self.t0_port_config + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_mvrf_minigraph, '-p', self.t0_port_config, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'mvrf_interfaces'), self.output_file)) - argument = '-m ' + self.t0_minigraph_two_mgmt + ' -p ' + self.t0_port_config + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_two_mgmt, '-p', self.t0_port_config, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'two_mgmt_interfaces'), self.output_file), self.output_file) # ZTP disabled, no MGMT_INTERFACE defined - argument = '-m ' + self.t0_minigraph_nomgmt + ' -p ' + self.t0_port_config + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph_nomgmt, '-p', self.t0_port_config, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'interfaces_nomgmt'), self.output_file)) - argument = '-m ' + self.t0_mvrf_minigraph_nomgmt + ' -p ' + self.t0_port_config + ' -a \'{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}\' -t ' + interfaces_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_mvrf_minigraph_nomgmt, '-p', self.t0_port_config, '-a', '{\"hwaddr\":\"e4:1d:2d:a5:f3:ad\"}', '-t', interfaces_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'mvrf_interfaces_nomgmt'), self.output_file)) def test_ports_json(self): ports_template = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-orchagent', 'ports.json.j2') - argument = '-m ' + self.simple_minigraph + ' -p ' + self.t0_port_config + ' -t ' + ports_template + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.simple_minigraph, '-p', self.t0_port_config, '-t', ports_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'ports.json'), self.output_file)) def test_dhcp_relay(self): # Test generation of wait_for_intf.sh dhc_sample_data = os.path.join(self.test_dir, "dhcp-relay-sample.json") template_path = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-dhcp-relay', 'wait_for_intf.sh.j2') - argument = '-m ' + self.t0_minigraph + ' -j ' + dhc_sample_data + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph, '-j', dhc_sample_data, '-p', self.t0_port_config, '-t', template_path] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'wait_for_intf.sh'), self.output_file)) # Test generation of docker-dhcp-relay.supervisord.conf template_path = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-dhcp-relay', 'docker-dhcp-relay.supervisord.conf.j2') - argument = '-m ' + self.t0_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph, '-p', self.t0_port_config, '-t', template_path] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'docker-dhcp-relay.supervisord.conf'), self.output_file)) # Test generation of docker-dhcp-relay.supervisord.conf when a vlan is missing ip/ipv6 helpers template_path = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-dhcp-relay', 'docker-dhcp-relay.supervisord.conf.j2') - argument = '-m ' + self.no_ip_helper_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.no_ip_helper_minigraph, '-p', self.t0_port_config, '-t', template_path] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'docker-dhcp-relay-no-ip-helper.supervisord.conf'), self.output_file)) def test_radv(self): # Test generation of radvd.conf with multiple ipv6 prefixes template_path = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-router-advertiser', 'radvd.conf.j2') - argument = '-m ' + self.radv_test_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.radv_test_minigraph, '-p', self.t0_port_config, '-t', template_path] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'radvd.conf'), self.output_file)) def test_lldp(self): @@ -176,32 +182,32 @@ class TestJ2Files(TestCase): # Test generation of lldpd.conf if IPv4 and IPv6 management interfaces exist mgmt_iface_ipv4_and_ipv6_json = os.path.join(self.test_dir, "data", "lldp", "mgmt_iface_ipv4_and_ipv6.json") - argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv4_and_ipv6_json, lldpd_conf_template, self.output_file) - self.run_script(argument) + argument = ['-j', mgmt_iface_ipv4_and_ipv6_json, '-t', lldpd_conf_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(expected_mgmt_ipv4_and_ipv6, self.output_file)) # Test generation of lldpd.conf if management interface IPv4 only exist mgmt_iface_ipv4_json = os.path.join(self.test_dir, "data", "lldp", "mgmt_iface_ipv4.json") - argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv4_json, lldpd_conf_template, self.output_file) - self.run_script(argument) + argument = ['-j', mgmt_iface_ipv4_json, '-t', lldpd_conf_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(expected_mgmt_ipv4, self.output_file)) # Test generation of lldpd.conf if Management interface IPv6 only exist mgmt_iface_ipv6_json = os.path.join(self.test_dir, "data", "lldp", "mgmt_iface_ipv6.json") - argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv6_json, lldpd_conf_template, self.output_file) - self.run_script(argument) + argument = ['-j', mgmt_iface_ipv6_json, '-t', lldpd_conf_template] + self.run_script(argument, output_file=self.output_file) self.assertTrue(utils.cmp(expected_mgmt_ipv6, self.output_file)) def test_ipinip(self): ipinip_file = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-orchagent', 'ipinip.json.j2') - argument = '-m ' + self.t0_minigraph + ' -p ' + self.t0_port_config + ' -t ' + ipinip_file + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.t0_minigraph, '-p', self.t0_port_config, '-t', ipinip_file] + self.run_script(argument, output_file=self.output_file) sample_output_file = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'ipinip.json') assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) def test_l2switch_template(self): - argument = '-k Mellanox-SN2700 --preset l2 -p ' + self.t0_port_config + argument = ['-k', 'Mellanox-SN2700', '--preset', 'l2', '-p', self.t0_port_config] output = self.run_script(argument) output_json = json.loads(output) @@ -212,14 +218,14 @@ class TestJ2Files(TestCase): self.assertTrue(json.dumps(sample_output_json, sort_keys=True) == json.dumps(output_json, sort_keys=True)) template_dir = os.path.join(self.test_dir, '..', 'data', 'l2switch.j2') - argument = '-t ' + template_dir + ' -k Mellanox-SN2700 -p ' + self.t0_port_config + argument = ['-t', template_dir, '-k', 'Mellanox-SN2700', '-p', self.t0_port_config] output = self.run_script(argument) output_json = json.loads(output) self.assertTrue(json.dumps(sample_output_json, sort_keys=True) == json.dumps(output_json, sort_keys=True)) def test_l1_ports_template(self): - argument = '-k 32x1000Gb --preset l1 -p ' + self.l1_l3_port_config + argument = ['-k', '32x1000Gb', '--preset', 'l1', '-p', self.l1_l3_port_config] output = self.run_script(argument) output_json = json.loads(output) @@ -230,14 +236,14 @@ class TestJ2Files(TestCase): self.assertTrue(json.dumps(sample_output_json, sort_keys=True) == json.dumps(output_json, sort_keys=True)) template_dir = os.path.join(self.test_dir, '..', 'data', 'l1intf.j2') - argument = '-t ' + template_dir + ' -k 32x1000Gb -p ' + self.l1_l3_port_config + argument = ['-t', template_dir, '-k', '32x1000Gb', '-p', self.l1_l3_port_config] output = self.run_script(argument) output_json = json.loads(output) self.assertTrue(json.dumps(sample_output_json, sort_keys=True) == json.dumps(output_json, sort_keys=True)) def test_l3_ports_template(self): - argument = '-k 32x1000Gb --preset l3 -p ' + self.l1_l3_port_config + argument = ['-k', '32x1000Gb', '--preset', 'l3', '-p', self.l1_l3_port_config] output = self.run_script(argument) output_json = json.loads(output) @@ -248,7 +254,7 @@ class TestJ2Files(TestCase): self.assertTrue(json.dumps(sample_output_json, sort_keys=True) == json.dumps(output_json, sort_keys=True)) template_dir = os.path.join(self.test_dir, '..', 'data', 'l3intf.j2') - argument = '-t ' + template_dir + ' -k 32x1000Gb -p ' + self.l1_l3_port_config + argument = ['-t', template_dir, '-k', '32x1000Gb', '-p', self.l1_l3_port_config] output = self.run_script(argument) output_json = json.loads(output) @@ -270,9 +276,7 @@ class TestJ2Files(TestCase): "Ethernet112", "Ethernet116", "Ethernet120", "Ethernet124" ] } - argument = '-a \'{}\' -k Arista-7050CX3-32S-D48C8 --preset l2 -p {}'.format( - json.dumps(extra_args), self.t0_7050cx3_port_config - ) + argument = ['-a', json.dumps(extra_args), '-k', 'Arista-7050CX3-32S-D48C8', '--preset', 'l2', '-p', self.t0_7050cx3_port_config] output = self.run_script(argument) output_json = json.loads(output) @@ -299,8 +303,8 @@ class TestJ2Files(TestCase): for template_file, cfg_file, sample_output_file in [(qos_file, 'qos_config.j2', 'qos-arista7800r3-48cq2-lc.json'), (buffer_file, 'buffers_config.j2', 'buffer-arista7800r3-48cq2-lc.json') ]: - argument = '-m ' + self.arista7800r3_48cq2_lc_t2_minigraph + ' -p ' + port_config_ini_file + ' -t ' + template_file + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', self.arista7800r3_48cq2_lc_t2_minigraph, '-p', port_config_ini_file, '-t', template_file] + self.run_script(argument, output_file=self.output_file) # cleanup cfg_file_new = os.path.join(arista_dir_path, cfg_file) @@ -335,8 +339,8 @@ class TestJ2Files(TestCase): shutil.copy2(qos_config_file, dir_path) minigraph = os.path.join(self.test_dir, minigraph) - argument = '-m ' + minigraph + ' -p ' + port_config_ini_file + ' -t ' + qos_file + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', minigraph, '-p', port_config_ini_file, '-t', qos_file] + self.run_script(argument, output_file=self.output_file) # cleanup qos_config_file_new = os.path.join(dir_path, 'qos_config.j2') @@ -398,8 +402,8 @@ class TestJ2Files(TestCase): qos_config_file = os.path.join(self.test_dir, '..', '..', '..', 'files', 'build_templates', 'qos_config.j2') shutil.copy2(qos_config_file, device_template_path) - argument = '-m ' + sample_minigraph_file + ' -p ' + port_config_ini_file + ' -t ' + qos_file + ' > ' + test_output - self.run_script(argument) + argument = ['-m', sample_minigraph_file, '-p', port_config_ini_file, '-t', qos_file] + self.run_script(argument, output_file=test_output) # cleanup qos_config_file_new = os.path.join(device_template_path, 'qos_config.j2') @@ -432,8 +436,8 @@ class TestJ2Files(TestCase): config_bcm_file = os.path.join(device_template_path, 'config.bcm.j2') config_test_output = os.path.join(self.test_dir, 'config_output.bcm') - argument = '-m ' + sample_minigraph_file + ' -p ' + port_config_ini_file + ' -t ' + config_bcm_file + ' > ' + config_test_output - self.run_script(argument) + argument = ['-m', sample_minigraph_file, '-p', port_config_ini_file, '-t', config_bcm_file] + self.run_script(argument, output_file=config_test_output) #check output config.bcm config_sample_output_file = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, config_sample_output) @@ -451,8 +455,8 @@ class TestJ2Files(TestCase): shutil.copy2(buffers_config_file, dir_path) minigraph = os.path.join(self.test_dir, minigraph) - argument = '-m ' + minigraph + ' -p ' + port_config_ini_file + ' -t ' + buffers_file + ' > ' + self.output_file - self.run_script(argument) + argument = ['-m', minigraph, '-p', port_config_ini_file, '-t', buffers_file] + self.run_script(argument, output_file=self.output_file) # cleanup buffers_config_file_new = os.path.join(dir_path, 'buffers_config.j2') @@ -516,9 +520,9 @@ class TestJ2Files(TestCase): def test_ipinip_multi_asic(self): ipinip_file = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-orchagent', 'ipinip.json.j2') - argument = '-m ' + self.multi_asic_minigraph + ' -p ' + self.multi_asic_port_config + ' -t ' + ipinip_file + ' -n asic0 ' + ' > ' + self.output_file + argument = ['-m', self.multi_asic_minigraph, '-p', self.multi_asic_port_config, '-t', ipinip_file, '-n', 'asic0'] print(argument) - self.run_script(argument) + self.run_script(argument, output_file=self.output_file) sample_output_file = os.path.join(self.test_dir, 'multi_npu_data', utils.PYvX_DIR, 'ipinip.json') assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) @@ -544,13 +548,11 @@ class TestJ2Files(TestCase): }, } for _, v in test_list.items(): - argument = " -m {} -p {} -y {} -t {} > {}".format( - v["graph"], v["port_config"], constants_yml, switch_template, self.output_file - ) + argument = ["-m", v["graph"], "-p", v["port_config"], "-y", constants_yml, "-t", switch_template] sample_output_file = os.path.join( self.test_dir, 'sample_output', v["output"] ) - self.run_script(argument) + self.run_script(argument, output_file=self.output_file) assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) def test_swss_switch_render_template_multi_asic(self): @@ -575,14 +577,11 @@ class TestJ2Files(TestCase): } for _, v in test_list.items(): os.environ["NAMESPACE_ID"] = v["namespace_id"] - argument = " -m {} -y {} -t {} > {}".format( - self.t1_mlnx_minigraph, constants_yml, switch_template, - self.output_file - ) + argument = ["-m", self.t1_mlnx_minigraph, "-y", constants_yml, "-t", switch_template] sample_output_file = os.path.join( self.test_dir, 'sample_output', v["output"] ) - self.run_script(argument) + self.run_script(argument, output_file=self.output_file) assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) os.environ["NAMESPACE_ID"] = "" @@ -591,8 +590,8 @@ class TestJ2Files(TestCase): vlan_interfaces_json = os.path.join(self.test_dir, "data", "ndppd", "vlan_interfaces.json") expected = os.path.join(self.test_dir, "sample_output", utils.PYvX_DIR, "ndppd.conf") - argument = '-j {} -t {} > {}'.format(vlan_interfaces_json, conf_template, self.output_file) - self.run_script(argument) + argument = ['-j', vlan_interfaces_json, '-t', conf_template] + self.run_script(argument, output_file=self.output_file) assert utils.cmp(expected, self.output_file), self.run_diff(expected, self.output_file) def test_ntp_conf(self): @@ -600,8 +599,8 @@ class TestJ2Files(TestCase): ntp_interfaces_json = os.path.join(self.test_dir, "data", "ntp", "ntp_interfaces.json") expected = os.path.join(self.test_dir, "sample_output", utils.PYvX_DIR, "ntp.conf") - argument = '-j {} -t {} > {}'.format(ntp_interfaces_json, conf_template, self.output_file) - self.run_script(argument) + argument = ['-j', ntp_interfaces_json, '-t', conf_template] + self.run_script(argument, output_file=self.output_file) assert utils.cmp(expected, self.output_file), self.run_diff(expected, self.output_file) def test_backend_acl_template_render(self): @@ -623,13 +622,11 @@ class TestJ2Files(TestCase): input_file = os.path.join( self.test_dir, 'data', 'backend_acl', v['input'] ) - argument = " -j {} -t {} > {}".format( - input_file, acl_template, self.output_file - ) + argument = ["-j", input_file, "-t", acl_template] sample_output_file = os.path.join( self.test_dir, 'data', 'backend_acl', v['output'] ) - self.run_script(argument) + self.run_script(argument, output_file=self.output_file) assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) def tearDown(self): diff --git a/src/sonic-config-engine/tests/test_j2files_t2_chassis_fe.py b/src/sonic-config-engine/tests/test_j2files_t2_chassis_fe.py index e6bc82941b..b23bb93ddd 100644 --- a/src/sonic-config-engine/tests/test_j2files_t2_chassis_fe.py +++ b/src/sonic-config-engine/tests/test_j2files_t2_chassis_fe.py @@ -3,15 +3,15 @@ import json import os import shutil import subprocess - from unittest import TestCase import tests.common_utils as utils +from sonic_py_common.general import getstatusoutput_noshell class TestJ2FilesT2ChassisFe(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.t2_chassis_fe_minigraph = os.path.join(self.test_dir, 't2-chassis-fe-graph.xml') self.t2_chassis_fe_vni_minigraph = os.path.join(self.test_dir, 't2-chassis-fe-graph-vni.xml') self.t2_chassis_fe_pc_minigraph = os.path.join(self.test_dir, 't2-chassis-fe-graph-pc.xml') @@ -24,25 +24,28 @@ class TestJ2FilesT2ChassisFe(TestCase): except OSError: pass - def run_script(self, argument): - print('CMD: sonic-cfggen ' + argument) - output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) + def run_script(self, argument, output_file=None): + print('CMD: sonic-cfggen ' + ' '.join(argument)) + output = subprocess.check_output(self.script_file + argument) if utils.PY3x: output = output.decode() + if output_file: + with open(output_file, 'w') as f: + f.write(output) 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_case(self, minigraph, template, target): template_dir = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-fpm-frr', "frr") conf_template = os.path.join(template_dir, template) constants = os.path.join(self.test_dir, '..', '..', '..', 'files', 'image_config', 'constants', 'constants.yml') - cmd_args = minigraph, self.t2_chassis_fe_port_config, constants, conf_template, template_dir, self.output_file - cmd = "-m %s -p %s -y %s -t %s -T %s > %s" % cmd_args - self.run_script(cmd) + cmd = ["-m", minigraph, "-p", self.t2_chassis_fe_port_config, "-y", constants, "-t", conf_template, "-T", template_dir] + self.run_script(cmd, output_file=self.output_file) original_filename = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, target) r = filecmp.cmp(original_filename, self.output_file) diff --git a/src/sonic-config-engine/tests/test_minigraph_case.py b/src/sonic-config-engine/tests/test_minigraph_case.py index ba3274c867..90731e0855 100644 --- a/src/sonic-config-engine/tests/test_minigraph_case.py +++ b/src/sonic-config-engine/tests/test_minigraph_case.py @@ -2,7 +2,6 @@ import json import os import subprocess import ipaddress - import tests.common_utils as utils import minigraph @@ -17,7 +16,7 @@ class TestCfgGenCaseInsensitive(TestCase): def setUp(self): self.yang = utils.YangWrapper() 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_graph = os.path.join(self.test_dir, 'simple-sample-graph-case.xml') self.sample_simple_graph = os.path.join(self.test_dir, 'simple-sample-graph.xml') self.sample_resource_graph = os.path.join(self.test_dir, 'sample-graph-resource-type.xml') @@ -27,13 +26,13 @@ class TestCfgGenCaseInsensitive(TestCase): self.port_config = os.path.join(self.test_dir, 't0-sample-port-config.ini') def run_script(self, argument, check_stderr=False): - print('\n Running sonic-cfggen ' + argument) + print('\n Running sonic-cfggen ' + ' '.join(argument)) self.assertTrue(self.yang.validate(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() @@ -46,47 +45,47 @@ class TestCfgGenCaseInsensitive(TestCase): return output def test_dummy_run(self): - argument = '' + argument = [] output = self.run_script(argument) self.assertEqual(output, '') def test_minigraph_sku(self): - argument = '-v "DEVICE_METADATA[\'localhost\'][\'hwsku\']" -m "' + self.sample_graph + '" -p "' + self.port_config + '"' + argument = ['-v', "DEVICE_METADATA[\'localhost\'][\'hwsku\']", '-m', self.sample_graph, '-p', self.port_config] output = self.run_script(argument) self.assertEqual(output.strip(), 'Force10-S6000') def test_print_data(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" --print-data' + argument = ['-m', self.sample_graph, '-p', self.port_config, '--print-data'] output = self.run_script(argument) self.assertTrue(len(output.strip()) > 0) def test_jinja_expression(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'type\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'type\']"] output = self.run_script(argument) self.assertEqual(output.strip(), 'ToRRouter') def test_minigraph_subtype(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'subtype\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'subtype\']"] output = self.run_script(argument) self.assertEqual(output.strip(), 'DualToR') def test_minigraph_peer_switch_hostname(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'peer_switch\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'peer_switch\']"] output = self.run_script(argument) self.assertEqual(output.strip(), 'switch2-t0') def test_additional_json_data(self): - argument = '-a \'{"key1":"value1"}\' -v key1' + argument = ['-a', '{"key1":"value1"}', '-v', 'key1'] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1') def test_read_yaml(self): - argument = '-v yml_item -y ' + os.path.join(self.test_dir, 'test.yml') + argument = ['-v', 'yml_item', '-y', os.path.join(self.test_dir, 'test.yml')] output = self.run_script(argument) self.assertEqual(output.strip(), '[\'value1\', \'value2\']') def test_render_template(self): - argument = '-y ' + os.path.join(self.test_dir, 'test.yml') + ' -t ' + os.path.join(self.test_dir, 'test.j2') + argument = ['-y', os.path.join(self.test_dir, 'test.yml'), '-t', os.path.join(self.test_dir, 'test.j2')] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1\nvalue2') @@ -97,12 +96,12 @@ class TestCfgGenCaseInsensitive(TestCase): # self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '10.0.100.1'}}") def test_minigraph_interfaces(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v \'INTERFACE.keys()|list\'' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'INTERFACE.keys()|list'] output = self.run_script(argument) self.assertEqual(output.strip(), "[('Ethernet0', '10.0.0.58/31'), 'Ethernet0', ('Ethernet0', 'FC00::75/126')]") def test_minigraph_vlans(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v VLAN' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'VLAN'] output = self.run_script(argument) expected = { @@ -126,7 +125,7 @@ class TestCfgGenCaseInsensitive(TestCase): ) def test_minigraph_vlan_members(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v VLAN_MEMBER' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'VLAN_MEMBER'] output = self.run_script(argument) expected = { 'Vlan1000|Ethernet8': {'tagging_mode': 'untagged'}, @@ -138,12 +137,12 @@ class TestCfgGenCaseInsensitive(TestCase): ) def test_minigraph_vlan_interfaces_keys(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VLAN_INTERFACE.keys()|list"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "VLAN_INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual(output.strip(), "[('Vlan1000', '192.168.0.1/27'), 'Vlan1000']") def test_minigraph_vlan_interfaces(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VLAN_INTERFACE"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "VLAN_INTERFACE"] output = self.run_script(argument) expected_table = { 'Vlan1000|192.168.0.1/27': {}, @@ -155,7 +154,7 @@ class TestCfgGenCaseInsensitive(TestCase): self.assertEqual(utils.to_dict(output.strip()), expected_table) def test_minigraph_portchannels(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v PORTCHANNEL' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'PORTCHANNEL'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -163,44 +162,44 @@ class TestCfgGenCaseInsensitive(TestCase): ) def test_minigraph_console_mgmt_feature(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v CONSOLE_SWITCH' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'CONSOLE_SWITCH'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), utils.to_dict("{'console_mgmt': {'enabled': 'no'}}")) def test_minigraph_console_port(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v CONSOLE_PORT' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', 'CONSOLE_PORT'] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), utils.to_dict("{'1': {'baud_rate': '9600', 'remote_device': 'managed_device', 'flow_control': 1}}")) def test_minigraph_dhcp_server_feature(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'dhcp_server\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'dhcp_server\']"] output = self.run_script(argument) self.assertEqual(output.strip(), '') try: # For DHCP server enabled device type - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BMC_MGMT_TOR_ROUTER, self.sample_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BMC_MGMT_TOR_ROUTER), self.sample_graph]) output = self.run_script(argument) self.assertEqual(output.strip(), 'enabled') finally: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BMC_MGMT_TOR_ROUTER, TOR_ROUTER, self.sample_graph), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BMC_MGMT_TOR_ROUTER, TOR_ROUTER), self.sample_graph]) def test_minigraph_deployment_id(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'deployment_id\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'deployment_id\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "1") def test_minigraph_cluster(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'cluster\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'cluster\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "AAA00PrdStr00") def test_minigraph_neighbor_metadata(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_NEIGHBOR_METADATA"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_NEIGHBOR_METADATA"] expected_table = { 'switch2-t0': { @@ -253,43 +252,43 @@ class TestCfgGenCaseInsensitive(TestCase): # self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '10.0.100.1'}}") def test_metadata_tacacs(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "TACPLUS_SERVER"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "TACPLUS_SERVER"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'10.0.10.7': {'priority': '1', 'tcp_port': '49'}, '10.0.10.8': {'priority': '1', 'tcp_port': '49'}}") def test_metadata_kube(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "KUBERNETES_MASTER[\'SERVER\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "KUBERNETES_MASTER[\'SERVER\']"] output = self.run_script(argument) self.assertEqual(json.loads(output.strip().replace("'", "\"")), json.loads('{"ip": "10.10.10.10", "disable": "True"}')) def test_minigraph_mgmt_port(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "MGMT_PORT"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "MGMT_PORT"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'eth0': {'alias': 'eth0', 'admin_status': 'up', 'speed': '1000'}}") def test_metadata_ntp(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "NTP_SERVER"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "NTP_SERVER"] output = self.run_script(argument) self.assertEqual(output.strip(), "{'10.0.10.1': {}, '10.0.10.2': {}}") def test_minigraph_vnet(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VNET"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "VNET"] output = self.run_script(argument) self.assertEqual(output.strip(), "") def test_minigraph_vxlan(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VXLAN_TUNNEL"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "VXLAN_TUNNEL"] output = self.run_script(argument) self.assertEqual(output.strip(), "") def test_minigraph_bgp_mon(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "BGP_MONITORS"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "BGP_MONITORS"] output = self.run_script(argument) self.assertEqual(output.strip(), "{}") def test_minigraph_peer_switch(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "PEER_SWITCH"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "PEER_SWITCH"] expected_table = { 'switch2-t0': { 'address_ipv4': "25.1.1.10" @@ -314,7 +313,7 @@ class TestCfgGenCaseInsensitive(TestCase): self.assertTrue("mux_cable" not in port) def test_minigraph_storage_device(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'storage_device\']"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'storage_device\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "true") @@ -331,23 +330,23 @@ class TestCfgGenCaseInsensitive(TestCase): try: print('\n Change device type to %s' % (BACKEND_TOR_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file]) - argument = '-m "' + graph_file + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'storage_device\']"' + argument = ['-m', graph_file, '-p', self.port_config, '-v', "DEVICE_METADATA[\'localhost\'][\'storage_device\']"] output = self.run_script(argument) self.assertEqual(output.strip(), "true") finally: print('\n Change device type back to %s' % (TOR_ROUTER)) if check_stderr: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) else: - output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), shell=True) + output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file]) def test_minigraph_tunnel_table(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "TUNNEL"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "TUNNEL"] expected_tunnel = { "MuxTunnel0": { "tunnel_type": "IPINIP", @@ -367,7 +366,7 @@ class TestCfgGenCaseInsensitive(TestCase): # Validate tunnel config is as before when tunnel_qos_remap = disabled sample_graph_disabled_remap = os.path.join(self.test_dir, 'simple-sample-graph-case-remap-disabled.xml') - argument = '-m "' + sample_graph_disabled_remap + '" -p "' + self.port_config + '" -v "TUNNEL"' + argument = ['-m', sample_graph_disabled_remap, '-p', self.port_config, '-v', "TUNNEL"] output = self.run_script(argument) self.assertEqual( @@ -377,7 +376,7 @@ class TestCfgGenCaseInsensitive(TestCase): # Validate extra config is generated when tunnel_qos_remap = enabled sample_graph_enabled_remap = os.path.join(self.test_dir, 'simple-sample-graph-case-remap-enabled.xml') - argument = '-m "' + sample_graph_enabled_remap + '" -p "' + self.port_config + '" -v "TUNNEL"' + argument = ['-m', sample_graph_enabled_remap, '-p', self.port_config, '-v', "TUNNEL"] expected_tunnel = { "MuxTunnel0": { "tunnel_type": "IPINIP", @@ -402,7 +401,7 @@ class TestCfgGenCaseInsensitive(TestCase): # Validate extra config for mux tunnel is generated automatically when tunnel_qos_remap = enabled sample_graph_enabled_remap = os.path.join(self.test_dir, 'simple-sample-graph-case-remap-enabled-no-tunnel-attributes.xml') - argument = '-m "' + sample_graph_enabled_remap + '" -p "' + self.port_config + '" -v "TUNNEL"' + argument = ['-m', sample_graph_enabled_remap, '-p', self.port_config, '-v', "TUNNEL"] output = self.run_script(argument) self.assertEqual( utils.to_dict(output.strip()), @@ -410,7 +409,7 @@ class TestCfgGenCaseInsensitive(TestCase): ) def test_minigraph_mux_cable_table(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "MUX_CABLE"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "MUX_CABLE"] expected_table = { 'Ethernet4': { 'state': 'auto', @@ -434,7 +433,7 @@ class TestCfgGenCaseInsensitive(TestCase): ) def test_dhcp_table(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DHCP_RELAY"' + argument = ['-m', self.sample_graph, '-p', self.port_config, '-v', "DHCP_RELAY"] expected = { 'Vlan1000': { 'dhcpv6_servers': [ diff --git a/src/sonic-config-engine/tests/test_multinpu_cfggen.py b/src/sonic-config-engine/tests/test_multinpu_cfggen.py index 744afd1fbc..8559c10394 100644 --- a/src/sonic-config-engine/tests/test_multinpu_cfggen.py +++ b/src/sonic-config-engine/tests/test_multinpu_cfggen.py @@ -5,10 +5,10 @@ import shutil import subprocess import unittest import yaml - import tests.common_utils as utils from unittest import TestCase +from sonic_py_common.general import getstatusoutput_noshell SKU = 'multi-npu-01' @@ -23,7 +23,7 @@ class TestMultiNpuCfgGen(TestCase): self.yang = utils.YangWrapper() self.test_dir = os.path.dirname(os.path.realpath(__file__)) self.test_data_dir = os.path.join(self.test_dir, 'multi_npu_data') - 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_graph = os.path.join(self.test_data_dir, 'sample-minigraph.xml') self.sample_graph1 = os.path.join(self.test_data_dir, 'sample-minigraph-noportchannel.xml') self.sample_port_config = os.path.join(self.test_data_dir, 'sample_port_config.ini') @@ -33,17 +33,19 @@ class TestMultiNpuCfgGen(TestCase): self.output_file = os.path.join(self.test_dir, 'output') os.environ["CFGGEN_UNIT_TESTING"] = "2" - def run_script(self, argument, check_stderr=False): - print('\n Running sonic-cfggen ' + argument) + def run_script(self, argument, check_stderr=False, output_file=None): + print('\n Running sonic-cfggen ' + ' '.join(argument)) self.assertTrue(self.yang.validate(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() + if output_file: + with open(output_file, 'w') as f: + f.write(output) linecount = output.strip().count('\n') if linecount <= 0: @@ -53,15 +55,15 @@ class TestMultiNpuCfgGen(TestCase): 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_frr_asic_case(self, template, target, asic, port_config): template_dir = os.path.join(self.test_dir, '..', '..', '..', 'dockers', 'docker-fpm-frr', "frr") conf_template = os.path.join(template_dir, template) constants = os.path.join(self.test_dir, '..', '..', '..', 'files', 'image_config', 'constants', 'constants.yml') - cmd_args = asic, self.sample_graph, port_config, constants, conf_template, template_dir, self.output_file - cmd = "-n %s -m %s -p %s -y %s -t %s -T %s > %s" % cmd_args - self.run_script(cmd) + cmd = ['-n', asic, '-m', self.sample_graph, '-p', port_config, '-y', constants, '-t', conf_template, '-T', template_dir] + self.run_script(cmd, output_file=self.output_file) original_filename = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, target) r = filecmp.cmp(original_filename, self.output_file) @@ -72,37 +74,37 @@ class TestMultiNpuCfgGen(TestCase): def run_script_for_asic(self,argument,asic, port_config=None): - argument = "{} -n asic{} ".format(argument, asic) + cmd = argument + ["-n", "asic{}".format(asic)] if port_config: - argument += "-p {}".format(port_config) - output = self.run_script(argument) + cmd = argument + ["-n", "asic{}".format(asic), "-p", port_config] + output = self.run_script(cmd) return output def test_dummy_run(self): - argument = '' + argument = [] output = self.run_script(argument) self.assertEqual(output, '') def test_hwsku(self): - argument = "-v \"DEVICE_METADATA[\'localhost\'][\'hwsku\']\" -m \"{}\" -p \"{}\"".format(self.sample_graph, self.sample_port_config) + argument = ["-v", "DEVICE_METADATA[\'localhost\'][\'hwsku\']", "-m", self.sample_graph, "-p", self.sample_port_config] output = self.run_script(argument) self.assertEqual(output.strip(), SKU) - argument = "-v \"DEVICE_METADATA[\'localhost\'][\'hwsku\']\" -m \"{}\"".format(self.sample_graph) + argument = ["-v", "DEVICE_METADATA[\'localhost\'][\'hwsku\']", "-m", self.sample_graph] for asic in range(NUM_ASIC): output = self.run_script_for_asic(argument, asic, self.port_config[asic]) self.assertEqual(output.strip(), SKU) def test_print_data(self): - argument = "-m \"{}\" -p \"{}\" --print-data".format(self.sample_graph, self.sample_port_config) + argument = ["-m", self.sample_graph, "-p", self.sample_port_config, "--print-data"] output = self.run_script(argument) self.assertGreater(len(output.strip()) , 0) - argument = "-m \"{}\" --print-data".format(self.sample_graph) + argument = ["-m", self.sample_graph, "--print-data"] for asic in range(NUM_ASIC): output = self.run_script_for_asic(argument, asic, self.port_config[asic]) self.assertGreater(len(output.strip()) , 0) def test_additional_json_data(self): - argument = '-a \'{"key1":"value1"}\' -v key1' + argument = ['-a', '{"key1":"value1"}', '-v', 'key1'] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1') for asic in range(NUM_ASIC): @@ -110,15 +112,15 @@ class TestMultiNpuCfgGen(TestCase): self.assertEqual(output.strip(), 'value1') def test_read_yaml(self): - argument = '-v yml_item -y ' + os.path.join(self.test_dir, 'test.yml') - output = yaml.load(self.run_script(argument)) + argument = ['-v', 'yml_item', '-y', os.path.join(self.test_dir, 'test.yml')] + output = yaml.safe_load(self.run_script(argument)) self.assertListEqual(output, ['value1', 'value2']) for asic in range(NUM_ASIC): - output = yaml.load(self.run_script_for_asic(argument, asic, self.port_config[asic])) + output = yaml.safe_load(self.run_script_for_asic(argument, asic, self.port_config[asic])) self.assertListEqual(output, ['value1', 'value2']) def test_render_template(self): - argument = '-y ' + os.path.join(self.test_dir, 'test.yml') + ' -t ' + os.path.join(self.test_dir, 'test.j2') + argument = ['-y', os.path.join(self.test_dir, 'test.yml'), '-t', os.path.join(self.test_dir, 'test.j2')] output = self.run_script(argument) self.assertEqual(output.strip(), 'value1\nvalue2') for asic in range(NUM_ASIC): @@ -126,37 +128,37 @@ class TestMultiNpuCfgGen(TestCase): self.assertEqual(output.strip(), 'value1\nvalue2') def test_metadata_tacacs(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.sample_port_config + '" --var-json "TACPLUS_SERVER"' + argument = ['-m', self.sample_graph, '-p', self.sample_port_config, '--var-json', "TACPLUS_SERVER"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {'123.46.98.21': {'priority': '1', 'tcp_port': '49'}}) #TACPLUS_SERVER not present in the asic configuration. - argument = '-m "' + self.sample_graph + '" --var-json "TACPLUS_SERVER"' + argument = ['-m', self.sample_graph, '--var-json', "TACPLUS_SERVER"] for asic in range(NUM_ASIC): output = json.loads(self.run_script_for_asic(argument, asic, self.port_config[asic])) self.assertDictEqual(output, {}) def test_metadata_ntp(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.sample_port_config + '" --var-json "NTP_SERVER"' + argument = ['-m', self.sample_graph, '-p', self.sample_port_config, '--var-json', "NTP_SERVER"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {'17.39.1.130': {}, '17.39.1.129': {}}) #NTP data is present only in the host config - argument = '-m "' + self.sample_graph + '" --var-json "NTP_SERVER"' + argument = ['-m', self.sample_graph, '--var-json', "NTP_SERVER"] for asic in range(NUM_ASIC): output = json.loads(self.run_script_for_asic(argument, asic, self.port_config[asic])) print("Log:asic{} sku {}".format(asic,output)) self.assertDictEqual(output, {}) def test_mgmt_port(self): - argument = '-m "' + self.sample_graph + '" -p "' + self.sample_port_config + '" --var-json "MGMT_PORT"' + argument = ['-m', self.sample_graph, '-p', self.sample_port_config, '--var-json', "MGMT_PORT"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {'eth0': {'alias': 'eth0', 'admin_status': 'up'}}) - argument = '-m "' + self.sample_graph + '" --var-json "MGMT_PORT"' + argument = ['-m', self.sample_graph, '--var-json', "MGMT_PORT"] for asic in range(NUM_ASIC): output = json.loads(self.run_script_for_asic(argument, asic, self.port_config[asic])) self.assertDictEqual(output, {'eth0': {'alias': 'eth0', 'admin_status': 'up'}}) def test_frontend_asic_portchannels(self): - argument = "-m {} -p {} -n asic0 --var-json \"PORTCHANNEL\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORTCHANNEL"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'PortChannel0002': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet0', 'Ethernet4'], 'mtu': '9100', 'tpid': '0x8100', 'lacp_key': 'auto'}, @@ -164,14 +166,14 @@ class TestMultiNpuCfgGen(TestCase): 'PortChannel4002': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet-BP8', 'Ethernet-BP12'], 'mtu': '9100', 'tpid': '0x8100', 'lacp_key': 'auto'}}) def test_backend_asic_portchannels(self): - argument = "-m {} -p {} -n asic3 --var-json \"PORTCHANNEL\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "PORTCHANNEL"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'PortChannel4013': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet-BP384', 'Ethernet-BP388'], 'mtu': '9100', 'tpid': '0x8100', 'lacp_key': 'auto'}, 'PortChannel4014': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet-BP392', 'Ethernet-BP396'], 'mtu': '9100', 'tpid': '0x8100', 'lacp_key': 'auto'}}) def test_frontend_asic_portchannel_mem(self): - argument = "-m {} -p {} -n asic0 -v \"PORTCHANNEL_MEMBER.keys()|list\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "-v", "PORTCHANNEL_MEMBER.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -179,7 +181,7 @@ class TestMultiNpuCfgGen(TestCase): ) def test_backend_asic_portchannels_mem(self): - argument = "-m {} -p {} -n asic3 -v \"PORTCHANNEL_MEMBER.keys()|list\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "-v", "PORTCHANNEL_MEMBER.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -187,7 +189,7 @@ class TestMultiNpuCfgGen(TestCase): ) def test_frontend_asic_portchannel_intf(self): - argument = "-m {} -p {} -n asic0 -v \"PORTCHANNEL_INTERFACE.keys()|list\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "-v", "PORTCHANNEL_INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -195,7 +197,7 @@ class TestMultiNpuCfgGen(TestCase): ) def test_frontend_asic_routerport_intf(self): - argument = "-m {} -p {} -n asic0 -v \"INTERFACE.keys()|list\"".format(self.sample_graph1, self.port_config[0]) + argument = ["-m", self.sample_graph1, "-p", self.port_config[0], "-n", "asic0", "-v", "INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -203,7 +205,7 @@ class TestMultiNpuCfgGen(TestCase): ) def test_backend_asic_portchannel_intf(self): - argument = "-m {} -p {} -n asic3 -v \"PORTCHANNEL_INTERFACE.keys()|list\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "-v", "PORTCHANNEL_INTERFACE.keys()|list"] output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -211,7 +213,7 @@ class TestMultiNpuCfgGen(TestCase): ) def test_frontend_asic_ports(self): - argument = "-m {} -p {} -n asic0 --var-json \"PORT\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORT"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {"Ethernet0": { "admin_status": "up", "alias": "Ethernet1/1", "asic_port_name": "Eth0-ASIC0", "description": "01T2:Ethernet1", "index": "0", "lanes": "33,34,35,36", "mtu": "9100", "tpid": "0x8100", "pfc_asym": "off", "role": "Ext", "speed": "40000", "autoneg": "on" }, @@ -224,7 +226,7 @@ class TestMultiNpuCfgGen(TestCase): "Ethernet-BP12": { "admin_status": "up", "alias": "Eth7-ASIC0", "asic_port_name": "Eth7-ASIC0", "description": "ASIC3:Eth1-ASIC3", "index": "3", "lanes": "25,26,27,28", "mtu": "9100", "tpid": "0x8100", "pfc_asym": "off", "role": "Int", "speed": "40000" }}) def test_frontend_asic_ports_config_db(self): - argument = "-m {} -p {} -n asic0 --var-json \"PORT\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORT"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {"Ethernet0": { "admin_status": "up", "alias": "Ethernet1/1", "asic_port_name": "Eth0-ASIC0", "description": "01T2:Ethernet1", "index": "0", "lanes": "33,34,35,36", "mtu": "9100", "tpid": "0x8100", "pfc_asym": "off", "role": "Ext", "speed": "40000", "autoneg": "on" }, @@ -237,7 +239,7 @@ class TestMultiNpuCfgGen(TestCase): "Ethernet-BP12": { "admin_status": "up", "alias": "Eth7-ASIC0", "asic_port_name": "Eth7-ASIC0", "description": "ASIC3:Eth1-ASIC3", "index": "3", "lanes": "25,26,27,28", "mtu": "9100", "tpid": "0x8100", "pfc_asym": "off", "role": "Int", "speed": "40000" }}) def test_frontend_asic_device_neigh(self): - argument = "-m {} -p {} -n asic0 --var-json \"DEVICE_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "DEVICE_NEIGHBOR"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'Ethernet0': {'name': '01T2', 'port': 'Ethernet1'}, @@ -248,7 +250,7 @@ class TestMultiNpuCfgGen(TestCase): 'Ethernet-BP8': {'name': 'ASIC3', 'port': 'Eth0-ASIC3'}}) def test_frontend_asic_device_neigh_metadata(self): - argument = "-m {} -p {} -n asic0 --var-json \"DEVICE_NEIGHBOR_METADATA\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "DEVICE_NEIGHBOR_METADATA"] output = json.loads(self.run_script(argument)) print(output) self.assertDictEqual(output, \ @@ -257,7 +259,7 @@ class TestMultiNpuCfgGen(TestCase): 'ASIC2': {'lo_addr_v6': '::/0', 'mgmt_addr': '0.0.0.0/0', 'hwsku': 'multi-npu-asic', 'lo_addr': '0.0.0.0/0', 'type': 'Asic', 'mgmt_addr_v6': '::/0'}}) def test_backend_asic_device_neigh(self): - argument = "-m {} -p {} -n asic3 --var-json \"DEVICE_NEIGHBOR\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "DEVICE_NEIGHBOR"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'Ethernet-BP396': {'name': 'ASIC1', 'port': 'Eth7-ASIC1'}, @@ -266,7 +268,7 @@ class TestMultiNpuCfgGen(TestCase): 'Ethernet-BP388': {'name': 'ASIC0', 'port': 'Eth7-ASIC0'}}) def test_backend_device_neigh_metadata(self): - argument = "-m {} -p {} -n asic3 --var-json \"DEVICE_NEIGHBOR_METADATA\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "DEVICE_NEIGHBOR_METADATA"] output = json.loads(self.run_script(argument)) print(output) self.assertDictEqual(output, \ @@ -274,28 +276,28 @@ class TestMultiNpuCfgGen(TestCase): 'ASIC0': {'lo_addr_v6': '::/0', 'mgmt_addr': '0.0.0.0/0', 'hwsku': 'multi-npu-asic', 'lo_addr': '0.0.0.0/0', 'type': 'Asic', 'mgmt_addr_v6': '::/0'}}) def test_frontend_bgp_neighbor(self): - argument = "-m {} -p {} -n asic0 --var-json \"BGP_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "BGP_NEIGHBOR"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'10.0.0.1': {'rrclient': 0, 'name': '01T2', 'local_addr': '10.0.0.0', 'nhopself': 0, 'holdtime': '10', 'asn': '65200', 'keepalive': '3'}, 'fc00::2': {'rrclient': 0, 'name': '01T2', 'local_addr': 'fc00::1', 'nhopself': 0, 'holdtime': '10', 'asn': '65200', 'keepalive': '3'}}) def test_frontend_asic_bgp_neighbor(self): - argument = "-m {} -p {} -n asic0 --var-json \"BGP_INTERNAL_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "BGP_INTERNAL_NEIGHBOR"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'10.1.0.0': {'rrclient': 0, 'name': 'ASIC2', 'local_addr': '10.1.0.1', 'nhopself': 0, 'admin_status': 'up', 'holdtime': '0', 'asn': '65100', 'keepalive': '0'}, '10.1.0.2': {'rrclient': 0, 'name': 'ASIC3', 'local_addr': '10.1.0.3', 'nhopself': 0, 'admin_status': 'up', 'holdtime': '0', 'asn': '65100', 'keepalive': '0'}}) def test_backend_asic_bgp_neighbor(self): - argument = "-m {} -p {} -n asic3 --var-json \"BGP_INTERNAL_NEIGHBOR\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "BGP_INTERNAL_NEIGHBOR"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'10.1.0.7': {'rrclient': 0, 'name': 'ASIC1', 'local_addr': '10.1.0.6', 'nhopself': 0, 'admin_status': 'up', 'holdtime': '0', 'asn': '65100', 'keepalive': '0'}, '10.1.0.3': {'rrclient': 0, 'name': 'ASIC0', 'local_addr': '10.1.0.2', 'nhopself': 0, 'admin_status': 'up', 'holdtime': '0', 'asn': '65100', 'keepalive': '0'}}) def test_device_asic_metadata(self): - argument = "-m {} --var-json DEVICE_METADATA".format(self.sample_graph) + argument = ["-m", self.sample_graph, "--var-json", "DEVICE_METADATA"] for asic in range(NUM_ASIC): output = json.loads(self.run_script_for_asic(argument, asic,self.port_config[asic])) asic_name = "asic{}".format(asic) @@ -309,7 +311,7 @@ class TestMultiNpuCfgGen(TestCase): self.assertEqual(output['localhost']['deployment_id'], "1") def test_global_asic_acl(self): - argument = "-m {} -p {} --var-json \"ACL_TABLE\"".format(self.sample_graph, self.sample_port_config) + argument = ["-m", self.sample_graph, "-p", self.sample_port_config, "--var-json", "ACL_TABLE"] output = json.loads(self.run_script(argument)) exp = {\ 'SNMP_ACL': {'policy_desc': 'SNMP_ACL', 'type': 'CTRLPLANE', 'stage': 'ingress', 'services': ['SNMP']}, @@ -326,7 +328,7 @@ class TestMultiNpuCfgGen(TestCase): self.assertDictEqual(output, exp) def test_global_asic_acl1(self): - argument = "-m {} -p {} --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.sample_port_config) + argument = ["-m", self.sample_graph1, "-p", self.sample_port_config, "--var-json", "ACL_TABLE"] self.maxDiff = None output = json.loads(self.run_script(argument)) exp = {\ @@ -343,7 +345,7 @@ class TestMultiNpuCfgGen(TestCase): self.assertDictEqual(output, exp) def test_front_end_asic_acl(self): - argument = "-m {} -p {} -n asic0 --var-json \"ACL_TABLE\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "ACL_TABLE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ 'DATAACL': {'policy_desc': 'DATAACL', 'ports': ['PortChannel0002'], 'stage': 'ingress', 'type': 'L3'}, @@ -353,7 +355,7 @@ class TestMultiNpuCfgGen(TestCase): 'SSH_ONLY': {'policy_desc': 'SSH_ONLY', 'services': ['SSH'], 'stage': 'ingress', 'type': 'CTRLPLANE'}}) def test_front_end_asic_acl1(self): - argument = "-m {} -p {} -n asic0 --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.port_config[0]) + argument = ["-m", self.sample_graph1, "-p", self.port_config[0], "-n", "asic0", "--var-json", "ACL_TABLE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ 'EVERFLOW': {'policy_desc': 'EVERFLOW', 'ports': ['Ethernet0','Ethernet4'], 'stage': 'ingress', 'type': 'MIRROR'}, @@ -363,18 +365,18 @@ class TestMultiNpuCfgGen(TestCase): def test_back_end_asic_acl(self): - argument = "-m {} -p {} -n asic3 --var-json \"ACL_TABLE\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "ACL_TABLE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {}) def test_back_end_asic_acl1(self): - argument = "-m {} -p {} -n asic3 --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.port_config[3]) + argument = ["-m", self.sample_graph1, "-p", self.port_config[3], "-n", "asic3", "--var-json", "ACL_TABLE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {}) def test_loopback_intfs(self): - argument = "-m {} -p {} --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.sample_port_config) + argument = ["-m", self.sample_graph, "-p", self.sample_port_config, "--var-json", "LOOPBACK_INTERFACE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ "Loopback0": {}, @@ -382,7 +384,7 @@ class TestMultiNpuCfgGen(TestCase): "Loopback0|FC00:1::32/128": {}}) # The asic configuration should have 2 loopback interfaces - argument = "-m {} -p {} -n asic0 --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.port_config[0]) + argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "LOOPBACK_INTERFACE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, { \ "Loopback0": {}, @@ -392,7 +394,7 @@ class TestMultiNpuCfgGen(TestCase): "Loopback4096|8.0.0.0/32": {}, "Loopback4096|FD00:1::32/128": {}}) - argument = "-m {} -p {} -n asic3 --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.port_config[3]) + argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "LOOPBACK_INTERFACE"] output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ "Loopback0": {}, @@ -425,9 +427,7 @@ class TestMultiNpuCfgGen(TestCase): ) # asic0 - mix of front end and back end ports shutil.copy2(buffer_template, device_config_dir) - argument = "-m {} -p {} -n asic0 -t {}".format( - self.sample_graph, port_config_ini_asic0, device_buffer_template - ) + argument = ["-m", self.sample_graph, "-p", port_config_ini_asic0, "-n", "asic0", "-t", device_buffer_template] output = json.loads(self.run_script(argument)) os.remove(os.path.join(device_config_dir, "buffers_config.j2")) self.assertDictEqual(