feat: support namespace arg for show mac

Author: cyw233

doc/Command-Reference.md

@@ -11827,12 +11827,13 @@ This command displays the MAC (FDB) entries either in full or partial as given b
 4) show mac -a <mac-address> - display the MACs that match a specific mac-address
 5) show mac -t <type> - display the MACs that match a specific type (static/dynamic)
 6) show mac -c - display the count of MAC addresses
+7) show mac -n <namespace> - display the MACs that belong to a specific namespace
 
 To show the default MAC address aging time on the switch.
 
 - Usage:
   ```
-  show mac [-v <vlan_id>] [-p <port_name>] [-a <mac_address>] [-t <type>] [-c]
+  show mac [-v <vlan_id>] [-p <port_name>] [-a <mac_address>] [-t <type>] [-c] [-n <namespace>]
   ```
 
 - Example:
@@ -11931,6 +11932,14 @@ Optionally, you can specify a VLAN ID or interface name or type or mac-address i
   admin@sonic:~$ show mac -c
   Total number of entries 18
   ```
+  ```
+  admin@sonic:~$ show mac -n asic0
+  No.    Vlan  MacAddress         Port           Type
+  -----  ------  -----------------  -----------  -------
+    2    1000  50:96:23:AD:F1:65  Ethernet192    Static
+    2    1000  C6:C9:5E:AE:24:42  Ethernet192    Static
+  Total number of entries 2
+  ```
 
 **show mac aging-time**
 

scripts/fdbshow

@@ -32,14 +32,18 @@ import sys
 import os
 import re
 
+from utilities_common.general import load_db_config
+
 # mock the redis for unit test purposes #
 try: # pragma: no cover
-    if os.environ["UTILITIES_UNIT_TESTING"] == "1": 
+    if os.environ["UTILITIES_UNIT_TESTING"] == "1":
         modules_path = os.path.join(os.path.dirname(__file__), "..")
         test_path = os.path.join(modules_path, "tests")
         sys.path.insert(0, modules_path)
         sys.path.insert(0, test_path)
-        import mock_tables.dbconnector
+        from tests.mock_tables import dbconnector
+
+    if os.environ["FDBSHOW_UNIT_TESTING"] == "1":
         mock_variants = { "1": 'asic_db',
                           "2": 'asic_db_def_vlan',
                           "3": 'asic_db_no_fdb',
@@ -50,23 +54,34 @@ try: # pragma: no cover
         mock_db_path = os.path.join(test_path, "fdbshow_input")
         file_name = mock_variants[os.environ["FDBSHOW_MOCK"]]
         jsonfile_asic = os.path.join(mock_db_path, file_name)
-        mock_tables.dbconnector.dedicated_dbs['ASIC_DB'] = jsonfile_asic
+        dbconnector.dedicated_dbs['ASIC_DB'] = jsonfile_asic
         jsonfile_counters = os.path.join(mock_db_path, 'counters_db')
-        mock_tables.dbconnector.dedicated_dbs['COUNTERS_DB'] = jsonfile_counters
+        dbconnector.dedicated_dbs['COUNTERS_DB'] = jsonfile_counters
+
+    if os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] == "multi_asic":
+        import tests.mock_tables.mock_multi_asic
+        dbconnector.load_namespace_config()
 except KeyError: # pragma: no cover
     pass
 
-from sonic_py_common import port_util
-from swsscommon.swsscommon import SonicV2Connector
+from sonic_py_common import port_util, multi_asic
+from swsscommon.swsscommon import SonicV2Connector, SonicDBConfig
 from tabulate import tabulate
 
 class FdbShow(object):
 
     HEADER = ['No.', 'Vlan', 'MacAddress', 'Port', 'Type']
 
-    def __init__(self):
+    def __init__(self, namespace=None):
         super(FdbShow,self).__init__()
-        self.db = SonicV2Connector(host="127.0.0.1")
+        if namespace is not None:
+            if not SonicDBConfig.isGlobalInit():
+                SonicDBConfig.load_sonic_global_db_config()
+
+            self.db = SonicV2Connector(use_unix_socket_path=True, namespace=namespace)
+        else:
+            self.db = SonicV2Connector(host="127.0.0.1")
+
         self.if_name_map, \
         self.if_oid_map = port_util.get_interface_oid_map(self.db)
         self.if_br_oid_map = port_util.get_bridge_port_map(self.db)
@@ -169,7 +184,7 @@ class FdbShow(object):
 
         print("Total number of entries {0}".format(len(self.bridge_mac_list)))
 
-    def validate_params(self, vlan, port, address, entry_type):
+    def validate_params(self, vlan, port, address, entry_type, namespace):
         if vlan is not None:
             if not vlan.isnumeric():
                 print("Error: Invalid vlan id {0}".format(vlan))
@@ -194,6 +209,15 @@ class FdbShow(object):
             print("Error: Invalid type {0}". format(entry_type))
             return False
 
+        if namespace is not None:
+            if not multi_asic.is_multi_asic():
+                print("Error: Namespace is not supported in single asic")
+                return False
+
+            if namespace not in multi_asic.get_namespace_list():
+                print("Error: Invalid namespace {0}".format(namespace))
+                return False
+
         return True
 
 def main():
@@ -205,11 +229,12 @@ def main():
     parser.add_argument('-a', '--address', type=str, help='FDB display based on specific mac address', default=None)
     parser.add_argument('-t', '--type', type=str, help='FDB display of specific type of mac address', default=None)
     parser.add_argument('-c', '--count', action='store_true', help='FDB display count of mac address')
+    parser.add_argument('-n', '--namespace', type=str, help='Namespace name or all', default=None)
     args = parser.parse_args()
 
     try:
-        fdb = FdbShow()
-        if not fdb.validate_params(args.vlan, args.port, args.address, args.type):
+        fdb = FdbShow(namespace=args.namespace)
+        if not fdb.validate_params(args.vlan, args.port, args.address, args.type, args.namespace):
            sys.exit(1)
 
         fdb.display(args.vlan, args.port, args.address, args.type, args.count)

show/main.py

@@ -1121,7 +1121,15 @@ def pwm_headroom_pool(namespace):
 @click.option('-t', '--type')
 @click.option('-c', '--count', is_flag=True)
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
-def mac(ctx, vlan, port, address, type, count, verbose):
+@click.option('-n',
+              '--namespace',
+              'namespace',
+              default=None,
+              type=str,
+              show_default=True,
+              help='Namespace name or all',
+              callback=multi_asic_util.multi_asic_namespace_validation_callback)
+def mac(ctx, vlan, port, address, type, count, verbose, namespace):
     """Show MAC (FDB) entries"""
 
     if ctx.invoked_subcommand is not None:
@@ -1144,6 +1152,9 @@ def mac(ctx, vlan, port, address, type, count, verbose):
     if count:
         cmd += ["-c"]
 
+    if namespace is not None:
+        cmd += ['-n', str(namespace)]
+
     run_command(cmd, display_cmd=verbose)
 
 @mac.command('aging-time')

tests/fdbshow_input/fdbshow_masic_test_vectors.py

@@ -0,0 +1,14 @@
+show_mac_masic_asic0_output = """\
+  No.    Vlan  MacAddress         Port       Type
+-----  ------  -----------------  ---------  -------
+    1       2  11:22:33:44:55:66  Ethernet0  Dynamic
+Total number of entries 1
+"""
+
+test_data = {
+    "show_mac_masic_asic0": {
+        "cmd": "mac",
+        "args": "-n asic0",
+        "expected_output": show_mac_masic_asic0_output,
+    }
+}

tests/fdbshow_test.py

@@ -148,10 +148,12 @@ def setup_class(cls):
         print("SETUP")
         os.environ["PATH"] += os.pathsep + scripts_path
         os.environ["UTILITIES_UNIT_TESTING"] = "1"
+        os.environ["FDBSHOW_UNIT_TESTING"] = "1"
         yield
         print("TEARDOWN")
         os.environ["PATH"] = os.pathsep.join(os.environ["PATH"].split(os.pathsep)[:-1])
         os.environ["UTILITIES_UNIT_TESTING"] = "0"
+        os.environ["FDBSHOW_UNIT_TESTING"] = "0"
 
     @pytest.fixture(scope="function", autouse=True)
     def setUp(self):

tests/mock_tables/asic0/asic_db.json

@@ -2,5 +2,19 @@
     "ASIC_STATE:SAI_OBJECT_TYPE_SWITCH:oid:0x21000000000000": {
         "SAI_SWITCH_ATTR_INIT_SWITCH": "true",
         "SAI_SWITCH_ATTR_SRC_MAC_ADDRESS": "DE:AD:BE:EF:CA:FE"
+    },
+    "ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT:oid:0x3a0000000005cb": {
+        "SAI_BRIDGE_PORT_ATTR_TYPE": "SAI_BRIDGE_PORT_TYPE_PORT",
+        "SAI_BRIDGE_PORT_ATTR_PORT_ID": "oid:0x1000000000002",
+        "SAI_BRIDGE_PORT_ATTR_ADMIN_STATE": "true",
+        "SAI_BRIDGE_PORT_ATTR_FDB_LEARNING_MODE": "SAI_BRIDGE_PORT_FDB_LEARNING_MODE_HW"
+    },
+    "ASIC_STATE:SAI_OBJECT_TYPE_FDB_ENTRY:{\"bvid\":\"oid:0x260000000005c5\",\"mac\":\"11:22:33:44:55:66\",\"switch_id\":\"oid:0x21000000000000\"}": {
+        "SAI_FDB_ENTRY_ATTR_BRIDGE_PORT_ID": "oid:0x3a0000000005cb",
+        "SAI_FDB_ENTRY_ATTR_TYPE": "SAI_FDB_ENTRY_TYPE_DYNAMIC",
+        "SAI_FDB_ENTRY_ATTR_PACKET_ACTION": "SAI_PACKET_ACTION_FORWARD"
+    },
+    "ASIC_STATE:SAI_OBJECT_TYPE_VLAN:oid:0x260000000005c5": {
+        "SAI_VLAN_ATTR_VLAN_ID": "2"
     }
 }

tests/multi_asic_fdbshow_test.py

@@ -0,0 +1,50 @@
+import os
+import sys
+
+from click.testing import CliRunner
+
+from tests.fdbshow_input.fdbshow_masic_test_vectors import show_mac_masic_asic0_output, test_data
+from tests.utils import get_result_and_return_code
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(test_path)
+scripts_path = os.path.join(modules_path, "scripts")
+sys.path.insert(0, test_path)
+sys.path.insert(0, modules_path)
+
+
+class TestFdbshowMultiAsic(object):
+    @classmethod
+    def setup_class(cls):
+        os.environ["PATH"] += os.pathsep + scripts_path
+        os.environ['UTILITIES_UNIT_TESTING'] = "1"
+        os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] = "multi_asic"
+        print("SETUP")
+
+    def test_show_mac_masic_asic0(self):
+        return_code, result = get_result_and_return_code([
+            'fdbshow', '-n', 'asic0'
+        ])
+        print("return_code: {}".format(return_code))
+        print("result: {}".format(result))
+        assert return_code == 0
+        assert result == show_mac_masic_asic0_output
+
+        self.command_executor(test_data["show_mac_masic_asic0"])
+
+    @classmethod
+    def teardown_class(cls):
+        os.environ["PATH"] = os.pathsep.join(os.environ["PATH"].split(os.pathsep)[:-1])
+        os.environ['UTILITIES_UNIT_TESTING'] = "0"
+        os.environ["UTILITIES_UNIT_TESTING_TOPOLOGY"] = ""
+        print("TEARDOWN")
+
+    @staticmethod
+    def command_executor(test_case_data):
+        import show.main as show
+        runner = CliRunner()
+        result = runner.invoke(show.cli.commands["mac"], test_case_data["args"])
+        print("result.exit_code: {}".format(result.exit_code))
+        print("result.output: {}".format(result.output))
+        assert result.exit_code == 0
+        assert result.output == test_case_data["expected_output"]