|
13 | 13 | ] |
14 | 14 |
|
15 | 15 |
|
16 | | -def parse_sfp_presence_output(sfp_presence_output): |
17 | | - """Parse the output of 'show interface transceiver presence' command. |
18 | | - Args: |
19 | | - sfp_presence_output: List of strings from the command output |
20 | | - Returns: |
21 | | - Dictionary mapping interface names to their presence status |
22 | | - """ |
23 | | - table_header_length = 2 |
24 | | - if not sfp_presence_output or len(sfp_presence_output) < table_header_length: |
25 | | - raise ValueError("Invalid output format: expected at least 2 lines") |
26 | | - sfp_presence_parsed = {} |
27 | | - for line in sfp_presence_output[table_header_length:]: |
28 | | - parts = line.split() |
29 | | - if len(parts) != 2: |
30 | | - raise ValueError(f"Invalid line: {line}") |
31 | | - port, status = parts |
32 | | - sfp_presence_parsed[port] = status |
33 | | - return sfp_presence_parsed |
34 | | - |
35 | | - |
36 | 16 | def test_check_sfp_presence(duthosts, rand_one_dut_hostname, conn_graph_facts, dpu_npu_port_list): # noqa F811 |
37 | 17 | """This test case is to check SFP presence status with CLI and sysfs. |
38 | 18 | """ |
39 | 19 | duthost = duthosts[rand_one_dut_hostname] |
40 | 20 | ports_config = json.loads(duthost.command("sudo sonic-cfggen -d --var-json PORT")["stdout"]) # noqa F841 |
41 | | - check_intf_presence_command = 'show interface transceiver presence' |
| 21 | + check_intf_presence_command = 'show interface transceiver presence {}' |
42 | 22 |
|
43 | 23 | logging.info("Use show interface status information") |
44 | | - check_presence_output = duthost.command(check_intf_presence_command) |
45 | | - assert check_presence_output["rc"] == 0, \ |
46 | | - f"Command '{check_intf_presence_command}' failed with return code {check_presence_output['rc']}" |
47 | | - |
48 | | - presence_stdout_lines = check_presence_output["stdout_lines"] |
49 | | - presence_dict = parse_sfp_presence_output(presence_stdout_lines) |
50 | | - logging.info(f"Found {len(presence_dict)} interfaces with presence status: {presence_dict}") |
51 | | - intfs = conn_graph_facts["device_conn"][duthost.hostname] |
52 | | - intf_list_to_check = {intf for intf in intfs if intf not in dpu_npu_port_list[duthost.hostname]} |
53 | | - logging.info(f"Interfaces to check: {intf_list_to_check}") |
54 | | - intf_not_present = intf_list_to_check - set(presence_dict.keys()) |
55 | | - assert len(intf_not_present) == 0, \ |
56 | | - f"Missing interfaces in presence output: {intf_not_present}" |
57 | | - logging.info("All expected interfaces are present in the output") |
| 24 | + for intf in conn_graph_facts["device_conn"][duthost.hostname]: |
| 25 | + if intf not in dpu_npu_port_list[duthost.hostname]: |
| 26 | + check_presence_output = duthost.command(check_intf_presence_command.format(intf)) |
| 27 | + assert check_presence_output["rc"] == 0, "Failed to read interface %s transceiver presence" % intf |
| 28 | + logging.info(str(check_presence_output["stdout_lines"][2])) |
| 29 | + presence_list = check_presence_output["stdout_lines"][2].split() |
| 30 | + logging.info(str(presence_list)) |
| 31 | + assert intf in presence_list, "Wrong interface name in the output %s" % str(presence_list) |
| 32 | + assert 'Present' in presence_list, "Status is not expected, output %s" % str(presence_list) |
0 commit comments