[pytest][qos][config] Added pytests for "config qos reload" commands" (sonic-net#1346)

Added pytest to verify QoS config generation on multi ASIC platform.

Author: smaheshm

config/main.py

@@ -37,6 +37,21 @@
 from . import vxlan
 from .config_mgmt import ConfigMgmtDPB
 
+# mock masic APIs for unit test
+try:
+    if os.environ["UTILITIES_UNIT_TESTING"] == "1" or os.environ["UTILITIES_UNIT_TESTING"] == "2":
+        modules_path = os.path.join(os.path.dirname(__file__), "..")
+        tests_path = os.path.join(modules_path, "tests")
+        sys.path.insert(0, modules_path)
+        sys.path.insert(0, tests_path)
+        import mock_tables.dbconnector
+    if os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] == "multi_asic":
+        import mock_tables.mock_multi_asic
+        mock_tables.dbconnector.load_namespace_config()
+except KeyError:
+    pass
+
+
 CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help', '-?'])
 
 SONIC_GENERATED_SERVICE_PATH = '/etc/sonic/generated_services.conf'
@@ -1756,12 +1771,24 @@ def _update_buffer_calculation_model(config_db, model):
 @qos.command('reload')
 @click.pass_context
 @click.option('--no-dynamic-buffer', is_flag=True, help="Disable dynamic buffer calculation")
-def reload(ctx, no_dynamic_buffer):
+@click.option(
+    '--json-data', type=click.STRING,
+    help="json string with additional data, valid with --dry-run option"
+)
+@click.option(
+    '--dry_run', type=click.STRING,
+    help="Dry run, writes config to the given file"
+)
+def reload(ctx, no_dynamic_buffer, dry_run, json_data):
     """Reload QoS configuration"""
     log.log_info("'qos reload' executing...")
     _clear_qos()
 
     _, hwsku_path = device_info.get_paths_to_platform_and_hwsku_dirs()
+    sonic_version_file = device_info.get_sonic_version_file()
+    from_db = "-d --write-to-db"
+    if dry_run:
+        from_db = "--additional-data \'{}\'".format(json_data) if json_data else ""
 
     namespace_list = [DEFAULT_NAMESPACE]
     if multi_asic.get_num_asics() > 1:
@@ -1798,17 +1825,17 @@ def reload(ctx, no_dynamic_buffer):
             buffer_template_file = os.path.join(hwsku_path, asic_id_suffix, "buffers.json.j2")
             if asic_type in vendors_supporting_dynamic_buffer:
                 buffer_model_updated |= _update_buffer_calculation_model(config_db, "traditional")
+
         if os.path.isfile(buffer_template_file):
-            qos_template_file = os.path.join(hwsku_path, asic_id_suffix, "qos.json.j2")
+            qos_template_file = os.path.join(
+                hwsku_path, asic_id_suffix, "qos.json.j2"
+            )
             if os.path.isfile(qos_template_file):
                 cmd_ns = "" if ns is DEFAULT_NAMESPACE else "-n {}".format(ns)
-                sonic_version_file = os.path.join('/', "etc", "sonic", "sonic_version.yml")
-                command = "{} {} -d -t {},config-db -t {},config-db -y {} --write-to-db".format(
-                    SONIC_CFGGEN_PATH,
-                    cmd_ns,
-                    buffer_template_file,
-                    qos_template_file,
-                    sonic_version_file
+                fname = "{}{}".format(dry_run, asic_id_suffix) if dry_run else "config-db"
+                command = "{} {} {} -t {},{} -t {},{} -y {}".format(
+                    SONIC_CFGGEN_PATH, cmd_ns, from_db, buffer_template_file,
+                    fname, qos_template_file, fname, sonic_version_file
                 )
                 # Apply the configurations only when both buffer and qos
                 # configuration files are present

pfcwd/main.py

@@ -1,4 +1,5 @@
 import os
+import imp
 import sys
 
 import click
@@ -21,6 +22,7 @@
         import mock_tables.dbconnector
     if os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] == "multi_asic":
         import mock_tables.mock_multi_asic
+        imp.reload(mock_tables.mock_multi_asic)
         mock_tables.dbconnector.load_namespace_config()
 
 except KeyError:

scripts/intfutil

@@ -22,7 +22,7 @@ try:
     if os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] == "multi_asic":
         import mock_tables.mock_multi_asic
         mock_tables.dbconnector.load_namespace_config()
-           
+
 except KeyError:
     pass
 
@@ -461,7 +461,7 @@ class IntfDescription(object):
             self.intf_name = intf_name
 
     def display_intf_description(self):
-        
+
         self.get_intf_description()
 
         # Sorting and tabulating the result table.

tests/config_test.py

@@ -1,10 +1,14 @@
+import filecmp
+import imp
 import os
 import traceback
+import json
 from unittest import mock
 
 import click
 from click.testing import CliRunner
 
+from sonic_py_common import device_info
 from utilities_common.db import Db
 
 load_minigraph_command_output="""\
@@ -61,8 +65,10 @@ class TestLoadMinigraph(object):
     def setup_class(cls):
         os.environ['UTILITIES_UNIT_TESTING'] = "1"
         print("SETUP")
+        import config.main
+        imp.reload(config.main)
 
-    def test_load_minigraph(self, get_cmd_module, setup_single_broacom_asic):
+    def test_load_minigraph(self, get_cmd_module, setup_single_broadcom_asic):
         with mock.patch("utilities_common.cli.run_command", mock.MagicMock(side_effect=mock_run_command_side_effect)) as mock_run_command:
             (config, show) = get_cmd_module
             runner = CliRunner()
@@ -74,7 +80,7 @@ def test_load_minigraph(self, get_cmd_module, setup_single_broacom_asic):
             assert "\n".join([l.rstrip() for l in result.output.split('\n')]) == load_minigraph_command_output
             assert mock_run_command.call_count == 38
 
-    def test_load_minigraph_with_disabled_telemetry(self, get_cmd_module, setup_single_broacom_asic):
+    def test_load_minigraph_with_disabled_telemetry(self, get_cmd_module, setup_single_broadcom_asic):
         with mock.patch("utilities_common.cli.run_command", mock.MagicMock(side_effect=mock_run_command_side_effect)) as mock_run_command:
             (config, show) = get_cmd_module
             db = Db()
@@ -95,3 +101,98 @@ def test_load_minigraph_with_disabled_telemetry(self, get_cmd_module, setup_sing
     def teardown_class(cls):
         os.environ['UTILITIES_UNIT_TESTING'] = "0"
         print("TEARDOWN")
+
+
+class TestConfigQos(object):
+    @classmethod
+    def setup_class(cls):
+        print("SETUP")
+        os.environ['UTILITIES_UNIT_TESTING'] = "2"
+        import config.main
+        imp.reload(config.main)
+
+    def test_qos_reload_single(
+            self, get_cmd_module, setup_qos_mock_apis,
+            setup_single_broadcom_asic
+        ):
+        (config, show) = get_cmd_module
+        runner = CliRunner()
+        output_file = os.path.join(os.sep, "tmp", "qos_config_output.json")
+        print("Saving output in {}".format(output_file))
+        try:
+            os.remove(output_file)
+        except OSError:
+            pass
+        json_data = '{"DEVICE_METADATA": {"localhost": {}}}'
+        result = runner.invoke(
+            config.config.commands["qos"],
+            ["reload", "--dry_run", output_file, "--json-data", json_data]
+        )
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code == 0
+
+        cwd = os.path.dirname(os.path.realpath(__file__))
+        expected_result = os.path.join(
+            cwd, "qos_config_input", "config_qos.json"
+        )
+        assert filecmp.cmp(output_file, expected_result, shallow=False)
+
+    @classmethod
+    def teardown_class(cls):
+        print("TEARDOWN")
+        os.environ['UTILITIES_UNIT_TESTING'] = "0"
+
+
+class TestConfigQosMasic(object):
+    @classmethod
+    def setup_class(cls):
+        print("SETUP")
+        os.environ['UTILITIES_UNIT_TESTING'] = "2"
+        os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] = "multi_asic"
+        import config.main
+        imp.reload(config.main)
+
+    def test_qos_reload_masic(
+            self, get_cmd_module, setup_qos_mock_apis,
+            setup_multi_broadcom_masic
+        ):
+        (config, show) = get_cmd_module
+        runner = CliRunner()
+        output_file = os.path.join(os.sep, "tmp", "qos_config_output.json")
+        print("Saving output in {}<0,1,2..>".format(output_file))
+        num_asic = device_info.get_num_npus()
+        for asic in range(num_asic):
+            try:
+                file = "{}{}".format(output_file, asic)
+                os.remove(file)
+            except OSError:
+                pass
+        json_data = '{"DEVICE_METADATA": {"localhost": {}}}'
+        result = runner.invoke(
+            config.config.commands["qos"],
+            ["reload", "--dry_run", output_file, "--json-data", json_data]
+        )
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code == 0
+
+        cwd = os.path.dirname(os.path.realpath(__file__))
+
+        for asic in range(num_asic):
+            expected_result = os.path.join(
+                cwd, "qos_config_input", str(asic), "config_qos.json"
+            )
+            file = "{}{}".format(output_file, asic)
+            assert filecmp.cmp(file, expected_result, shallow=False)
+
+    @classmethod
+    def teardown_class(cls):
+        print("TEARDOWN")
+        os.environ['UTILITIES_UNIT_TESTING'] = "0"
+        os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] = ""
+        # change back to single asic config
+        from .mock_tables import dbconnector
+        from .mock_tables import mock_single_asic
+        imp.reload(mock_single_asic)
+        dbconnector.load_namespace_config()

tests/conftest.py

@@ -56,18 +56,47 @@ def get_cmd_module():
 
     return (config, show)
 
+def set_mock_apis():
+    import config.main as config
+    cwd = os.path.dirname(os.path.realpath(__file__))
+    config.asic_type = mock.MagicMock(return_value="broadcom")
+    config._get_device_type = mock.MagicMock(return_value="ToRRouter")
+
+@pytest.fixture
+def setup_qos_mock_apis():
+    cwd = os.path.dirname(os.path.realpath(__file__))
+    device_info.get_paths_to_platform_and_hwsku_dirs = mock.MagicMock(
+        return_value=(
+            os.path.join(cwd, "."), os.path.join(cwd, "qos_config_input")
+        )
+    )
+    device_info.get_sonic_version_file = mock.MagicMock(
+        return_value=os.path.join(cwd, "qos_config_input/sonic_version.yml")
+    )
 
 @pytest.fixture
-def setup_single_broacom_asic():
+def setup_single_broadcom_asic():
     import config.main as config
     import show.main as show
 
+    set_mock_apis()
     device_info.get_num_npus = mock.MagicMock(return_value=1)
     config._get_sonic_generated_services = \
         mock.MagicMock(return_value=(generated_services_list, []))
 
-    config.asic_type = mock.MagicMock(return_value="broadcom")
-    config._get_device_type = mock.MagicMock(return_value="ToRRouter")
+
+@pytest.fixture
+def setup_multi_broadcom_masic():
+    import config.main as config
+    import show.main as show
+
+    set_mock_apis()
+    device_info.get_num_npus = mock.MagicMock(return_value=2)
+
+    yield
+
+    device_info.get_num_npus = mock.MagicMock(return_value=1)
+
 
 @pytest.fixture
 def setup_t1_topo():

tests/crm_test.py

@@ -1,3 +1,4 @@
+import imp
 import os
 import sys
 from importlib import reload
@@ -1574,5 +1575,7 @@ def teardown_class(cls):
         print("TEARDOWN")
         os.environ["UTILITIES_UNIT_TESTING"] = "0"
         os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] = ""
+        from .mock_tables import dbconnector
         from .mock_tables import mock_single_asic
-        reload(mock_single_asic)
+        imp.reload(mock_single_asic)
+        dbconnector.load_namespace_config()

tests/mock_tables/mock_multi_asic.py

@@ -14,6 +14,8 @@ def mock_is_multi_asic():
 def mock_get_namespace_list(namespace=None):
     return ['asic0', 'asic1']
 
+
 multi_asic.get_num_asics = mock_get_num_asics
 multi_asic.is_multi_asic = mock_is_multi_asic
 multi_asic.get_namespace_list = mock_get_namespace_list
+multi_asic.get_namespaces_from_linux = mock_get_namespace_list

tests/qos_config_input/0/buffers.json.j2

@@ -0,0 +1,3 @@
+{%- set default_topo = 't1' %}
+{%- include 'buffers_config.j2' %}
+