[sfpshow/sfputil] Enhance sfpshow and sfputil to behavior correctly on RJ45 ports

scripts/sfpshow

@@ -214,6 +214,8 @@ QSFP_DD_DOM_VALUE_UNIT_MAP = {
     'voltage': 'Volts'
 }
 
+RJ45_PORT_TYPE = 'RJ45'
+
 
 def display_invalid_intf_eeprom(intf_name):
     output = intf_name + ': SFP EEPROM Not detected\n'
@@ -391,15 +393,18 @@ class SFPShow(object):
         output = ''
 
         sfp_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interface_name))
-        output = interface_name + ': ' + 'SFP EEPROM detected' + '\n'
-        sfp_info_output = self.convert_sfp_info_to_output_string(sfp_info_dict)
-        output += sfp_info_output
+        if sfp_info_dict['type'] == RJ45_PORT_TYPE:
+            output = interface_name + ': ' + 'SFP EEPROM is not applicable for RJ45 port' + '\n'
+        else:
+            output = interface_name + ': ' + 'SFP EEPROM detected' + '\n'
+            sfp_info_output = self.convert_sfp_info_to_output_string(sfp_info_dict)
+            output += sfp_info_output
 
-        if dump_dom:
-            sfp_type = sfp_info_dict['type']
-            dom_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_DOM_SENSOR|{}'.format(interface_name))
-            dom_output = self.convert_dom_to_output_string(sfp_type, dom_info_dict)
-            output += dom_output
+            if dump_dom:
+                sfp_type = sfp_info_dict['type']
+                dom_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_DOM_SENSOR|{}'.format(interface_name))
+                dom_output = self.convert_dom_to_output_string(sfp_type, dom_info_dict)
+                output += dom_output
 
         return output
 
@@ -427,26 +432,34 @@ class SFPShow(object):
 
                     self.output += '\n'
 
+    def convert_presence_info_to_cli_output_string(self, state_db, interface_name):
+        sfp_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interface_name))
+        if not sfp_info_dict:
+            status_string = 'Not present'
+        elif sfp_info_dict.get('type') == RJ45_PORT_TYPE:
+            status = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_STATUS|{}'.format(interface_name))
+            status_string = status.get('error')
+            if status_string == 'N/A':
+                status_string = 'Present'
+        else:
+            status_string = 'Present'
+
+        return status_string
+
     @multi_asic_util.run_on_multi_asic
     def get_presence(self):
         port_table = []
 
         if self.intf_name is not None:
-            presence = self.db.exists(self.db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(self.intf_name))
-            if presence:
-                port_table.append((self.intf_name, 'Present'))
-            else:
-                port_table.append((self.intf_name, 'Not present'))
+            presence_str = self.convert_presence_info_to_cli_output_string(self.db, self.intf_name)
+            port_table.append((self.intf_name, presence_str))
         else:
             port_table_keys = self.db.keys(self.db.APPL_DB, "PORT_TABLE:*")
             for i in port_table_keys:
                 key = re.split(':', i, maxsplit=1)[-1].strip()
                 if key and key.startswith(front_panel_prefix()) and not key.startswith((backplane_prefix(), inband_prefix(), recirc_prefix())):
-                    presence = self.db.exists(self.db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(key))
-                    if presence:
-                        port_table.append((key, 'Present'))
-                    else:
-                        port_table.append((key, 'Not present'))
+                    presence_str = self.convert_presence_info_to_cli_output_string(self.db, key)
+                    port_table.append((key, presence_str))
 
         self.table += port_table
 

sfputil/main.py

@@ -225,6 +225,7 @@
     'voltage': 'Volts'
 }
 
+RJ45_PORT_TYPE = 'RJ45'
 
 # Global platform-specific Chassis class instance
 platform_chassis = None
@@ -247,6 +248,30 @@ def is_sfp_present(port_name):
 
     return bool(presence)
 
+
+# Determine whether it is a RJ45 port
+def is_rj45_port_from_db(port_name, db):
+    intf_type = db.get(db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(port_name), 'type')
+    return intf_type == RJ45_PORT_TYPE
+
+
+def is_rj45_port_from_api(port_name):
+    physical_port = logical_port_to_physical_port_index(port_name)
+    sfp = platform_chassis.get_sfp(physical_port)
+
+    try:
+        port_type = sfp.get_transceiver_info()['type']
+    except NotImplementedError:
+        click.echo("Not able to judge the port type due to get_transceiver_info not implemented!", err=True)
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+
+    return port_type == RJ45_PORT_TYPE
+
+
+def skip_if_port_is_rj45(port_name): 
+    if is_rj45_port_from_api(port_name): 
+        click.echo("This functionality is not applicable for RJ45 port {}.".format(port_name)) 
+        sys.exit(EXIT_FAIL)
 # ========================== Methods for formatting output ==========================
 
 # Convert dict values to cli output string
@@ -570,6 +595,11 @@ def eeprom(port, dump_dom, namespace):
         for physical_port in physical_port_list:
             port_name = get_physical_port_name(logical_port_name, i, ganged)
 
+            if is_rj45_port_from_api(port_name):
+                output += "{}: SFP EEPROM is not applicable for RJ45 port\n".format(port_name)
+                output += '\n'
+                continue
+
             try:
                 presence = platform_chassis.get_sfp(physical_port).get_presence()
             except NotImplementedError:
@@ -631,6 +661,13 @@ def presence(port):
 
         logical_port_list = [port]
 
+    state_db = SonicV2Connector(host='127.0.0.1')
+    if state_db is not None:
+        state_db.connect(state_db.STATE_DB)
+    else:
+        click.echo("Failed to connect to STATE_DB")
+        return
+
     for logical_port_name in logical_port_list:
         ganged = False
         i = 1
@@ -646,13 +683,19 @@ def presence(port):
         for physical_port in physical_port_list:
             port_name = get_physical_port_name(logical_port_name, i, ganged)
 
-            try:
-                presence = platform_chassis.get_sfp(physical_port).get_presence()
-            except NotImplementedError:
-                click.echo("This functionality is currently not implemented for this platform")
-                sys.exit(ERROR_NOT_IMPLEMENTED)
+            if is_rj45_port_from_db(port_name, state_db):
+                status = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_STATUS|{}'.format(port_name))
+                status_string = status.get('error')
+                if status_string == 'N/A':
+                    status_string = 'Present'
+            else:
+                try:
+                    presence = platform_chassis.get_sfp(physical_port).get_presence()
+                except NotImplementedError:
+                    click.echo("This functionality is currently not implemented for this platform")
+                    sys.exit(ERROR_NOT_IMPLEMENTED)
 
-            status_string = "Present" if presence else "Not present"
+                status_string = "Present" if presence else "Not present"
             output_table.append([port_name, status_string])
 
             i += 1
@@ -723,7 +766,10 @@ def fetch_error_status_from_platform_api(port):
         physical_port_list = logical_port_name_to_physical_port_list(logical_port_name)
         port_name = get_physical_port_name(logical_port_name, 1, False)
 
-        output.append([port_name, output_dict.get(physical_port_list[0])])
+        if is_rj45_port_from_api(logical_port_name):
+            output.append([port_name, "N/A"])
+        else:
+            output.append([port_name, output_dict.get(physical_port_list[0])])
 
     return output
 
@@ -746,15 +792,18 @@ def fetch_error_status_from_state_db(port, state_db):
     sorted_ports = natsort.natsorted(status)
     output = []
     for port in sorted_ports:
-        statestring = status[port].get('status')
-        description = status[port].get('error')
-        if statestring == '1':
-            description = 'OK'
-        elif statestring == '0':
-            description = 'Unplugged'
-        elif description == 'N/A':
-            log.log_error("Inconsistent state found for port {}: state is {} but error description is N/A".format(port, statestring))
-            description = 'Unknown state: {}'.format(statestring)
+        if is_rj45_port_from_db(port, state_db):
+            description = "N/A"
+        else:
+            statestring = status[port].get('status')
+            description = status[port].get('error')
+            if statestring == '1':
+                description = 'OK'
+            elif statestring == '0':
+                description = 'Unplugged'
+            elif description == 'N/A':
+                log.log_error("Inconsistent state found for port {}: state is {} but error description is N/A".format(port, statestring))
+                description = 'Unknown state: {}'.format(statestring)
 
         output.append([port, description])
 
@@ -819,24 +868,27 @@ def lpmode(port):
             click.echo("Error: No physical ports found for logical port '{}'".format(logical_port_name))
             return
 
-        if len(physical_port_list) > 1:
-            ganged = True
+        if is_rj45_port_from_api(logical_port_name):
+            output_table.append([logical_port_name, "N/A"])
+        else:
+            if len(physical_port_list) > 1:
+                ganged = True
 
-        for physical_port in physical_port_list:
-            port_name = get_physical_port_name(logical_port_name, i, ganged)
+            for physical_port in physical_port_list:
+                port_name = get_physical_port_name(logical_port_name, i, ganged)
 
-            try:
-                lpmode = platform_chassis.get_sfp(physical_port).get_lpmode()
-            except NotImplementedError:
-                click.echo("This functionality is currently not implemented for this platform")
-                sys.exit(ERROR_NOT_IMPLEMENTED)
+                try:
+                    lpmode = platform_chassis.get_sfp(physical_port).get_lpmode()
+                except NotImplementedError:
+                    click.echo("This functionality is currently not implemented for this platform")
+                    sys.exit(ERROR_NOT_IMPLEMENTED)
 
-            if lpmode:
-                output_table.append([port_name, "On"])
-            else:
-                output_table.append([port_name, "Off"])
+                if lpmode:
+                    output_table.append([port_name, "On"])
+                else:
+                    output_table.append([port_name, "Off"])
 
-            i += 1
+                i += 1
 
     click.echo(tabulate(output_table, table_header, tablefmt='simple'))
 
@@ -859,6 +911,10 @@ def fwversion(port_name):
     physical_port = logical_port_to_physical_port_index(port_name)
     sfp = platform_chassis.get_sfp(physical_port)
 
+    if is_rj45_port_from_api(port_name):
+        click.echo("Show firmware version is not applicable for RJ45 port {}.".format(port_name))
+        sys.exit(EXIT_FAIL)
+
     try:
         presence = sfp.get_presence()
     except NotImplementedError:
@@ -894,6 +950,10 @@ def set_lpmode(logical_port, enable):
         click.echo("Error: No physical ports found for logical port '{}'".format(logical_port))
         return
 
+    if is_rj45_port_from_api(logical_port):
+        click.echo("{} low-power mode is not applicable for RJ45 port {}.".format("Enabling" if enable else "Disabling", logical_port))
+        sys.exit(EXIT_FAIL)
+
     if len(physical_port_list) > 1:
         ganged = True
 
@@ -950,6 +1010,10 @@ def reset(port_name):
         click.echo("Error: No physical ports found for logical port '{}'".format(port_name))
         return
 
+    if is_rj45_port_from_api(port_name):
+        click.echo("Reset is not applicable for RJ45 port {}.".format(port_name))
+        sys.exit(EXIT_FAIL)
+
     if len(physical_port_list) > 1:
         ganged = True
 
@@ -1115,6 +1179,8 @@ def run(port_name, mode):
         click.echo("{}: SFP EEPROM not detected\n".format(port_name))
         sys.exit(EXIT_FAIL)
 
+    skip_if_port_is_rj45(port_name)
+
     status = run_firmware(port_name, int(mode))
     if status != 1:
         click.echo('Failed to run firmware in mode={}! CDB status: {}'.format(mode, status))
@@ -1132,6 +1198,8 @@ def commit(port_name):
         click.echo("{}: SFP EEPROM not detected\n".format(port_name))
         sys.exit(EXIT_FAIL)
 
+    skip_if_port_is_rj45(port_name)
+
     status = commit_firmware(port_name)
     if status != 1:
         click.echo('Failed to commit firmware! CDB status: {}'.format(status))
@@ -1152,6 +1220,8 @@ def upgrade(port_name, filepath):
         click.echo("{}: SFP EEPROM not detected\n".format(port_name))
         sys.exit(EXIT_FAIL)
 
+    skip_if_port_is_rj45(port_name)
+
     show_firmware_version(physical_port)
 
     status = download_firmware(port_name, filepath)
@@ -1186,6 +1256,8 @@ def download(port_name, filepath):
        click.echo("{}: SFP EEPROM not detected\n".format(port_name))
        sys.exit(EXIT_FAIL)
 
+    skip_if_port_is_rj45(port_name)
+
     start = time.time()
     status = download_firmware(port_name, filepath)
     if status == 1:
@@ -1206,6 +1278,8 @@ def unlock(port_name, password):
     physical_port = logical_port_to_physical_port_index(port_name)
     sfp = platform_chassis.get_sfp(physical_port)
 
+    skip_if_port_is_rj45(port_name)
+
     if not is_sfp_present(port_name):
        click.echo("{}: SFP EEPROM not detected\n".format(port_name))
        sys.exit(EXIT_FAIL)

tests/mock_tables/state_db.json

@@ -149,6 +149,18 @@
         "status": "255",
         "error": "N/A"
     },
+    "TRANSCEIVER_STATUS|Ethernet16": {
+        "status": "255",
+        "error": "Not present"
+    },
+    "TRANSCEIVER_STATUS|Ethernet28": {
+        "status": "0",
+        "error": "N/A"
+    },
+    "TRANSCEIVER_STATUS|Ethernet36": {
+        "status": "255",
+        "error": "Unknown"
+    },
     "CHASSIS_INFO|chassis 1": {
         "psu_num": "2"
     },

tests/sfp_test.py

@@ -344,6 +344,30 @@ def test_sfp_presence(self):
         expected = """Port         Presence
 -----------  -----------
 Ethernet200  Not present
+"""
+        assert result.exit_code == 0
+        assert result.output == expected
+
+        result = runner.invoke(show.cli.commands["interfaces"].commands["transceiver"].commands["presence"], ["Ethernet16"])
+        expected = """Port        Presence
+----------  -----------
+Ethernet16  Not present
+"""
+        assert result.exit_code == 0
+        assert result.output == expected
+
+        result = runner.invoke(show.cli.commands["interfaces"].commands["transceiver"].commands["presence"], ["Ethernet28"])
+        expected = """Port        Presence
+----------  ----------
+Ethernet28  Present
+"""
+        assert result.exit_code == 0
+        assert result.output == expected
+
+        result = runner.invoke(show.cli.commands["interfaces"].commands["transceiver"].commands["presence"], ["Ethernet36"])
+        expected = """Port        Presence
+----------  ----------
+Ethernet36  Unknown
 """
         assert result.exit_code == 0
         assert result.output == expected
@@ -377,6 +401,13 @@ def test_qsfp_dd_eeprom(self):
         assert result.exit_code == 0
         assert "result.output == test_qsfp_dd_eeprom_output"
 
+    def test_rj45_eeprom(self):
+        runner = CliRunner()
+        result = runner.invoke(show.cli.commands["interfaces"].commands["transceiver"].commands["eeprom"], ["Ethernet36"])
+        result_lines = result.output.strip('\n')
+        expected = "Ethernet36: SFP EEPROM is not applicable for RJ45 port"
+        assert result_lines == expected
+
     @classmethod
     def teardown_class(cls):
         print("TEARDOWN")