diff --git a/src/sonic-config-engine/sonic-cfggen b/src/sonic-config-engine/sonic-cfggen index 287640d8a119..d5358f633dbf 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.safe_load(stream) + additional_data = yaml.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 d2be32c8544d..72325ecbc1e0 100644 --- a/src/sonic-config-engine/tests/common_utils.py +++ b/src/sonic-config-engine/tests/common_utils.py @@ -5,6 +5,7 @@ import sys import subprocess import argparse +import shlex PY3x = sys.version_info >= (3, 0) PYvX_DIR = "py3" if PY3x else "py2" @@ -46,7 +47,7 @@ def __init__(self, path=YANG_MODELS_DIR): self.yang_parser = sonic_yang.SonicYang(path) self.yang_parser.loadYangModel() self.test_dir = os.path.dirname(os.path.realpath(__file__)) - self.script_file = [PYTHON_INTERPRETTER, os.path.join(self.test_dir, '..', 'sonic-cfggen')] + self.script_file = PYTHON_INTERPRETTER + ' ' + os.path.join(self.test_dir, '..', 'sonic-cfggen') def validate(self, argument): """ @@ -61,22 +62,22 @@ def validate(self, argument): parser.add_argument("-p", "--port-config", help="port config file, used with -m or -k", nargs='?', const=None) parser.add_argument("-S", "--hwsku-config", help="hwsku config file, used with -p and -m or -k", nargs='?', const=None) parser.add_argument("-j", "--json", help="additional json file input, used with -p, -S and -m or -k", nargs='?', const=None) - args, unknown = parser.parse_known_args(argument) + args, unknown = parser.parse_known_args(shlex.split(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).decode() + cmd += ' -j ' + args.json + cmd += ' --print-data' + output = subprocess.check_output(cmd, shell=True).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 3b979d4a52ec..bcc0625ec8a8 100644 --- a/src/sonic-config-engine/tests/test_cfggen.py +++ b/src/sonic-config-engine/tests/test_cfggen.py @@ -1,6 +1,7 @@ import json import subprocess import os + import tests.common_utils as utils from unittest import TestCase @@ -15,7 +16,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') @@ -51,12 +52,13 @@ def tearDown(self): pass def run_script(self, argument, check_stderr=False, verbose=False): - print('\n Running sonic-cfggen ' + ' '.join(argument)) + print('\n Running sonic-cfggen ' + argument) self.assertTrue(self.yang.validate(argument)) + if check_stderr: - output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) if utils.PY3x: output = output.decode() @@ -71,52 +73,52 @@ def run_script(self, argument, check_stderr=False, verbose=False): 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) @@ -125,29 +127,29 @@ def test_jinja_expression(self, graph=None, port_config=None, expected_router_ty 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( @@ -173,20 +175,20 @@ def test_var_json_data(self, **kwargs): ) 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)) @@ -197,10 +199,10 @@ def test_template_batch_mode(self): 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(): @@ -210,7 +212,7 @@ def test_template_json_batch_mode(self): # 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", '')), @@ -234,7 +236,7 @@ def test_minigraph_acl(self): # 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()), @@ -242,13 +244,13 @@ def test_minigraph_mgmt_ports(self): ) 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()), @@ -264,7 +266,7 @@ def test_minigraph_vlans(self, **kwargs): 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( @@ -291,12 +293,12 @@ def test_minigraph_vlan_members(self, **kwargs): 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()), @@ -306,7 +308,7 @@ def test_minigraph_ecmp_fg_nhg(self): )) 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'," @@ -314,7 +316,7 @@ def test_minigraph_ecmp_members(self): " '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'," @@ -324,7 +326,7 @@ def test_minigraph_ecmp_neighbors(self): 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()), @@ -332,7 +334,7 @@ def test_minigraph_portchannels(self, **kwargs): ) 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()), @@ -340,7 +342,7 @@ def test_minigraph_portchannel_with_more_member(self): ) 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()), @@ -348,7 +350,7 @@ def test_minigraph_portchannel_members(self): ) 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()), @@ -356,7 +358,7 @@ def test_minigraph_portchannel_interfaces(self): ) 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()), @@ -367,7 +369,7 @@ def test_minigraph_neighbors(self): # 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()), @@ -381,7 +383,7 @@ def test_minigraph_extra_neighbors(self): ) 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()), @@ -390,7 +392,7 @@ def test_minigraph_port_description(self): 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()), @@ -399,7 +401,7 @@ def test_minigraph_port_fec_disabled(self): 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()), @@ -440,7 +442,7 @@ def test_minigraph_port_autonegotiation(self): ) # 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()), @@ -481,7 +483,7 @@ def test_minigraph_port_autonegotiation(self): ) 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()), @@ -489,7 +491,7 @@ def test_minigraph_port_rs(self): ) 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()), @@ -497,7 +499,7 @@ def test_minigraph_bgp(self): ) 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()), @@ -505,24 +507,24 @@ def test_minigraph_peers_with_range(self): ) 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()), @@ -530,7 +532,7 @@ def test_minigraph_ethernet_interfaces(self, **kwargs): ) 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( @@ -573,7 +575,7 @@ def test_minigraph_neighbor_interfaces(self): 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( @@ -616,7 +618,7 @@ def test_minigraph_neighbor_interfaces_config_db(self): 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( @@ -664,7 +666,7 @@ def test_minigraph_extra_ethernet_interfaces(self, **kwargs): # 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()), @@ -672,24 +674,24 @@ def test_metadata_tacacs(self): ) 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()), @@ -697,7 +699,7 @@ def test_minigraph_bgp_mon(self): ) 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()), @@ -705,7 +707,7 @@ def test_minigraph_bgp_voq_chassis_peer(self): ) # 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(), "") @@ -725,14 +727,14 @@ def test_minigraph_backend_acl_leaf(self, check_stderr=True): try: print('\n Change device type to %s' % (BACKEND_LEAF_ROUTER)) if check_stderr: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_LEAF_ROUTER), self.sample_backend_graph], stderr=subprocess.STDOUT) + 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) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_LEAF_ROUTER), self.sample_backend_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_LEAF_ROUTER, self.sample_backend_graph), shell=True) 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 @@ -741,9 +743,9 @@ def test_minigraph_backend_acl_leaf(self, check_stderr=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' % (BACKEND_LEAF_ROUTER, TOR_ROUTER), self.sample_backend_graph], stderr=subprocess.STDOUT) + 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) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, TOR_ROUTER), self.sample_backend_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, TOR_ROUTER, self.sample_backend_graph), shell=True) self.test_jinja_expression(self.sample_backend_graph, self.port_config, TOR_ROUTER) @@ -751,23 +753,23 @@ def test_minigraph_sub_port_no_vlan_member(self, check_stderr=True): try: print('\n Change device type to %s' % (BACKEND_LEAF_ROUTER)) if check_stderr: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (LEAF_ROUTER, BACKEND_LEAF_ROUTER), self.sample_graph], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (LEAF_ROUTER, BACKEND_LEAF_ROUTER, self.sample_graph), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (LEAF_ROUTER, BACKEND_LEAF_ROUTER), self.sample_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (LEAF_ROUTER, BACKEND_LEAF_ROUTER, self.sample_graph), shell=True) 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' % (BACKEND_LEAF_ROUTER, LEAF_ROUTER), self.sample_graph], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, LEAF_ROUTER, self.sample_graph), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_LEAF_ROUTER, LEAF_ROUTER), self.sample_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_LEAF_ROUTER, LEAF_ROUTER, self.sample_graph), shell=True) 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(), "{}") @@ -785,33 +787,33 @@ def verify_sub_intf(self, **kwargs): try: print('\n Change device type to %s' % (BACKEND_TOR_ROUTER)) if check_stderr: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), shell=True) 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 @@ -827,7 +829,7 @@ def verify_sub_intf(self, **kwargs): 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 @@ -860,24 +862,24 @@ def verify_sub_intf(self, **kwargs): finally: print('\n Change device type back to %s' % (TOR_ROUTER)) if check_stderr: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), shell=True) 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", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "DEVICE_METADATA"] + argument = "-j {} -m {} -p {} --var-json DEVICE_METADATA".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) output = json.loads(self.run_script(argument)) self.assertEqual(output['localhost']['asic_name'], 'Asic0') self.assertEqual(output['localhost']['switch_id'], '0') @@ -885,7 +887,7 @@ def test_minigraph_voq_metadata(self): self.assertEqual(output['localhost']['max_cores'], '16') def test_minigraph_voq_system_ports(self): - argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "SYSTEM_PORT"] + argument = "-j {} -m {} -p {} --var-json SYSTEM_PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) self.assertDictEqual( json.loads(self.run_script(argument)), { @@ -904,7 +906,7 @@ def test_minigraph_voq_system_ports(self): ) 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()), @@ -912,7 +914,7 @@ def test_minigraph_voq_port_macsec_enabled(self): ) def test_minigraph_voq_inband_interface_port(self): - argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "VOQ_INBAND_INTERFACE"] + argument = "-j {} -m {} -p {} --var-json VOQ_INBAND_INTERFACE".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -925,7 +927,7 @@ def test_minigraph_voq_inband_interface_port(self): ) def test_minigraph_voq_inband_port(self): - argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "PORT"] + argument = "-j {} -m {} -p {} --var-json PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -943,7 +945,7 @@ def test_minigraph_voq_inband_port(self): }) def test_minigraph_voq_recirc_ports(self): - argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "--var-json", "PORT"] + argument = "-j {} -m {} -p {} --var-json PORT".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) output = self.run_script(argument) output_dict = utils.to_dict(output.strip()) self.assertDictEqual( @@ -961,7 +963,7 @@ def test_minigraph_voq_recirc_ports(self): }) 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()), @@ -972,7 +974,7 @@ def test_minigraph_dhcp(self): ) 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()), @@ -980,14 +982,14 @@ def test_minigraph_bgp_packet_chassis_peer(self): ) 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()), @@ -995,7 +997,7 @@ def test_minigraph_bgp_packet_chassis_static_route(self): ) 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()), @@ -1003,14 +1005,14 @@ def test_minigraph_bgp_packet_chassis_vlan_subintf(self): ) def test_minigraph_voq_400g_zr_port_config(self): - argument = ["-j", self.macsec_profile, "-m", self.sample_graph_voq, "-p", self.voq_port_config, "-v" "PORT[\'Ethernet4\']"] + argument = "-j {} -m {} -p {} -v \"PORT[\'Ethernet4\']\"".format(self.macsec_profile, self.sample_graph_voq, self.voq_port_config) 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", self.packet_chassis_graph, "-p", self.packet_chassis_port_ini, "-n", "asic1", "-v", "PORT[\'Ethernet13\']"] + argument = "-m {} -p {} -n asic1 -v \"PORT[\'Ethernet13\']\"".format(self.packet_chassis_graph, self.packet_chassis_port_ini) 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 ddc77c8b0311..801340ea4722 100644 --- a/src/sonic-config-engine/tests/test_cfggen_from_yang.py +++ b/src/sonic-config-engine/tests/test_cfggen_from_yang.py @@ -4,7 +4,6 @@ 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 @@ -22,18 +21,20 @@ 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) + output = subprocess.check_output(self.script_file + ' ' + arg, + stderr=subprocess.STDOUT, + shell=True) else: - output = subprocess.check_output(self.script_file + arg) + output = subprocess.check_output(self.script_file + ' ' + arg, + shell=True) if utils.PY3x: output = output.decode() @@ -47,31 +48,32 @@ def run_script(self, arg, check_stderr=False): return output def run_diff(self, file1, file2): - _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) - return output + return subprocess.check_output('diff -u {} {} || true'.format( + file1, file2), + shell=True) def run_script_with_yang_arg(self, arg, check_stderr=False): - args = ["-Y", self.sample_yang_file] + arg + args = "-Y {} {}".format(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', @@ -85,7 +87,7 @@ def test_device_metadata(self): def test_port_table(self): - arg = ["--var-json", "PORT"] + arg = "--var-json \"PORT\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == \ {'Ethernet0': {'admin_status': 'up', 'alias': 'eth0', 'description': 'Ethernet0', 'fec': 'rs', 'lanes': '65, 66', 'mtu': '9100', 'pfc_asym': 'on', 'speed': '40000'}, @@ -99,7 +101,7 @@ def test_port_table(self): }) def test_portchannel_table(self): - arg = ["--var-json", "PORTCHANNEL"] + arg = "--var-json \"PORTCHANNEL\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == \ {'PortChannel1001': {'admin_status': 'up', @@ -114,7 +116,7 @@ def test_portchannel_table(self): 'mtu': '9100'}}) def test_portchannel_member_table(self): - arg = ["--var-json", "PORTCHANNEL_MEMBER"] + arg = "--var-json \"PORTCHANNEL_MEMBER\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output ==\ { "PortChannel1001|Ethernet0": {}, @@ -124,7 +126,7 @@ def test_portchannel_member_table(self): }) def test_interface_table(self): - arg = ["--var-json", "INTERFACE"] + arg = "--var-json \"INTERFACE\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output =={\ "Ethernet8": {}, @@ -148,7 +150,7 @@ def test_interface_table(self): }) def test_portchannel_interface_table(self): - arg = ["--var-json", "PORTCHANNEL_INTERFACE"] + arg = "--var-json \"PORTCHANNEL_INTERFACE\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output =={\ "PortChannel1001|10.0.0.1/31": {}, @@ -156,7 +158,7 @@ def test_portchannel_interface_table(self): }) 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": {}, @@ -171,7 +173,7 @@ def test_loopback_table(self): }) def test_acl_table(self): - arg = ["--var-json", "ACL_TABLE"] + arg = "--var-json \"ACL_TABLE\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ 'DATAACL': {'policy_desc': 'DATAACL', 'ports': ['PortChannel1001','PortChannel1002'], 'stage': 'ingress', 'type': 'L3'}, @@ -181,7 +183,7 @@ def test_acl_table(self): 'SSH_ONLY': {'policy_desc': 'SSH_ONLY', 'services': ['SSH'], 'stage': 'ingress', 'type': 'CTRLPLANE'}}) def test_acl_rule(self): - arg = ["--var-json", "ACL_RULE"] + arg = "--var-json \"ACL_RULE\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ "DATAACL|Rule1": { @@ -199,7 +201,7 @@ def test_acl_rule(self): }) def test_vlan_table(self): - arg = ["--var-json", "VLAN"] + arg = "--var-json \"VLAN\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ "Vlan100": { @@ -216,7 +218,7 @@ def test_vlan_table(self): }) def test_vlan_interface(self): - arg = ["--var-json", "VLAN_INTERFACE"] + arg = "--var-json \"VLAN_INTERFACE\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ "Vlan100": {}, @@ -231,7 +233,7 @@ def test_vlan_interface(self): }) def test_vlan_member(self): - arg = ["--var-json", "VLAN_MEMBER"] + arg = "--var-json \"VLAN_MEMBER\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ "Vlan100|Ethernet24": { @@ -243,7 +245,7 @@ def test_vlan_member(self): }) def test_vlan_crm(self): - arg = ["--var-json", "CRM"] + arg = "--var-json \"CRM\"" output = json.loads(self.run_script_with_yang_arg(arg)) assert(output == {\ "Config": { 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 b3cad3aa2152..1ac2b7f7f5f3 100644 --- a/src/sonic-config-engine/tests/test_cfggen_pfx_filter.py +++ b/src/sonic-config-engine/tests/test_cfggen_pfx_filter.py @@ -9,14 +9,13 @@ class TestPfxFilter(TestCase): def test_comprehensive(self): # Generate output data_dir = "tests/data/pfx_filter" - 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) + 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) # 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) + res = subprocess.check_output(cmd, shell=True) 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 5d39fd2f3660..0af361718b99 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 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) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) if utils.PY3x: output = output.decode() @@ -44,18 +44,18 @@ def run_script(self, argument, check_stderr=False): return output def test_dummy_run(self): - argument = [] + argument = '' output = self.run_script(argument) self.assertEqual(output, '') def test_print_data(self): - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '--print-data'] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" --print-data' output = self.run_script(argument) self.assertTrue(len(output.strip()) > 0) # Check whether all interfaces present or not as per platform.json def test_platform_json_interfaces_keys(self): - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT.keys()|list"] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT.keys()|list"' output = self.run_script(argument) self.maxDiff = None @@ -71,24 +71,24 @@ def test_platform_json_interfaces_keys(self): 'Ethernet139', 'Ethernet140', 'Ethernet141', 'Ethernet142', 'Ethernet144' ] - self.assertEqual(sorted(ast.literal_eval(output.strip())), sorted(expected)) + self.assertEqual(sorted(eval(output.strip())), sorted(expected)) # Check specific Interface with it's proper configuration as per platform.json def test_platform_json_specific_ethernet_interfaces(self): - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet8\']"] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT[\'Ethernet8\']"' output = self.run_script(argument) self.maxDiff = None expected = "{'index': '3', 'lanes': '8', 'description': 'Eth3/1', 'mtu': '9100', 'alias': 'Eth3/1', 'pfc_asym': 'off', 'speed': '25000', 'tpid': '0x8100'}" self.assertEqual(utils.to_dict(output.strip()), utils.to_dict(expected)) - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet112\']"] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT[\'Ethernet112\']"' output = self.run_script(argument) self.maxDiff = None expected = "{'index': '29', 'lanes': '112', 'description': 'Eth29/1', 'mtu': '9100', 'alias': 'Eth29/1', 'pfc_asym': 'off', 'speed': '25000', 'tpid': '0x8100'}" self.assertEqual(utils.to_dict(output.strip()), utils.to_dict(expected)) - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT[\'Ethernet4\']"] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT[\'Ethernet4\']"' output = self.run_script(argument) self.maxDiff = None expected = "{'index': '2', 'lanes': '4,5', 'description': 'Eth2/1', 'admin_status': 'up', 'mtu': '9100', 'alias': 'Eth2/1', 'pfc_asym': 'off', 'speed': '50000', 'tpid': '0x8100'}" @@ -97,7 +97,7 @@ def test_platform_json_specific_ethernet_interfaces(self): # Check all Interface with it's proper configuration as per platform.json def test_platform_json_all_ethernet_interfaces(self): - argument = ['-m', self.platform_sample_graph, '-p', self.platform_json, '-S', self.hwsku_json, '-v', "PORT"] + argument = '-m "' + self.platform_sample_graph + '" -p "' + self.platform_json + '" -S "' + self.hwsku_json + '" -v "PORT"' output = self.run_script(argument) self.maxDiff = None 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 9bf3fc8a3ed5..a3d6d02a7ff2 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,5 +1,6 @@ import os import subprocess + import tests.common_utils as utils from unittest import TestCase @@ -9,18 +10,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 ' + ' '.join(argument)) + print('\n Running sonic-cfggen ' + argument) if check_stderr: - output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) if utils.PY3x: output = output.decode() @@ -33,12 +34,12 @@ def run_script(self, argument, check_stderr=False): 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()), @@ -52,7 +53,7 @@ def test_minigraph_t2_chassis_fe_interfaces(self): ) ) 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()), @@ -69,17 +70,17 @@ def test_minigraph_t2_chassis_fe_pc_interfaces(self): # 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 c30bc8b4969f..3934f8c7d70c 100644 --- a/src/sonic-config-engine/tests/test_frr.py +++ b/src/sonic-config-engine/tests/test_frr.py @@ -3,13 +3,13 @@ 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,19 +21,15 @@ def tearDown(self): pass - def run_script(self, argument, check_stderr=False, output_file=None): + def run_script(self, argument, check_stderr=False): # print '\n Running sonic-cfggen ' + argument - if check_stderr: - output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) 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: @@ -43,7 +39,8 @@ def run_script(self, argument, check_stderr=False, output_file=None): return output def run_diff(self, file1, file2): - _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) + output = subprocess.check_output('diff -u {} {} || true'.format(file1, file2), shell=True) + if utils.PY3x: output = output.decode() @@ -53,8 +50,9 @@ 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 = ['-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) + 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) original_filename = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, target) r = filecmp.cmp(original_filename, self.output_file) @@ -71,3 +69,4 @@ def test_bgpd_frr(self): 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 ad86f1cb4b2a..3ac219468e43 100644 --- a/src/sonic-config-engine/tests/test_j2files.py +++ b/src/sonic-config-engine/tests/test_j2files.py @@ -5,13 +5,12 @@ from unittest import TestCase import tests.common_utils as utils -from sonic_py_common.general import 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") @@ -41,21 +40,17 @@ def setUp(self): self.output_file = os.path.join(self.test_dir, 'output') os.environ["CFGGEN_UNIT_TESTING"] = "2" - def run_script(self, argument, output_file=None): - print('CMD: sonic-cfggen ', argument) - output = subprocess.check_output(self.script_file + argument) + def run_script(self, argument): + print('CMD: sonic-cfggen ' + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) 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): - _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) - return output + return subprocess.check_output('diff -u {} {} || true'.format(file1, file2), shell=True) def create_machine_conf(self, platform, vendor): file_exist = True @@ -64,24 +59,23 @@ def create_machine_conf(self, platform, vendor): 'dell': 'onie', 'mellanox': 'onie' } - echo_cmd1 = ["echo", '{}_platform={}'.format(mode[vendor], platform)] - echo_cmd2 = ["sudo", "tee", "-a", "/host/machine.conf"] + echo_cmd = "echo '{}_platform={}' | sudo tee -a /host/machine.conf > /dev/null".format(mode[vendor], platform) if not os.path.exists('/host/machine.conf'): file_exist = False if not os.path.isdir('/host'): dir_exist = False - subprocess.call(['sudo', 'mkdir', '/host']) - subprocess.call(['sudo', 'touch', '/host/machine.conf']) - getstatusoutput_noshell_pipe(echo_cmd1, echo_cmd2) + os.system('sudo mkdir /host') + os.system('sudo touch /host/machine.conf') + os.system(echo_cmd) return file_exist, dir_exist def remove_machine_conf(self, file_exist, dir_exist): if not file_exist: - subprocess.call(['sudo', 'rm', '-f', '/host/machine.conf']) + os.system('sudo rm -f /host/machine.conf') if not dir_exist: - subprocess.call(['sudo', 'rmdir', '/host']) + os.system('sudo rmdir /host') def modify_cable_len(self, base_file, file_dir): input_file = os.path.join(file_dir, base_file) @@ -101,76 +95,76 @@ def test_interfaces(self): 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.run_script(argument, output_file=self.output_file) + 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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=self.output_file) + 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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=self.output_file) + 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) 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.run_script(argument, output_file=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) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.simple_minigraph + ' -p ' + self.t0_port_config + ' -t ' + ports_template + ' > ' + self.output_file + self.run_script(argument) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.t0_minigraph + ' -j ' + dhc_sample_data + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file + self.run_script(argument) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.t0_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file + self.run_script(argument) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.no_ip_helper_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file + self.run_script(argument) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.radv_test_minigraph + ' -p ' + self.t0_port_config + ' -t ' + template_path + ' > ' + self.output_file + self.run_script(argument) self.assertTrue(utils.cmp(os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, 'radvd.conf'), self.output_file)) def test_lldp(self): @@ -182,32 +176,32 @@ def test_lldp(self): # 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', mgmt_iface_ipv4_and_ipv6_json, '-t', lldpd_conf_template] - self.run_script(argument, output_file=self.output_file) + argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv4_and_ipv6_json, lldpd_conf_template, self.output_file) + self.run_script(argument) 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', mgmt_iface_ipv4_json, '-t', lldpd_conf_template] - self.run_script(argument, output_file=self.output_file) + argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv4_json, lldpd_conf_template, self.output_file) + self.run_script(argument) 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', mgmt_iface_ipv6_json, '-t', lldpd_conf_template] - self.run_script(argument, output_file=self.output_file) + argument = '-j {} -t {} > {}'.format(mgmt_iface_ipv6_json, lldpd_conf_template, self.output_file) + self.run_script(argument) 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.t0_minigraph + ' -p ' + self.t0_port_config + ' -t ' + ipinip_file + ' > ' + self.output_file + self.run_script(argument) 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) @@ -218,14 +212,14 @@ def test_l2switch_template(self): 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) @@ -236,14 +230,14 @@ def test_l1_ports_template(self): 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) @@ -254,7 +248,7 @@ def test_l3_ports_template(self): 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) @@ -276,7 +270,9 @@ def test_l2switch_template_dualtor(self): "Ethernet112", "Ethernet116", "Ethernet120", "Ethernet124" ] } - argument = ['-a', json.dumps(extra_args), '-k', 'Arista-7050CX3-32S-D48C8', '--preset', 'l2', '-p', self.t0_7050cx3_port_config] + argument = '-a \'{}\' -k Arista-7050CX3-32S-D48C8 --preset l2 -p {}'.format( + json.dumps(extra_args), self.t0_7050cx3_port_config + ) output = self.run_script(argument) output_json = json.loads(output) @@ -303,8 +299,8 @@ def do_test_qos_and_buffer_arista7800r3_48cq2_lc_render_template(self, platform, 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + self.arista7800r3_48cq2_lc_t2_minigraph + ' -p ' + port_config_ini_file + ' -t ' + template_file + ' > ' + self.output_file + self.run_script(argument) # cleanup cfg_file_new = os.path.join(arista_dir_path, cfg_file) @@ -339,8 +335,8 @@ def _test_qos_render_template(self, vendor, platform, sku, minigraph, expected): 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + minigraph + ' -p ' + port_config_ini_file + ' -t ' + qos_file + ' > ' + self.output_file + self.run_script(argument) # cleanup qos_config_file_new = os.path.join(dir_path, 'qos_config.j2') @@ -402,8 +398,8 @@ def test_qos_dscp_remapping_render_template(self): 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] - self.run_script(argument, output_file=test_output) + argument = '-m ' + sample_minigraph_file + ' -p ' + port_config_ini_file + ' -t ' + qos_file + ' > ' + test_output + self.run_script(argument) # cleanup qos_config_file_new = os.path.join(device_template_path, 'qos_config.j2') @@ -436,8 +432,8 @@ def test_config_brcm_render_template(self): 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] - self.run_script(argument, output_file=config_test_output) + argument = '-m ' + sample_minigraph_file + ' -p ' + port_config_ini_file + ' -t ' + config_bcm_file + ' > ' + config_test_output + self.run_script(argument) #check output config.bcm config_sample_output_file = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, config_sample_output) @@ -455,8 +451,8 @@ def _test_buffers_render_template(self, vendor, platform, sku, minigraph, buffer 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.run_script(argument, output_file=self.output_file) + argument = '-m ' + minigraph + ' -p ' + port_config_ini_file + ' -t ' + buffers_file + ' > ' + self.output_file + self.run_script(argument) # cleanup buffers_config_file_new = os.path.join(dir_path, 'buffers_config.j2') @@ -520,9 +516,9 @@ def test_extra_lossless_buffer_for_tunnel_remapping(self): 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'] + argument = '-m ' + self.multi_asic_minigraph + ' -p ' + self.multi_asic_port_config + ' -t ' + ipinip_file + ' -n asic0 ' + ' > ' + self.output_file print(argument) - self.run_script(argument, output_file=self.output_file) + self.run_script(argument) 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) @@ -548,11 +544,13 @@ def test_swss_switch_render_template(self): }, } for _, v in test_list.items(): - argument = ["-m", v["graph"], "-p", v["port_config"], "-y", constants_yml, "-t", switch_template] + argument = " -m {} -p {} -y {} -t {} > {}".format( + v["graph"], v["port_config"], constants_yml, switch_template, self.output_file + ) sample_output_file = os.path.join( self.test_dir, 'sample_output', v["output"] ) - self.run_script(argument, output_file=self.output_file) + self.run_script(argument) 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): @@ -577,11 +575,14 @@ def test_swss_switch_render_template_multi_asic(self): } for _, v in test_list.items(): os.environ["NAMESPACE_ID"] = v["namespace_id"] - argument = ["-m", self.t1_mlnx_minigraph, "-y", constants_yml, "-t", switch_template] + argument = " -m {} -y {} -t {} > {}".format( + self.t1_mlnx_minigraph, constants_yml, switch_template, + self.output_file + ) sample_output_file = os.path.join( self.test_dir, 'sample_output', v["output"] ) - self.run_script(argument, output_file=self.output_file) + self.run_script(argument) assert utils.cmp(sample_output_file, self.output_file), self.run_diff(sample_output_file, self.output_file) os.environ["NAMESPACE_ID"] = "" @@ -590,8 +591,8 @@ def test_ndppd_conf(self): 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', vlan_interfaces_json, '-t', conf_template] - self.run_script(argument, output_file=self.output_file) + argument = '-j {} -t {} > {}'.format(vlan_interfaces_json, conf_template, self.output_file) + self.run_script(argument) assert utils.cmp(expected, self.output_file), self.run_diff(expected, self.output_file) def test_ntp_conf(self): @@ -599,8 +600,8 @@ def test_ntp_conf(self): 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', ntp_interfaces_json, '-t', conf_template] - self.run_script(argument, output_file=self.output_file) + argument = '-j {} -t {} > {}'.format(ntp_interfaces_json, conf_template, self.output_file) + self.run_script(argument) assert utils.cmp(expected, self.output_file), self.run_diff(expected, self.output_file) def test_backend_acl_template_render(self): @@ -622,11 +623,13 @@ def test_backend_acl_template_render(self): input_file = os.path.join( self.test_dir, 'data', 'backend_acl', v['input'] ) - argument = ["-j", input_file, "-t", acl_template] + argument = " -j {} -t {} > {}".format( + input_file, acl_template, self.output_file + ) sample_output_file = os.path.join( self.test_dir, 'data', 'backend_acl', v['output'] ) - self.run_script(argument, output_file=self.output_file) + self.run_script(argument) 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 7041bc1b9b5c..e6bc82941bf6 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,6 +3,7 @@ import os import shutil import subprocess + from unittest import TestCase import tests.common_utils as utils @@ -10,7 +11,7 @@ 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') @@ -23,28 +24,25 @@ def tearDown(self): except OSError: pass - def run_script(self, argument, output_file=None): - print('CMD: sonic-cfggen ' + ' '.join(argument)) - output = subprocess.check_output(self.script_file + argument) + def run_script(self, argument): + print('CMD: sonic-cfggen ' + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) 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): - _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) - return output + return subprocess.check_output('diff -u {} {} || true'.format(file1, file2), shell=True) 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 = ["-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) + 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) 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 8c399920b88b..9ee8a49db6ae 100644 --- a/src/sonic-config-engine/tests/test_minigraph_case.py +++ b/src/sonic-config-engine/tests/test_minigraph_case.py @@ -2,6 +2,7 @@ import os import subprocess import ipaddress + import tests.common_utils as utils import minigraph @@ -16,7 +17,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') @@ -26,13 +27,13 @@ def setUp(self): 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 ' + ' '.join(argument)) + print('\n Running sonic-cfggen ' + argument) self.assertTrue(self.yang.validate(argument)) if check_stderr: - output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) if utils.PY3x: output = output.decode() @@ -45,47 +46,47 @@ def run_script(self, argument, check_stderr=False): return output def test_dummy_run(self): - argument = [] + argument = '' output = self.run_script(argument) self.assertEqual(output, '') def test_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') @@ -96,12 +97,12 @@ def test_render_template(self): # 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 = { @@ -125,7 +126,7 @@ def test_minigraph_vlans(self): ) 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'}, @@ -137,12 +138,12 @@ def test_minigraph_vlan_members(self): ) 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': {}, @@ -154,7 +155,7 @@ def test_minigraph_vlan_interfaces(self): 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()), @@ -162,44 +163,44 @@ def test_minigraph_portchannels(self): ) 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' % (TOR_ROUTER, BMC_MGMT_TOR_ROUTER), self.sample_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BMC_MGMT_TOR_ROUTER, self.sample_graph), shell=True) output = self.run_script(argument) self.assertEqual(output.strip(), 'enabled') finally: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BMC_MGMT_TOR_ROUTER, TOR_ROUTER), self.sample_graph]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BMC_MGMT_TOR_ROUTER, TOR_ROUTER, self.sample_graph), shell=True) 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': { @@ -252,43 +253,43 @@ def test_minigraph_neighbor_metadata(self): # 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" @@ -313,7 +314,7 @@ def test_mux_cable_parsing(self): 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") @@ -330,23 +331,23 @@ def verify_storage_device_set(self, graph_file, check_stderr=False): try: print('\n Change device type to %s' % (BACKEND_TOR_ROUTER)) if check_stderr: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (TOR_ROUTER, BACKEND_TOR_ROUTER), graph_file]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (TOR_ROUTER, BACKEND_TOR_ROUTER, graph_file), shell=True) - 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' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file], stderr=subprocess.STDOUT) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(["sed", "-i", 's/%s/%s/g' % (BACKEND_TOR_ROUTER, TOR_ROUTER), graph_file]) + output = subprocess.check_output("sed -i \'s/%s/%s/g\' %s" % (BACKEND_TOR_ROUTER, TOR_ROUTER, graph_file), shell=True) 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", @@ -366,7 +367,7 @@ def test_minigraph_tunnel_table(self): # 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( @@ -376,7 +377,7 @@ def test_minigraph_tunnel_table(self): # 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", @@ -401,7 +402,7 @@ def test_minigraph_tunnel_table(self): # 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()), @@ -409,7 +410,7 @@ def test_minigraph_tunnel_table(self): ) 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', @@ -433,7 +434,7 @@ def test_minigraph_mux_cable_table(self): ) 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 070137af3960..2bfb879a1ad2 100644 --- a/src/sonic-config-engine/tests/test_multinpu_cfggen.py +++ b/src/sonic-config-engine/tests/test_multinpu_cfggen.py @@ -5,6 +5,7 @@ import subprocess import unittest import yaml + import tests.common_utils as utils from unittest import TestCase @@ -22,7 +23,7 @@ def setUp(self): 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') @@ -32,19 +33,17 @@ def setUp(self): self.output_file = os.path.join(self.test_dir, 'output') os.environ["CFGGEN_UNIT_TESTING"] = "2" - def run_script(self, argument, check_stderr=False, output_file=None): - print('\n Running sonic-cfggen ' + ' '.join(argument)) + def run_script(self, argument, check_stderr=False): + print('\n Running sonic-cfggen ' + argument) self.assertTrue(self.yang.validate(argument)) + if check_stderr: - output = subprocess.check_output(self.script_file + argument, stderr=subprocess.STDOUT) + output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True) else: - output = subprocess.check_output(self.script_file + argument) + output = subprocess.check_output(self.script_file + ' ' + argument, shell=True) 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: @@ -54,15 +53,15 @@ def run_script(self, argument, check_stderr=False, output_file=None): return output def run_diff(self, file1, file2): - _, output = getstatusoutput_noshell(['diff', '-u', file1, file2]) - return output + return subprocess.check_output('diff -u {} {} || true'.format(file1, file2), shell=True) 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 = ['-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) + 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) original_filename = os.path.join(self.test_dir, 'sample_output', utils.PYvX_DIR, target) r = filecmp.cmp(original_filename, self.output_file) @@ -73,37 +72,37 @@ def run_frr_asic_case(self, template, target, asic, port_config): def run_script_for_asic(self,argument,asic, port_config=None): - cmd = argument + ["-n", "asic{}".format(asic)] + argument = "{} -n asic{} ".format(argument, asic) if port_config: - cmd = argument + ["-n", "asic{}".format(asic), "-p", port_config] - output = self.run_script(cmd) + argument += "-p {}".format(port_config) + output = self.run_script(argument) 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", self.sample_graph, "-p", self.sample_port_config] + argument = "-v \"DEVICE_METADATA[\'localhost\'][\'hwsku\']\" -m \"{}\" -p \"{}\"".format(self.sample_graph, self.sample_port_config) output = self.run_script(argument) self.assertEqual(output.strip(), SKU) - argument = ["-v", "DEVICE_METADATA[\'localhost\'][\'hwsku\']", "-m", self.sample_graph] + argument = "-v \"DEVICE_METADATA[\'localhost\'][\'hwsku\']\" -m \"{}\"".format(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", self.sample_graph, "-p", self.sample_port_config, "--print-data"] + argument = "-m \"{}\" -p \"{}\" --print-data".format(self.sample_graph, self.sample_port_config) output = self.run_script(argument) self.assertGreater(len(output.strip()) , 0) - argument = ["-m", self.sample_graph, "--print-data"] + argument = "-m \"{}\" --print-data".format(self.sample_graph) 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): @@ -111,15 +110,15 @@ def test_additional_json_data(self): 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.safe_load(self.run_script(argument)) + argument = '-v yml_item -y ' + os.path.join(self.test_dir, 'test.yml') + output = yaml.load(self.run_script(argument)) self.assertListEqual(output, ['value1', 'value2']) for asic in range(NUM_ASIC): - output = yaml.safe_load(self.run_script_for_asic(argument, asic, self.port_config[asic])) + output = yaml.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): @@ -127,37 +126,37 @@ def test_render_template(self): 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", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORTCHANNEL"] + argument = "-m {} -p {} -n asic0 --var-json \"PORTCHANNEL\"".format(self.sample_graph, self.port_config[0]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'PortChannel0002': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet0', 'Ethernet4'], 'mtu': '9100', 'tpid': '0x8100'}, @@ -165,14 +164,14 @@ def test_frontend_asic_portchannels(self): 'PortChannel4002': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet-BP8', 'Ethernet-BP12'], 'mtu': '9100', 'tpid': '0x8100'}}) def test_backend_asic_portchannels(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "PORTCHANNEL"] + argument = "-m {} -p {} -n asic3 --var-json \"PORTCHANNEL\"".format(self.sample_graph, self.port_config[3]) 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'}, 'PortChannel4014': {'admin_status': 'up', 'min_links': '2', 'members': ['Ethernet-BP392', 'Ethernet-BP396'], 'mtu': '9100', 'tpid': '0x8100'}}) def test_frontend_asic_portchannel_mem(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "-v", "PORTCHANNEL_MEMBER.keys()|list"] + argument = "-m {} -p {} -n asic0 -v \"PORTCHANNEL_MEMBER.keys()|list\"".format(self.sample_graph, self.port_config[0]) output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -180,7 +179,7 @@ def test_frontend_asic_portchannel_mem(self): ) def test_backend_asic_portchannels_mem(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "-v", "PORTCHANNEL_MEMBER.keys()|list"] + argument = "-m {} -p {} -n asic3 -v \"PORTCHANNEL_MEMBER.keys()|list\"".format(self.sample_graph, self.port_config[3]) output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -188,7 +187,7 @@ def test_backend_asic_portchannels_mem(self): ) def test_frontend_asic_portchannel_intf(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "-v", "PORTCHANNEL_INTERFACE.keys()|list"] + argument = "-m {} -p {} -n asic0 -v \"PORTCHANNEL_INTERFACE.keys()|list\"".format(self.sample_graph, self.port_config[0]) output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -196,7 +195,7 @@ def test_frontend_asic_portchannel_intf(self): ) def test_frontend_asic_routerport_intf(self): - argument = ["-m", self.sample_graph1, "-p", self.port_config[0], "-n", "asic0", "-v", "INTERFACE.keys()|list"] + argument = "-m {} -p {} -n asic0 -v \"INTERFACE.keys()|list\"".format(self.sample_graph1, self.port_config[0]) output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -204,7 +203,7 @@ def test_frontend_asic_routerport_intf(self): ) def test_backend_asic_portchannel_intf(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "-v", "PORTCHANNEL_INTERFACE.keys()|list"] + argument = "-m {} -p {} -n asic3 -v \"PORTCHANNEL_INTERFACE.keys()|list\"".format(self.sample_graph, self.port_config[3]) output = self.run_script(argument) self.assertEqual( utils.liststr_to_dict(output.strip()), @@ -212,7 +211,7 @@ def test_backend_asic_portchannel_intf(self): ) def test_frontend_asic_ports(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORT"] + argument = "-m {} -p {} -n asic0 --var-json \"PORT\"".format(self.sample_graph, self.port_config[0]) 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" }, @@ -225,7 +224,7 @@ def test_frontend_asic_ports(self): "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", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "PORT"] + argument = "-m {} -p {} -n asic0 --var-json \"PORT\"".format(self.sample_graph, self.port_config[0]) 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" }, @@ -238,7 +237,7 @@ def test_frontend_asic_ports_config_db(self): "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", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "DEVICE_NEIGHBOR"] + argument = "-m {} -p {} -n asic0 --var-json \"DEVICE_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'Ethernet0': {'name': '01T2', 'port': 'Ethernet1'}, @@ -249,7 +248,7 @@ def test_frontend_asic_device_neigh(self): 'Ethernet-BP8': {'name': 'ASIC3', 'port': 'Eth0-ASIC3'}}) def test_frontend_asic_device_neigh_metadata(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "DEVICE_NEIGHBOR_METADATA"] + argument = "-m {} -p {} -n asic0 --var-json \"DEVICE_NEIGHBOR_METADATA\"".format(self.sample_graph, self.port_config[0]) output = json.loads(self.run_script(argument)) print(output) self.assertDictEqual(output, \ @@ -258,7 +257,7 @@ def test_frontend_asic_device_neigh_metadata(self): '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", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "DEVICE_NEIGHBOR"] + argument = "-m {} -p {} -n asic3 --var-json \"DEVICE_NEIGHBOR\"".format(self.sample_graph, self.port_config[3]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, \ {'Ethernet-BP396': {'name': 'ASIC1', 'port': 'Eth7-ASIC1'}, @@ -267,7 +266,7 @@ def test_backend_asic_device_neigh(self): 'Ethernet-BP388': {'name': 'ASIC0', 'port': 'Eth7-ASIC0'}}) def test_backend_device_neigh_metadata(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "DEVICE_NEIGHBOR_METADATA"] + argument = "-m {} -p {} -n asic3 --var-json \"DEVICE_NEIGHBOR_METADATA\"".format(self.sample_graph, self.port_config[3]) output = json.loads(self.run_script(argument)) print(output) self.assertDictEqual(output, \ @@ -275,28 +274,28 @@ def test_backend_device_neigh_metadata(self): '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", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "BGP_NEIGHBOR"] + argument = "-m {} -p {} -n asic0 --var-json \"BGP_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) 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", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "BGP_INTERNAL_NEIGHBOR"] + argument = "-m {} -p {} -n asic0 --var-json \"BGP_INTERNAL_NEIGHBOR\"".format(self.sample_graph, self.port_config[0]) 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", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "BGP_INTERNAL_NEIGHBOR"] + argument = "-m {} -p {} -n asic3 --var-json \"BGP_INTERNAL_NEIGHBOR\"".format(self.sample_graph, self.port_config[3]) 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", self.sample_graph, "--var-json", "DEVICE_METADATA"] + argument = "-m {} --var-json DEVICE_METADATA".format(self.sample_graph) 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) @@ -310,7 +309,7 @@ def test_device_asic_metadata(self): self.assertEqual(output['localhost']['deployment_id'], "1") def test_global_asic_acl(self): - argument = ["-m", self.sample_graph, "-p", self.sample_port_config, "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} --var-json \"ACL_TABLE\"".format(self.sample_graph, self.sample_port_config) output = json.loads(self.run_script(argument)) exp = {\ 'SNMP_ACL': {'policy_desc': 'SNMP_ACL', 'type': 'CTRLPLANE', 'stage': 'ingress', 'services': ['SNMP']}, @@ -327,7 +326,7 @@ def test_global_asic_acl(self): self.assertDictEqual(output, exp) def test_global_asic_acl1(self): - argument = ["-m", self.sample_graph1, "-p", self.sample_port_config, "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.sample_port_config) self.maxDiff = None output = json.loads(self.run_script(argument)) exp = {\ @@ -344,7 +343,7 @@ def test_global_asic_acl1(self): self.assertDictEqual(output, exp) def test_front_end_asic_acl(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} -n asic0 --var-json \"ACL_TABLE\"".format(self.sample_graph, self.port_config[0]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ 'DATAACL': {'policy_desc': 'DATAACL', 'ports': ['PortChannel0002'], 'stage': 'ingress', 'type': 'L3'}, @@ -354,7 +353,7 @@ def test_front_end_asic_acl(self): 'SSH_ONLY': {'policy_desc': 'SSH_ONLY', 'services': ['SSH'], 'stage': 'ingress', 'type': 'CTRLPLANE'}}) def test_front_end_asic_acl1(self): - argument = ["-m", self.sample_graph1, "-p", self.port_config[0], "-n", "asic0", "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} -n asic0 --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.port_config[0]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ 'EVERFLOW': {'policy_desc': 'EVERFLOW', 'ports': ['Ethernet0','Ethernet4'], 'stage': 'ingress', 'type': 'MIRROR'}, @@ -364,18 +363,18 @@ def test_front_end_asic_acl1(self): def test_back_end_asic_acl(self): - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} -n asic3 --var-json \"ACL_TABLE\"".format(self.sample_graph, self.port_config[3]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {}) def test_back_end_asic_acl1(self): - argument = ["-m", self.sample_graph1, "-p", self.port_config[3], "-n", "asic3", "--var-json", "ACL_TABLE"] + argument = "-m {} -p {} -n asic3 --var-json \"ACL_TABLE\"".format(self.sample_graph1, self.port_config[3]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {}) def test_loopback_intfs(self): - argument = ["-m", self.sample_graph, "-p", self.sample_port_config, "--var-json", "LOOPBACK_INTERFACE"] + argument = "-m {} -p {} --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.sample_port_config) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ "Loopback0": {}, @@ -383,7 +382,7 @@ def test_loopback_intfs(self): "Loopback0|FC00:1::32/128": {}}) # The asic configuration should have 2 loopback interfaces - argument = ["-m", self.sample_graph, "-p", self.port_config[0], "-n", "asic0", "--var-json", "LOOPBACK_INTERFACE"] + argument = "-m {} -p {} -n asic0 --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.port_config[0]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, { \ "Loopback0": {}, @@ -393,7 +392,7 @@ def test_loopback_intfs(self): "Loopback4096|8.0.0.0/32": {}, "Loopback4096|FD00:1::32/128": {}}) - argument = ["-m", self.sample_graph, "-p", self.port_config[3], "-n", "asic3", "--var-json", "LOOPBACK_INTERFACE"] + argument = "-m {} -p {} -n asic3 --var-json \"LOOPBACK_INTERFACE\"".format(self.sample_graph, self.port_config[3]) output = json.loads(self.run_script(argument)) self.assertDictEqual(output, {\ "Loopback0": {}, @@ -426,7 +425,9 @@ def test_buffers_multi_asic_template(self): ) # asic0 - mix of front end and back end ports shutil.copy2(buffer_template, device_config_dir) - argument = ["-m", self.sample_graph, "-p", port_config_ini_asic0, "-n", "asic0", "-t", device_buffer_template] + argument = "-m {} -p {} -n asic0 -t {}".format( + self.sample_graph, port_config_ini_asic0, device_buffer_template + ) output = json.loads(self.run_script(argument)) os.remove(os.path.join(device_config_dir, "buffers_config.j2")) self.assertDictEqual(