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

scripts/sfpshow

@@ -215,6 +215,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'
@@ -392,15 +394,18 @@ class SFPShow(object):
         output = ''
 
         sfp_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interface_name))
-        output = '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 sfp_info_dict['type'] == RJ45_PORT_TYPE:
+            output = 'SFP EEPROM is not applicable for RJ45 port\n'
+        else:
+            output = '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
 
         return output
 

sfputil/main.py

@@ -267,6 +267,7 @@
     'voltage': 'Volts'
 }
 
+RJ45_PORT_TYPE = 'RJ45'
 
 # Global platform-specific Chassis class instance
 platform_chassis = None
@@ -289,6 +290,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
@@ -630,6 +655,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:
@@ -783,7 +813,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
 
@@ -806,15 +839,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])
 
@@ -879,24 +915,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'))
 
@@ -919,6 +958,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:
@@ -954,6 +997,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
 
@@ -1010,6 +1057,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
 
@@ -1175,6 +1226,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))
@@ -1192,6 +1245,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))
@@ -1212,6 +1267,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)
@@ -1246,6 +1303,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:
@@ -1266,6 +1325,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

@@ -221,6 +221,18 @@
         "status": "255",
         "error": "N/A"
     },
+    "TRANSCEIVER_STATUS|Ethernet16": {
+        "status": "0",
+        "error": "N/A"
+    },
+    "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  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  Present
 """
         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")