-- Copyright (c) 2024 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- You can use this software according to the terms and conditions of the Mulan PSL v2.
-- You may obtain a copy of Mulan PSL v2 at: http://license.coscl.org.cn/MulanPSL2
-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
-- See the Mulan PSL v2 for more details.

local lu = require 'luaunit'
local c_network_port = require 'device.class.network_port'
local c_object_manage = require 'mc.orm.object_manage'
local device_comm_fun = require 'device.class.nic_mgmt.comm_fun'
local device_comm_defs = require 'device.class.nic_mgmt.comm_defs'
local npu_imu_cmd = require 'npu.hdk_cmd'
local test_utils = require 'test_utils'
local class_mgnt = require "mc.class_mgnt"
local fructl = require 'infrastructure.fructl'
skynet = require 'skynet'
local context = require 'mc.context'
local client = require 'network_adapter.client'
local vos = require 'utils.vos'
local c_optical_module = require 'device.class.optical_module'

TEST_network_port = {}

local INVALID_DATA_STRING<const> = 'N/A'

local mock_imu_info = {
    received = 'test',
    transmitted = 'test_transmitted',
    dropped = 'test_dropped',
    mac_addr = 'test_mac_addr',
    rx_fcs_err_pkt_num = 'test_rx_fcs_err_pkt_num',
    subnet_mask = 'test_subnet_mask',
    gateway = 'test_gateway',
    ip_addr = 'test_ipaddr',
    link_up_cnt = 1,
    link_down_cnt = 2,
    link_up_timestamp = {3},
    link_down_timestamp = {0,44}
}

local mock_imu_hccs_info = {
    {
        macro_id = 1,
        health_status = 1,
        lane_mode = {},
        link_lane_list = {},
        link_speed = {},
        tx_packets = {},
        tx_bytes = {},
        rx_packets = {},
        rx_bytes = {},
        retry_count = {},
        error_count = {},
        first_error_lane = 2,
        snr = {0, 2, 4, 6},
        half_height = {0, 1, 2, 3}
    }
}

local function mock_get_info_from_imu(id, power_on)
    return mock_imu_info
end

local function mock_get_hccs_info_from_imu(id, power_on)
    return mock_imu_hccs_info
end

local function mock_create_ipv4_addr(Object, SystemID, ID1, ID2, ID3, prop_setting_cb)
    local path = '/bmc/kepler/Systems/' .. SystemID .. '/NetworkAdapters/' .. ID1 .. '/Ports/' ..
                ID2 .. '/IPv4Addresses/' .. ID3 .. ''
    local obj = {path = path}
    prop_setting_cb(obj)
    return obj
end

local function mock_get_instance()
    return {
        app = {
            CreateIPv4Address = mock_create_ipv4_addr
        }
    }
end

local function mock_get_postion()
    return 'test_postion'
end

local function mock_class_mgnt(...)
    return {
        remove = function(...)
            return nil
        end
    }
end

local function mock_get_power_status(...)
    return 'OFF'
end

local mock_network_port = {
    NpuID = 1,
    per_module_log_cnt = 0,
    record_flag = 0,
    link_up_cnt = 0,
    link_down_cnt = 0,
    link_up_timestamp = {},
    link_down_timestamp = {},
    npu_port_log_buffer = {},
    PortID = 1,
    RXFrames = 'test',
    TXFrames = 'test',
    PacketsDropped = 'test',
    MACAddress = 'test',
    RXFCSErrors = 'test',
    npu_ipv4_obj = false,
    npu_ipv4_info = {
        Address = INVALID_DATA_STRING,
        SubnetMask = INVALID_DATA_STRING,
        Gateway = INVALID_DATA_STRING
    },
    get_position = mock_get_postion,
    hccs_info = {
        npu_id = 1,
        health_status = 0,
        lane_mode = {},
        link_lane_list = {},
        link_speed = {},
        tx_packets = {},
        tx_bytes = {},
        rx_packets = {},
        rx_bytes = {},
        retry_count = {},
        error_count = {},
        first_error_lane = 255,
        snr_max = {{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0},
            {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}},
        snr_min = {{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0},
            {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}},
        half_height_max = {{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0},
            {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}},
        half_height_min = {{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0},
            {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}}
    },
    clear_npu_port_basic_info = function(...)
        return nil
    end,
    get_npu_port_basic_info = function()
        return 'str' + 1
    end,
    clear_hccs_info = function()
        return
    end,
    update_npu_port_basic_info = function()
        return
    end,
    update_npu_port_hccs_info = function()
        return
    end,
    npu_port_log_collect_record = function()
        return
    end,
    sleep_ms = function(...)

    end,
    get_parent = function()
        return {
            Model = ''
        }
    end
}

local get_info_from_imu = npu_imu_cmd.get_info_from_imu
local get_hccs_info_from_imu = npu_imu_cmd.get_hccs_info_from_imu
local get_instance = c_object_manage.get_instance
local get_power_status = fructl.get_power_status

function TEST_network_port:setUp()
    npu_imu_cmd.get_info_from_imu = mock_get_info_from_imu
    npu_imu_cmd.get_hccs_info_from_imu = mock_get_hccs_info_from_imu
    c_object_manage.get_instance = mock_get_instance
    fructl.get_power_status = mock_get_power_status
end

function TEST_network_port:tearDown()
    npu_imu_cmd.get_info_from_imu = get_info_from_imu
    npu_imu_cmd.get_hccs_info_from_imu = get_hccs_info_from_imu
    c_object_manage.get_instance = get_instance
    fructl.get_power_status = get_power_status
end

local test_data = {
    LinkStatus = 'N/A',
    NetworkAdapterId = 0,
    PortID = 0,
    SpeedMbps = 0,
    AutoSpeedNegotiation = false,
    FullDuplex = false
}
-- 测试更新链接状态
function TEST_network_port:test_update_ncsi_link_status()
    local origin_schedulers = c_network_port.schedulers
    local origin_smbus_schedulers = c_network_port.smbus_schedulers
    local origin_get_properties_res = c_network_port.get_properties_res
    c_network_port.t_prop_without_bma = {}
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.get_properties_res = {}
    c_network_port.ncsi_config_obj = {
        LinkStatus = function()
            return {
                on_data_change = {
                    on = function(self, func)
                        func(test_data)
                        c_network_port.ncsi_get_linkstatus_retry_count = 2
                    end
                },
                on_error = {
                    on = function(self, func)
                        func()
                    end
                },
                start = function()
                end
            }
        end
    }
    c_network_port:update_ncsi_link_status()
    lu.assertEquals(c_network_port.AutoSpeedNegotiation, 255)
    lu.assertEquals(c_network_port.FullDuplex, 255)
    lu.assertEquals(c_network_port.LinkStatus, test_data.LinkStatus)
    lu.assertEquals(c_network_port.LinkStatus, test_data.LinkStatus)
    lu.assertEquals(c_network_port.SpeedMbps, test_data.SpeedMbps)
    c_network_port.schedulers = origin_schedulers
    c_network_port.smbus_schedulers = origin_smbus_schedulers
    c_network_port.get_properties_res = origin_get_properties_res
end

function TEST_network_port:test_get_npu_port_basic_info()
    mock_network_port.npu_ipv4_obj = true
    mock_imu_info.ip_addr = INVALID_DATA_STRING
    mock_imu_info.subnet_mask = INVALID_DATA_STRING
    mock_imu_info.gateway = INVALID_DATA_STRING

    -- 测试npu仅能获取一个ipv4信息，当ip网关掩码均不变时不做资源树对象操作场景
    c_network_port.get_npu_port_basic_info(mock_network_port)
    lu.assertEquals(mock_network_port.npu_ipv4_info.Address, INVALID_DATA_STRING)

    -- 测试当ip网关掩码均为无效值时不上树场景
    mock_network_port.npu_ipv4_obj = false
    c_network_port.get_npu_port_basic_info(mock_network_port)
    lu.assertEquals(mock_network_port.npu_ipv4_info.Address, INVALID_DATA_STRING)

    -- 测试更新ipv4缓存信息并上树场景
    mock_imu_info.ip_addr = 'test_ip_addr_xx'
    mock_imu_info.subnet_mask = 'test_ip_addr'
    mock_imu_info.gateway = 'test_ip_addr'
    c_network_port.get_npu_port_basic_info(mock_network_port)
    lu.assertEquals(mock_network_port.npu_ipv4_obj.Address, 'test_ip_addr_xx')
    lu.assertEquals(mock_network_port.per_module_log_cnt, 0)
    lu.assertEquals(#mock_network_port.npu_port_log_buffer, 0)
end

function TEST_network_port:test_get_npu_port_hccs_info()
    c_network_port.get_npu_port_hccs_info(mock_network_port)
    lu.assertEquals(mock_network_port.hccs_info.first_error_lane, 255)
end

function TEST_network_port:test_update_npu_port_hccs_info()
    c_network_port.update_npu_port_hccs_info(mock_network_port, mock_imu_hccs_info)
    lu.assertEquals(mock_network_port.hccs_info.first_error_lane, 2)
end

function TEST_network_port:test_clear_hccs_info()
    mock_network_port.hccs_info.health_status = 1
    pcall(function ()
            c_network_port.clear_hccs_info(mock_network_port)
        end
    )
    lu.assertEquals(mock_network_port.hccs_info.health_status, 0)
end

function TEST_network_port:test_clear_npu_port_basic_info()
    mock_network_port.npu_ipv4_obj = {
        IPv4Address = 'test'
    }
    pcall(function ()
            c_network_port.clear_npu_port_basic_info(mock_network_port)
        end
    )
    lu.assertEquals(mock_network_port.npu_ipv4_info.Address, INVALID_DATA_STRING)
end

function TEST_network_port:test_get_npu_port_info()
    c_network_port.get_npu_port_info(mock_network_port)
end

function TEST_network_port:test_get_info_from_imu()
    c_network_port.get_info_from_imu(mock_network_port)
end

function TEST_network_port:test_get_npu_cdr_temp_from_imu_retry()
    local origin_get_npu_cdr_temp_from_imu = npu_imu_cmd.get_npu_cdr_temp_from_imu
    npu_imu_cmd.get_npu_cdr_temp_from_imu = function()
        return false
    end
    local origin_get_parent = c_network_port.get_parent
    c_network_port.get_parent = function()
        return {
            Model = ''
        }
    end
    local ok = c_network_port:get_npu_cdr_temp_from_imu_retry()
    lu.assertIsTrue(ok)
    c_network_port.get_parent = origin_get_parent
    npu_imu_cmd.get_npu_cdr_temp_from_imu = origin_get_npu_cdr_temp_from_imu
end

function TEST_network_port:test_get_npu_cdr_temp_from_imu_fail()
    c_network_port.is_npu_heartbeat_loss = true
    c_network_port.get_npu_cdr_temp_from_imu(mock_network_port)
end

function TEST_network_port:test_get_npu_cdr_temp_from_imu()
    c_network_port.is_npu_heartbeat_loss = false
    c_network_port.get_npu_cdr_temp_from_imu(mock_network_port)

    local origin_get_power_status = fructl.get_power_status
    local origin_get_npu_cdr_temp_from_imu_retry = c_network_port.get_npu_cdr_temp_from_imu_retry
    fructl.get_power_status = function()
        return 'OFF'
    end
    local temp = c_network_port:get_npu_cdr_temp_from_imu()
    lu.assertEquals(temp, 0)

    fructl.get_power_status = function()
        return 'ON'
    end
    c_network_port.PowerOn = 1
    c_network_port.is_npu_heartbeat_loss = true
    c_network_port.is_dmp_status_ready = true
    c_network_port:get_npu_cdr_temp_from_imu()
    lu.assertIsTrue(c_network_port.npu_port_cdr_info_abnormal)

    c_network_port.is_npu_heartbeat_loss = false

    c_network_port.get_npu_cdr_temp_from_imu_retry = function()
        return true, true
    end
    c_network_port:get_npu_cdr_temp_from_imu()
    lu.assertIsFalse(c_network_port.npu_port_cdr_info_abnormal)

    c_network_port.get_npu_cdr_temp_from_imu_retry = function()
        return true, false
    end
    c_network_port:get_npu_cdr_temp_from_imu()
    lu.assertIsTrue(c_network_port.npu_port_cdr_info_abnormal)

    fructl.get_power_status = origin_get_power_status
    c_network_port.get_npu_cdr_temp_from_imu_retry = origin_get_npu_cdr_temp_from_imu_retry
end

function TEST_network_port:test_mctp()
    local original_new = context.new
    local original_get_context = context.get_context
    local original_set_context = context.set_context
    local original_get_parent = c_network_port.get_parent
    local data = {
        SpeedMbps = "SpeedMbps",
        AutoSpeedNegotiation = "AutoSpeedNegotiation",
        FullDuplex = "FullDuplex"
    }

    local mock_ctx = {
        initiator = 'test',
        has_initiator = function() return true end,
        get_initiator = function() return 'test' end
    }
    context.new = function() return mock_ctx end
    context.get_context = function() return mock_ctx end
    context.set_context = function(ctx) end
    c_network_port.get_parent = function()
        return {
            Model = "CX4"
        }
    end
    c_network_port.schedulers = {}
    c_network_port.ncsi_config_obj = {
        GetLLDPCapability = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
        PortMetrics = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
        LinkStatus = function()
            return {
                on_data_change = { on = function(data) end },
                on_error = { on = function(data) end },
                start = function () end
            }
        end,
        InitializeNCSIChannel =  function() return { value = function() return {} end } end,
        BDF = function()
            return { value = function() return { bus = 1, device = 1, func = 1 } end }
        end,
        MacAddrNCSI =  function() return { value = function() return {} end } end,
        DefaultMacAddrNCSI =  function() return { value = function() return {} end } end,
        DCBX =  function()
            return {
                value = function()
                    return {}
                end,
                on_data_change = {
                    on = function(self, func)
                        func(data)
                    end
                },
                start = function () end
            }
        end,
        SetOnLLDPCapability =  function() return { value = function() return true end } end,
        SetOffLLDPCapability = function() return { value = function() return false end } end,
        ResetNic =  function() return { value = function() return {} end } end,
        GetLLDPTxEnable = function()
            return {
                on_data_change = { on = function() end },
                start = function() end
            }
        end
    }
    c_network_port:update_lldp_tx_enabled_by_ncsi()
    c_network_port:update_ncsi_port_stats()
    c_network_port:update_ncsi_link_status()
    c_network_port:initialize_ncsi_channel()
    c_network_port:update_BDF_by_ncsi()
    c_network_port:update_DCBX_by_ncsi()
    c_network_port:set_package_id(1)

    pcall(function() c_network_port:redfish_set_lldp_capability(true) end)
    pcall(function()
        c_network_port:redfish_set_lldp_capability(false)
    end)

    context.new = original_new
    context.get_context = original_get_context
    context.set_context = original_set_context
    c_network_port.get_parent = original_get_parent
end

function TEST_network_port:test_mctp_mac()
    local original_new = context.new
    local original_get_context = context.get_context
    local original_set_context = context.set_context
    local pre_fun = c_network_port.get_parent
    local origin_get_properties_res = c_network_port.get_properties_res
    local origin_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    c_network_port.get_parent = function()
        return {
            Model = "BroadCom"
        }
    end
    c_network_port.get_properties_res = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.ncsi_config_obj = {
        MacAddrNCSI =  function()
            return {
                value = function()
                    return {}
                end,
                on_data_change = {
                    on = function(self, func)
                        func("asdf:asdf:asdf:asdf:asdf")
                    end
                },
                on_error = {
                    on = function(self, func)
                        func()
                    end
                },
                start = function () end
            }
        end,
        DefaultMacAddrNCSI =  function()
            return {
                value = function()
                    return {}
                end,
                on_data_change = {
                    on = function(self, func)
                        func("")
                    end
                },
                on_error = {
                    on = function(self, func)
                        func()
                    end
                },
                start = function () end
            }
        end
    }
    c_network_port:update_MAC_by_ncsi()
    c_network_port:update_default_MAC_by_ncsi()

    context.new = original_new
    context.get_context = original_get_context
    context.set_context = original_set_context
    c_network_port.get_parent = pre_fun
    c_network_port.get_properties_res = origin_get_properties_res
    c_network_port.smbus_schedulers = origin_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
end

local mockup_func = {}
local function mockup_func_backup()
    mockup_func.connect_signal = c_network_port.connect_signal
    mockup_func.get_power_status = fructl.get_power_status
    mockup_func.get_system_reset_flag = fructl.get_system_reset_flag
end
local function mockup_function()
    c_network_port.connect_signal = function(self, on_cb, cb)
        cb(self.name, self.value)
    end
    fructl.get_power_status = function()
        return "ON"
    end
end
local function mockup_func_restore()
    c_network_port.connect_signal = mockup_func.connect_signal
    fructl.get_power_status = mockup_func.get_power_status
    fructl.get_system_reset_flag = mockup_func.get_system_reset_flag
end

function TEST_network_port:test_start_npu_link_status_listening()
    mockup_func_backup()
    mockup_function()
    fructl.get_system_reset_flag = function()
        return 2
    end
    c_network_port.value = 0
    c_network_port.name = "LinkStatusValue"
    c_network_port.LinkStatusNumeric = 255
    c_network_port:start_npu_link_status_listening()
    lu.assertEquals(c_network_port.LinkStatusNumeric, 2)

    c_network_port.LinkStatusNumeric = 1
    c_network_port:start_npu_link_status_listening()
    lu.assertEquals(c_network_port.LinkStatusNumeric, 0)

    c_network_port.value = 1
    c_network_port:start_npu_link_status_listening()
    lu.assertEquals(c_network_port.LinkStatusNumeric, 1)

    mockup_func_restore()
end

function TEST_network_port:test_register_npu_changed_callback()
    mockup_func_backup()
    mockup_function()
    local start_npu_link_status_listening = c_network_port.start_npu_link_status_listening
    c_network_port.start_npu_link_status_listening = function()
    end
    fructl.get_system_reset_flag = function()
        return 0
    end
    c_network_port.LinkStatusValue = 1
    c_network_port:register_npu_changed_callback()
    lu.assertEquals(c_network_port.LinkStatusNumeric, 1)

    c_network_port.LinkStatusValue = 2
    c_network_port:register_npu_changed_callback()
    lu.assertEquals(c_network_port.LinkStatusNumeric, 2)

    c_network_port.start_npu_link_status_listening = start_npu_link_status_listening
    mockup_func_restore()
end

function TEST_network_port:test_npu_port_link_event()
    local origin = c_network_port.npu_port_get_diff_timestamp
    c_network_port.link_down_cnt = 1
    c_network_port.link_down_timestamp = {0}
    c_network_port.npu_port_log_buffer = {}
    c_network_port.per_module_log_cnt = 0
    c_network_port.NpuID = 1
    c_network_port:npu_port_link_event(mock_imu_info, true)
    lu.assertEquals(c_network_port.per_module_log_cnt, 1)
    c_network_port.npu_port_get_diff_timestamp = origin
end

-- 通用的mock设置函数
local function setup_redfish_lldp_mocks()
    local mocks = {}
    
    -- 保存原始值
    mocks.original_get_context = context.get_context
    mocks.original_get_parent = c_network_port.get_parent
    mocks.original_get_port_id = c_network_port.get_port_id
    mocks.original_ncsi_config_obj = c_network_port.ncsi_config_obj
    mocks.original_NodeId = c_network_port.NodeId
    mocks.original_NetworkAdapterId = c_network_port.NetworkAdapterId
    mocks.original_PortID = c_network_port.PortID
    mocks.original_package_id = c_network_port.package_id
    mocks.original_get_position = c_network_port.get_position
    
    -- Mock对象
    mocks.mock_ctx = {
        has_initiator = function() return true end,
        get_initiator = function() return 'test_user' end
    }
    
    mocks.mock_parent_supported = {
        SupportedLLDP = true,
        Model = 'Hi1822'
    }

    mocks.mock_parent_not_supported = {
        SupportedLLDP = false,
        Model = 'Hi1822'
    }

    local dpu_obj = {}
    local extra_params = {}
    extra_params.Position = '00010102'
    dpu_obj.extra_params = extra_params
    mocks.set_mcu_on_result = true
    mocks.set_mcu_off_result = true
    dpu_obj.GetLLDPStatus = function()
        mocks.get_mcu_called = true
        return true
    end
    dpu_obj.SetLLDPStatus = function()
        mocks.set_mcu_called = true
        return true
    end
    client.GetDPUCardObjects = function()
        return {dpu_obj}
    end
    mocks.dpu = dpu_obj

    -- 控制变量
    mocks.set_on_called = false
    mocks.set_off_called = false
    mocks.reset_nic_called = false
    mocks.set_on_result = true
    mocks.set_off_result = true
    mocks.reset_nic_result = true
    
    return mocks
end

-- 应用通用的mock设置
local function apply_redfish_lldp_mocks(mocks)
    context.get_context = function() return mocks.mock_ctx end
    c_network_port.get_parent = function() return mocks.mock_parent_supported end
    c_network_port.get_port_id = function() return 1 end
    c_network_port.NodeId = 'test_node_1'
    c_network_port.NetworkAdapterId = 'NetworkAdapter_1'
    c_network_port.PortID = 1
    c_network_port.package_id = 0
    c_network_port.get_position = function() return '00010102' end
    
    c_network_port.ncsi_config_obj = {
        SetOnLLDPCapability = function()
            mocks.set_on_called = true
            return { value = function() return mocks.set_on_result end }
        end,
        SetOffLLDPCapability = function()
            mocks.set_off_called = true
            return { value = function() return mocks.set_off_result end }
        end,
        ResetNic = function()
            mocks.reset_nic_called = true
            return { value = function() return mocks.reset_nic_result end }
        end
    }
end

-- 还原所有mock
local function restore_redfish_lldp_mocks(mocks)
    context.get_context = mocks.original_get_context
    c_network_port.get_parent = mocks.original_get_parent
    c_network_port.get_port_id = mocks.original_get_port_id
    c_network_port.ncsi_config_obj = mocks.original_ncsi_config_obj
    c_network_port.NodeId = mocks.original_NodeId
    c_network_port.NetworkAdapterId = mocks.original_NetworkAdapterId
    c_network_port.PortID = mocks.original_PortID
    c_network_port.package_id = mocks.original_package_id
end

-- 测试成功场景
function TEST_network_port:test_redfish_set_lldp_capability_success()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    -- 添加Hi182x需要的mock方法
    c_network_port.ncsi_config_obj.SetOnLLDPTxEnable = function()
        mocks.set_on_called = true
        return { value = function() return mocks.set_on_result end }
    end
    c_network_port.ncsi_config_obj.SetOffLLDPTxEnable = function()
        mocks.set_off_called = true
        return { value = function() return mocks.set_off_result end }
    end

    -- 测试启用LLDP (Hi182x网卡)
    mocks.set_on_called = false
    mocks.reset_nic_called = false

    c_network_port:redfish_set_lldp_capability(true)

    lu.assertIsTrue(mocks.set_on_called, "SetOnLLDPTxEnable should be called")
    lu.assertIsFalse(mocks.reset_nic_called, "ResetNic should NOT be called for Hi182x")

    -- 测试禁用LLDP (Hi182x网卡)
    mocks.set_off_called = false
    mocks.reset_nic_called = false

    c_network_port:redfish_set_lldp_capability(false)

    lu.assertIsTrue(mocks.set_off_called, "SetOffLLDPTxEnable should be called")
    lu.assertIsFalse(mocks.reset_nic_called, "ResetNic should NOT be called for Hi182x")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试调用mcu通路成功场景
function TEST_network_port:test_redfish_set_lldp_capability_by_mcu()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    -- 测试启用LLDP (非CX4网卡)
    mocks.set_mcu_called = false
    c_network_port.WorkloadType = 1

    c_network_port:redfish_set_lldp_capability(true)

    lu.assertIsTrue(mocks.set_mcu_called, "set lldp on by mcu failed")

    -- 测试禁用LLDP (非CX4网卡)
    c_network_port:redfish_set_lldp_capability(false)

    lu.assertIsTrue(mocks.set_mcu_called, "set lldp off by mcu failed")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试调用mcu通路成功场景
function TEST_network_port:test_redfish_set_lldp_capability_by_mcu_failed()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    mocks.dpu.extra_params.Position = '00010103'
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    lu.assertIsFalse(success, "can not get dpu object")

    mocks.dpu.extra_params.Position = '00010102'
    mocks.dpu.SetLLDPStatus = function()
        return error('error')
    end
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    lu.assertIsFalse(success, "set lldp failed")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试调用mcu通路更新LLDP失败场景
function TEST_network_port:test_update_LLDP_enabled_by_mcu_failed()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    mocks.dpu.extra_params.Position = '00010103'
    c_network_port:update_LLDP_enabled_by_mcu()
    lu.assertIsFalse(false, "can not get dpu object")

    mocks.dpu.extra_params.Position = '00010102'
    mocks.dpu.GetLLDPStatus = function()
        return error('error')
    end
    c_network_port:update_LLDP_enabled_by_mcu()
    lu.assertIsFalse(false, "get lldp failed")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试调用mcu更新LLDP成功场景
function TEST_network_port:test_update_LLDP_enabled_by_mcu()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    -- 测试更新LLDP (非CX4网卡)
    mocks.get_mcu_called = false

    c_network_port:update_LLDP_enabled_by_mcu()

    lu.assertIsTrue(mocks.get_mcu_called, "get lldp by mcu failed")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试CX4网卡的特殊处理
function TEST_network_port:test_redfish_set_lldp_capability_cx4()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 设置CX4网卡
    local mock_parent_cx4 = {
        SupportedLLDP = true,
        Model = 'CX4'
    }
    c_network_port.get_parent = function() return mock_parent_cx4 end
    c_network_port.WorkloadType = function() return 0 end
    
    -- 测试CX4启用LLDP成功
    mocks.set_on_called = false
    mocks.reset_nic_called = false
    
    c_network_port:redfish_set_lldp_capability(true)
    
    lu.assertIsTrue(mocks.set_on_called, "SetOnLLDPCapability should be called for CX4")
    lu.assertIsTrue(mocks.reset_nic_called, "ResetNic should be called for CX4")
    
    -- 测试CX4禁用LLDP成功
    mocks.set_off_called = false
    mocks.reset_nic_called = false
    
    c_network_port:redfish_set_lldp_capability(false)
    
    lu.assertIsTrue(mocks.set_off_called, "SetOffLLDPCapability should be called for CX4")
    lu.assertIsTrue(mocks.reset_nic_called, "ResetNic should be called for CX4")
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试CX4网卡ResetNic失败的场景
function TEST_network_port:test_redfish_set_lldp_capability_cx4_reset_fail()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 设置CX4网卡
    local mock_parent_cx4 = {
        SupportedLLDP = true,
        Model = 'CX4'
    }
    c_network_port.get_parent = function() return mock_parent_cx4 end
    
    -- 设置ResetNic失败
    mocks.reset_nic_result = false
    
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    
    lu.assertIsFalse(success, "Should fail when CX4 ResetNic fails")
    lu.assertIsTrue(mocks.set_on_called, "SetOnLLDPCapability should be called before ResetNic fails")
    lu.assertIsTrue(mocks.reset_nic_called, "ResetNic should be called for CX4")
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试非CX4网卡的正常处理
function TEST_network_port:test_redfish_set_lldp_capability_non_cx4()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)

    -- 添加Hi182x需要的mock方法
    c_network_port.ncsi_config_obj.SetOnLLDPTxEnable = function()
        mocks.set_on_called = true
        return { value = function() return mocks.set_on_result end }
    end
    c_network_port.ncsi_config_obj.SetOffLLDPTxEnable = function()
        mocks.set_off_called = true
        return { value = function() return mocks.set_off_result end }
    end

    -- 测试Hi182x网卡
    local mock_parent_hi182x = {
        SupportedLLDP = true,
        Model = 'Hi1822'
    }
    c_network_port.get_parent = function() return mock_parent_hi182x end

    -- 重置状态
    mocks.set_on_called = false
    mocks.reset_nic_called = false

    c_network_port:redfish_set_lldp_capability(true)

    lu.assertIsTrue(mocks.set_on_called, "SetOnLLDPTxEnable should be called for Hi1822")
    lu.assertIsFalse(mocks.reset_nic_called, "ResetNic should NOT be called for Hi1822")

    restore_redfish_lldp_mocks(mocks)
end

-- 测试不支持LLDP的场景
function TEST_network_port:test_redfish_set_lldp_capability_not_supported()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 设置不支持LLDP的父对象
    c_network_port.get_parent = function() return mocks.mock_parent_not_supported end
    
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    
    lu.assertIsFalse(success, "Should fail when LLDP not supported")
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试NCSI命令失败场景
function TEST_network_port:test_redfish_set_lldp_capability_failures()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 测试SetOnLLDPCapability失败
    mocks.set_on_result = false
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    lu.assertIsFalse(success, "Should fail when SetOnLLDPCapability fails")
    
    -- 测试SetOffLLDPCapability失败
    mocks.set_on_result = true
    mocks.set_off_result = false
    success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(false)
    end)
    lu.assertIsFalse(success, "Should fail when SetOffLLDPCapability fails")
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试Context相关场景
function TEST_network_port:test_redfish_set_lldp_capability_context()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 测试无context的情况
    context.get_context = function() return nil end
    c_network_port:redfish_set_lldp_capability(true)
    
    -- 测试空context的情况
    local mock_empty_ctx = {
        has_initiator = function() return false end
    }
    context.get_context = function() return mock_empty_ctx end
    c_network_port:redfish_set_lldp_capability(false)
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试异常处理场景
function TEST_network_port:test_redfish_set_lldp_capability_exceptions()
    local mocks = setup_redfish_lldp_mocks()
    apply_redfish_lldp_mocks(mocks)
    
    -- 测试SetOnLLDPCapability抛出异常
    c_network_port.ncsi_config_obj.SetOnLLDPCapability = function()
        error("Mock exception")
    end
    local success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    lu.assertIsFalse(success, "Should fail when SetOnLLDPCapability throws exception")
    
    -- 测试SetOffLLDPCapability抛出异常
    c_network_port.ncsi_config_obj.SetOffLLDPCapability = function()
        error("Mock exception")
    end
    success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(false)
    end)
    lu.assertIsFalse(success, "Should fail when SetOffLLDPCapability throws exception")
    
    -- 测试CX4网卡ResetNic抛出异常
    local mock_parent_cx4 = {
        SupportedLLDP = true,
        Model = 'CX4'
    }
    c_network_port.get_parent = function() return mock_parent_cx4 end
    c_network_port.ncsi_config_obj.SetOnLLDPCapability = function()
        mocks.set_on_called = true
        return { value = function() return mocks.set_on_result end }
    end
    c_network_port.ncsi_config_obj.ResetNic = function()
        error("Mock exception")
    end
    success, _ = pcall(function()
        c_network_port:redfish_set_lldp_capability(true)
    end)
    lu.assertIsFalse(success, "Should fail when CX4 ResetNic throws exception")
    
    restore_redfish_lldp_mocks(mocks)
end

-- 测试日志操作封装函数
function TEST_network_port:test_log_lldp_operation()
    local original_get_context = context.get_context
    local original_NodeId = c_network_port.NodeId
    local original_NetworkAdapterId = c_network_port.NetworkAdapterId  
    local original_PortID = c_network_port.PortID
    
    -- 设置测试数据
    c_network_port.NodeId = 'test_node_1'
    c_network_port.NetworkAdapterId = 'NetworkAdapter_1'
    c_network_port.PortID = 1
    
    local mock_ctx = {
        has_initiator = function() return true end,
        get_initiator = function() return 'test_user' end
    }
    
    -- 测试有context且成功的情况
    context.get_context = function() return mock_ctx end
    c_network_port:log_lldp_operation(true, mock_ctx, true)
    
    -- 测试有context但失败的情况  
    c_network_port:log_lldp_operation(false, mock_ctx, false)
    
    -- 测试空context的情况
    local mock_empty_ctx = {
        has_initiator = function() return false end
    }
    c_network_port:log_lldp_operation(true, mock_empty_ctx, true)
    
    -- 测试nil context的情况
    c_network_port:log_lldp_operation(false, nil, false)
    
    -- 还原原始值
    context.get_context = original_get_context
    c_network_port.NodeId = original_NodeId
    c_network_port.NetworkAdapterId = original_NetworkAdapterId
    c_network_port.PortID = original_PortID
end

function TEST_network_port:test_npu_drive_monitor()
    local c_PortID = c_network_port.PortID
    local c_LinkStatusValue = c_network_port.LinkStatusValue
    local c_PowerOn = c_network_port.PowerOn
    local c_LinkDown = c_network_port.LinkDown
    local c_MediumType = c_network_port.MediumType
    c_network_port.PortID = 1
    c_network_port.LinkStatusValue = 0
    c_network_port.PowerOn = 0
    c_network_port.LinkDown = 0
    c_network_port.MediumType = "FiberOptic"
    local time = 0
    local c_sleep_ms = c_network_port.sleep_ms
    c_network_port.sleep_ms = function()
        time = time + 60000
        if time > 1000000 then
            time = 0
            error()
        elseif time > 60000 then
            c_network_port.PowerOn = 1
        end
    end
    local c_vos_tick_get = vos.vos_tick_get
    vos.vos_tick_get = function()
        return time
    end
    local c_collection = c_optical_module.collection
    local tab = {
        {
            RelatedNetworkPorts = {0,1},
            RXInputPowerMilliWatts = {1}
        }
    }
    c_optical_module.collection = {
        fold = function(obj, cb)
            for _, v in pairs(tab) do
                cb(_, v)
            end
        end
    }
    pcall(function()
        c_network_port:npu_driver_monitor()
    end)
    lu.assertEquals(c_network_port.LinkDown, 1)
    c_network_port.sleep_ms = c_sleep_ms
    vos.vos_tick_get = c_vos_tick_get
    c_optical_module.collection = c_collection
    c_network_port.PortID = c_PortID
    c_network_port.LinkStatusValue = c_LinkStatusValue
    c_network_port.PowerOn = c_PowerOn
    c_network_port.LinkDown = c_LinkDown
    c_network_port.MediumType = c_MediumType
end

-- 测试update_ncsi_link_status函数的NCSI恢复时停止SMBUS fallback逻辑
function TEST_network_port:test_update_ncsi_link_status_ncsi_recovered_stops_smbus_fallback()
    -- 备份原始状态
    local original_schedulers = c_network_port.schedulers
    local original_smbus_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_t_prop_without_bma = c_network_port.t_prop_without_bma
    local original_ncsi_get_linkstatus_retry_count = c_network_port.ncsi_get_linkstatus_retry_count
    local original_get_ncsi_link_status_succeess = c_network_port.get_ncsi_link_status_succeess
    local original_LinkStatus = c_network_port.LinkStatus
    local original_SpeedMbps = c_network_port.SpeedMbps
    local original_AutoSpeedNegotiation = c_network_port.AutoSpeedNegotiation
    local original_FullDuplex = c_network_port.FullDuplex
    
    -- 初始化测试状态
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.t_prop_without_bma = {}
    c_network_port.ncsi_get_linkstatus_retry_count = 0
    c_network_port.get_ncsi_link_status_succeess = false
    
    -- 模拟SMBUS任务正在运行
    local smbus_task_paused = false
    c_network_port.smbus_schedulers.link_status_task = {
        is_paused = false,
        pause = function() smbus_task_paused = true end,
        data = "some_data"
    }
    
    -- 模拟NCSI配置对象
    local scheduler_mock = {
        on_data_change = {
            on = function(self, callback)
                self.callback = callback
            end
        },
        on_error = {
            on = function(self, callback)
                self.error_callback = callback
            end
        },
        start = function() end
    }
    
    c_network_port.ncsi_config_obj = {
        LinkStatus = function(params) return scheduler_mock end
    }
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_port:update_ncsi_link_status()
    
    -- 模拟NCSI数据变化（恢复正常）
    local link_data = {
        LinkStatus = 'LinkUp',
        SpeedMbps = 1000,
        AutoSpeedNegotiation = true,
        FullDuplex = true
    }
    scheduler_mock.on_data_change.callback(link_data)
    
    -- 验证结果
    lu.assertEquals(c_network_port.LinkStatus, link_data.LinkStatus)
    lu.assertEquals(c_network_port.SpeedMbps, link_data.SpeedMbps)
    lu.assertEquals(c_network_port.AutoSpeedNegotiation, link_data.AutoSpeedNegotiation)
    lu.assertEquals(c_network_port.FullDuplex, link_data.FullDuplex)
    lu.assertTrue(smbus_task_paused)
    lu.assertNil(c_network_port.smbus_schedulers.link_status_task.data)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_port.schedulers = original_schedulers
    c_network_port.smbus_schedulers = original_smbus_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.t_prop_without_bma = original_t_prop_without_bma
    c_network_port.ncsi_get_linkstatus_retry_count = original_ncsi_get_linkstatus_retry_count
    c_network_port.get_ncsi_link_status_succeess = original_get_ncsi_link_status_succeess
    c_network_port.LinkStatus = original_LinkStatus
    c_network_port.SpeedMbps = original_SpeedMbps
    c_network_port.AutoSpeedNegotiation = original_AutoSpeedNegotiation
    c_network_port.FullDuplex = original_FullDuplex
end

-- 测试update_ncsi_link_status函数的NCSI失败时启动SMBUS fallback逻辑
function TEST_network_port:test_update_ncsi_link_status_ncsi_failed_activates_smbus_fallback()
    -- 备份原始状态
    local original_schedulers = c_network_port.schedulers
    local original_smbus_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_t_prop_without_bma = c_network_port.t_prop_without_bma
    local original_ncsi_get_linkstatus_retry_count = c_network_port.ncsi_get_linkstatus_retry_count
    local original_get_ncsi_link_status_succeess = c_network_port.get_ncsi_link_status_succeess
    local original_LinkStatus = c_network_port.LinkStatus
    local original_SpeedMbps = c_network_port.SpeedMbps
    local original_AutoSpeedNegotiation = c_network_port.AutoSpeedNegotiation
    local original_FullDuplex = c_network_port.FullDuplex
    
    -- 初始化测试状态
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.t_prop_without_bma = {}
    c_network_port.ncsi_get_linkstatus_retry_count = 0
    c_network_port.get_ncsi_link_status_succeess = false
    
    -- 模拟SMBUS任务已暂停
    local smbus_task_resumed = false
    c_network_port.smbus_schedulers.link_status_task = {
        is_paused = true,
        resume = function() smbus_task_resumed = true end
    }
    
    -- 模拟NCSI配置对象
    local scheduler_mock = {
        on_data_change = {
            on = function(self, callback)
                self.callback = callback
            end
        },
        on_error = {
            on = function(self, callback)
                self.error_callback = callback
            end
        },
        start = function() end
    }
    
    c_network_port.ncsi_config_obj = {
        LinkStatus = function(params) return scheduler_mock end
    }
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_port:update_ncsi_link_status()
    
    -- 模拟NCSI连续失败达到阈值（NCSI_FAILED_THRESHOLD = 3）
    for i = 1, 3 do
        scheduler_mock.on_error.error_callback()
    end
    
    -- 验证结果
    lu.assertTrue(smbus_task_resumed)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_port.schedulers = original_schedulers
    c_network_port.smbus_schedulers = original_smbus_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.t_prop_without_bma = original_t_prop_without_bma
    c_network_port.ncsi_get_linkstatus_retry_count = original_ncsi_get_linkstatus_retry_count
    c_network_port.get_ncsi_link_status_succeess = original_get_ncsi_link_status_succeess
    c_network_port.LinkStatus = original_LinkStatus
    c_network_port.SpeedMbps = original_SpeedMbps
    c_network_port.AutoSpeedNegotiation = original_AutoSpeedNegotiation
    c_network_port.FullDuplex = original_FullDuplex
end

-- 测试update_MAC_by_ncsi函数的NCSI恢复时停止SMBUS fallback逻辑
function TEST_network_port:test_update_MAC_by_ncsi_ncsi_recovered_stops_smbus_fallback()
    -- 备份原始状态
    local original_schedulers = c_network_port.schedulers
    local original_smbus_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_get_parent = c_network_port.get_parent
    local original_get_properties_res = c_network_port.get_properties_res
    local original_MACAddress = c_network_port.MACAddress
    
    -- 初始化测试状态
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.get_properties_res = {}
    
    -- 模拟父对象
    c_network_port.get_parent = function()
        return { Model = 'BroadCom' }
    end
    
    -- 模拟SMBUS任务正在运行
    local smbus_task_paused = false
    c_network_port.smbus_schedulers.mac_address_task = {
        is_paused = false,
        pause = function() smbus_task_paused = true end,
        data = "some_data"
    }
    
    -- 模拟NCSI配置对象
    local scheduler_mock = {
        on_data_change = {
            on = function(self, callback)
                self.callback = callback
            end
        },
        on_error = {
            on = function(self, callback)
                self.error_callback = callback
            end
        },
        start = function() end
    }
    
    c_network_port.ncsi_config_obj = {
        MacAddrNCSI = function(params) return scheduler_mock end
    }
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_port:update_MAC_by_ncsi()
    
    -- 模拟NCSI数据变化（恢复正常）
    local mac_data = 'AA:BB:CC:DD:EE:FF'
    scheduler_mock.on_data_change.callback(mac_data)
    
    -- 验证结果
    lu.assertEquals(c_network_port.MACAddress, mac_data)
    lu.assertTrue(c_network_port.get_properties_res.MacAddressFromNcsi)
    lu.assertTrue(smbus_task_paused)
    lu.assertNil(c_network_port.smbus_schedulers.mac_address_task.data)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_port.schedulers = original_schedulers
    c_network_port.smbus_schedulers = original_smbus_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.get_parent = original_get_parent
    c_network_port.get_properties_res = original_get_properties_res
    c_network_port.MACAddress = original_MACAddress
end

-- 测试update_MAC_by_ncsi函数的NCSI失败时启动SMBUS fallback逻辑
function TEST_network_port:test_update_MAC_by_ncsi_ncsi_failed_activates_smbus_fallback()
    -- 备份原始状态
    local original_schedulers = c_network_port.schedulers
    local original_smbus_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_get_parent = c_network_port.get_parent
    local original_get_properties_res = c_network_port.get_properties_res
    local original_MACAddress = c_network_port.MACAddress
    
    -- 初始化测试状态
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.get_properties_res = {MacAddressFromNcsi = true}
    
    -- 模拟父对象
    c_network_port.get_parent = function()
        return { Model = 'BroadCom' }
    end
    
    -- 模拟SMBUS任务已暂停
    local smbus_task_resumed = false
    c_network_port.smbus_schedulers.mac_address_task = {
        is_paused = true,
        resume = function() smbus_task_resumed = true end
    }
    
    -- 模拟NCSI配置对象
    local scheduler_mock = {
        on_data_change = {
            on = function(self, callback)
                self.callback = callback
            end
        },
        on_error = {
            on = function(self, callback)
                self.error_callback = callback
            end
        },
        start = function() end
    }
    
    c_network_port.ncsi_config_obj = {
        MacAddrNCSI = function(params) return scheduler_mock end
    }
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_port:update_MAC_by_ncsi()
    
    -- 模拟NCSI连续失败达到阈值（NCSI_FAILED_THRESHOLD = 3）
    for i = 1, 3 do
        scheduler_mock.on_error.error_callback()
    end
    
    -- 验证结果
    lu.assertFalse(c_network_port.get_properties_res.MacAddressFromNcsi)
    lu.assertTrue(smbus_task_resumed)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_port.schedulers = original_schedulers
    c_network_port.smbus_schedulers = original_smbus_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.get_parent = original_get_parent
    c_network_port.get_properties_res = original_get_properties_res
    c_network_port.MACAddress = original_MACAddress
end

-- 测试update_BDF_by_ncsi函数的while循环逻辑
function TEST_network_port:test_update_BDF_by_ncsi_while_loop_logic()
    -- 备份原始状态
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_bdf_task_is_running = c_network_port.bdf_task_is_running
    local original_next_tick = c_network_port.next_tick
    local original_sleep_ms = c_network_port.sleep_ms
    local original_WorkloadType = c_network_port.WorkloadType
    local original_BDF = c_network_port.BDF
    local original_t_prop_without_bma = c_network_port.t_prop_without_bma
    
    -- 初始化测试状态
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.bdf_task_is_running = false
    c_network_port.WorkloadType = 0
    c_network_port.t_prop_without_bma = {}
    -- 模拟BDF获取成功的情况
    local call_count = 0
    c_network_port.ncsi_config_obj = {
        BDF = function(params)
            call_count = call_count + 1
            return {
                value = function()
                    if call_count == 1 then
                        return { bus = 0, device = 0, func = 0 } -- 无效BDF
                    else
                        return { bus = 1, device = 2, func = 3 } -- 有效BDF
                    end
                end
            }
        end
    }
    
    -- 模拟next_tick和sleep_ms
    local task_executed = false
    c_network_port.next_tick = function(self, func)
        task_executed = true
        func() -- 立即执行任务
    end
    
    c_network_port.sleep_ms = function()
    end
    
    -- 执行函数
    c_network_port:update_BDF_by_ncsi()
    
    -- 验证结果
    lu.assertTrue(task_executed)
    lu.assertEquals(call_count, 2) -- 应该调用两次：第一次无效，第二次有效
    lu.assertEquals(c_network_port.BDF, '0000:01:02.3')
    lu.assertFalse(c_network_port.bdf_task_is_running)
    
    -- 还原全局状态
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.bdf_task_is_running = original_bdf_task_is_running
    c_network_port.next_tick = original_next_tick
    c_network_port.sleep_ms = original_sleep_ms
    c_network_port.WorkloadType = original_WorkloadType
    c_network_port.BDF = original_BDF
    c_network_port.t_prop_without_bma = original_t_prop_without_bma
end

-- 测试update_BDF_by_ncsi函数的重复调用保护
function TEST_network_port:test_update_BDF_by_ncsi_duplicate_call_protection()
    -- 备份原始状态
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_bdf_task_is_running = c_network_port.bdf_task_is_running
    local original_next_tick = c_network_port.next_tick
    
    -- 设置任务正在运行
    c_network_port.bdf_task_is_running = true
    c_network_port.ncsi_config_obj = {
        BDF = function(params)
            return { value = function() return { bus = 1, device = 2, func = 3 } end }
        end
    }
    
    -- 模拟next_tick
    local task_executed = false
    c_network_port.next_tick = function(self, func)
        task_executed = true
        func()
    end
    
    -- 执行函数
    c_network_port:update_BDF_by_ncsi()
    
    -- 验证结果：任务不应该被执行
    lu.assertFalse(task_executed)
    
    -- 还原全局状态
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.bdf_task_is_running = original_bdf_task_is_running
    c_network_port.next_tick = original_next_tick
end

-- 测试update_BDF_by_ncsi函数的BDF配置对象不存在的情况
function TEST_network_port:test_update_BDF_by_ncsi_no_bdf_config()
    -- 备份原始状态
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_bdf_task_is_running = c_network_port.bdf_task_is_running
    local original_next_tick = c_network_port.next_tick
    
    -- 设置BDF配置对象不存在
    c_network_port.ncsi_config_obj = {}
    c_network_port.bdf_task_is_running = false
    
    -- 模拟next_tick
    local task_executed = false
    c_network_port.next_tick = function(self, func)
        task_executed = true
        func()
    end
    
    -- 执行函数
    c_network_port:update_BDF_by_ncsi()
    
    -- 验证结果：任务不应该被执行
    lu.assertFalse(task_executed)
    
    -- 还原全局状态
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.bdf_task_is_running = original_bdf_task_is_running
    c_network_port.next_tick = original_next_tick
end

-- 测试当SMBUS任务已暂停时NCSI恢复不会重复暂停
function TEST_network_port:test_ncsi_recovered_when_smbus_already_paused()
    -- 备份原始状态
    local original_schedulers = c_network_port.schedulers
    local original_smbus_schedulers = c_network_port.smbus_schedulers
    local original_ncsi_config_obj = c_network_port.ncsi_config_obj
    local original_package_id = c_network_port.package_id
    local original_get_port_id = c_network_port.get_port_id
    local original_t_prop_without_bma = c_network_port.t_prop_without_bma
    local original_ncsi_get_linkstatus_retry_count = c_network_port.ncsi_get_linkstatus_retry_count
    local original_get_ncsi_link_status_succeess = c_network_port.get_ncsi_link_status_succeess
    local original_LinkStatus = c_network_port.LinkStatus
    local original_SpeedMbps = c_network_port.SpeedMbps
    local original_AutoSpeedNegotiation = c_network_port.AutoSpeedNegotiation
    local original_FullDuplex = c_network_port.FullDuplex
    
    -- 初始化测试状态
    c_network_port.schedulers = {}
    c_network_port.smbus_schedulers = {}
    c_network_port.package_id = 0
    c_network_port.get_port_id = function() return 1 end
    c_network_port.t_prop_without_bma = {}
    c_network_port.ncsi_get_linkstatus_retry_count = 0
    c_network_port.get_ncsi_link_status_succeess = false
    
    -- 模拟SMBUS任务已暂停
    local pause_call_count = 0
    c_network_port.smbus_schedulers.link_status_task = {
        is_paused = true,
        pause = function() pause_call_count = pause_call_count + 1 end,
        data = "some_data"
    }
    
    -- 模拟NCSI配置对象
    local scheduler_mock = {
        on_data_change = {
            on = function(self, callback)
                self.callback = callback
            end
        },
        on_error = {
            on = function(self, callback)
                self.error_callback = callback
            end
        },
        start = function() end
    }
    
    c_network_port.ncsi_config_obj = {
        LinkStatus = function(params) return scheduler_mock end
    }
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_port:update_ncsi_link_status()
    
    -- 模拟NCSI数据变化（恢复正常）
    local link_data = {
        LinkStatus = 'LinkUp',
        SpeedMbps = 1000,
        AutoSpeedNegotiation = true,
        FullDuplex = true
    }
    scheduler_mock.on_data_change.callback(link_data)
    
    -- 验证结果：不应该重复调用pause，因为任务已经暂停
    lu.assertEquals(pause_call_count, 0)
    lu.assertEquals(c_network_port.LinkStatus, link_data.LinkStatus)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_port.schedulers = original_schedulers
    c_network_port.smbus_schedulers = original_smbus_schedulers
    c_network_port.ncsi_config_obj = original_ncsi_config_obj
    c_network_port.package_id = original_package_id
    c_network_port.get_port_id = original_get_port_id
    c_network_port.ncsi_get_linkstatus_retry_count = original_ncsi_get_linkstatus_retry_count
    c_network_port.get_ncsi_link_status_succeess = original_get_ncsi_link_status_succeess
    c_network_port.LinkStatus = original_LinkStatus
    c_network_port.SpeedMbps = original_SpeedMbps
    c_network_port.AutoSpeedNegotiation = original_AutoSpeedNegotiation
    c_network_port.FullDuplex = original_FullDuplex
    c_network_port.t_prop_without_bma = original_t_prop_without_bma
end

-- 测试enable_lldp_over_mctp函数第一次就成功的场景
-- function TEST_network_port:test_enable_lldp_over_mctp()
--     -- 备份原始状态
--     local original_next_tick = c_network_port.next_tick
--     local original_ncsi_config_obj = c_network_port.ncsi_config_obj
--     local original_get_port_id = c_network_port.get_port_id
--     local original_package_id = c_network_port.package_id
--     local original_NodeId = c_network_port.NodeId
--     local log_module = require 'mc.logging'
--     local original_log_notice = log_module.notice
    
--     -- 初始化测试状态
--     c_network_port.get_port_id = function() return 1 end
--     c_network_port.package_id = 0
--     c_network_port.NodeId = 'test_node_1'
    
--     -- 打桩next_tick，立即执行函数
--     c_network_port.next_tick = function(self, func)
--         func() -- 立即执行
--     end
    
--     -- 打桩log:notice验证是否执行到
--     local log_notice_called = false
--     log_module.notice = function(self, fmt, ...)
--         if fmt and string.find(tostring(fmt), 'enable lldp over mctp successfully') then
--             log_notice_called = true
--         end
--         if original_log_notice then
--             return original_log_notice(self, fmt, ...)
--         end
--     end
    
--     -- 打桩SetOnLLDPOverMCTPEnable，value返回true
--     c_network_port.ncsi_config_obj = {
--         SetOnLLDPOverMCTPEnable = function(params)
--             return {
--                 value = function()
--                     return true
--                 end
--             }
--         end
--     }
    
--     -- 执行函数
--     c_network_port:enable_lldp_over_mctp()
    
--     -- 验证结果：log:notice应该被调用
--     lu.assertIsTrue(log_notice_called, 'log:notice should be called')
    
--     -- 还原原始状态
--     c_network_port.next_tick = original_next_tick
--     c_network_port.ncsi_config_obj = original_ncsi_config_obj
--     c_network_port.get_port_id = original_get_port_id
--     c_network_port.package_id = original_package_id
--     c_network_port.NodeId = original_NodeId
--     log_module.notice = original_log_notice
-- end

-- -- 测试 reset_bma_info 函数
-- function TEST_network_port:test_reset_bma_info()
--     local log_module = require 'mc.logging'
--     local original_log_notice = log_module.notice
--     local original_log_debug = log_module.debug
--     local original_restore_port_prop_without_bma = c_network_port.restore_port_prop_without_bma
    
--     -- 备份原始状态
--     local original_NetworkAdapterId = c_network_port.NetworkAdapterId
--     local original_PortID = c_network_port.PortID
--     local original_OSLinkStatus = c_network_port.OSLinkStatus
--     local original_Name = c_network_port.Name
--     local original_WorkMode = c_network_port.WorkMode
--     local original_DriverName = c_network_port.DriverName
--     local original_DriverVersion = c_network_port.DriverVersion
--     local original_SpeedMbps = c_network_port.SpeedMbps
--     local original_AutoSpeedNegotiation = c_network_port.AutoSpeedNegotiation
--     local original_MACAddress = c_network_port.MACAddress
--     local original_PermanentMACAddress = c_network_port.PermanentMACAddress
--     local original_FCId = c_network_port.FCId
--     local original_WWPN = c_network_port.WWPN
--     local original_WWNN = c_network_port.WWNN
--     local original_SpeedGbps = c_network_port.SpeedGbps
--     local original_FirmwareVersion = c_network_port.FirmwareVersion
--     local original_ncsi_config_obj = c_network_port.ncsi_config_obj
--     local original_t_prop_without_bma = c_network_port.t_prop_without_bma
    
--     -- 记录日志调用
--     local log_notice_calls = {}
--     local log_debug_calls = {}
--     log_module.notice = function(self, fmt, ...)
--         table.insert(log_notice_calls, {fmt = fmt, args = {...}})
--     end
--     log_module.debug = function(self, fmt, ...)
--         table.insert(log_debug_calls, {fmt = fmt, args = {...}})
--     end
    
--     -- 测试场景1: 基本属性重置和ncsi_config_obj为空时重置SpeedMbps和AutoSpeedNegotiation
--     c_network_port.NetworkAdapterId = "TestAdapter"
--     c_network_port.PortID = 1
--     c_network_port.OSLinkStatus = "Up"
--     c_network_port.Name = "eth0"
--     c_network_port.WorkMode = "Loop"
--     c_network_port.DriverName = "test_driver"
--     c_network_port.DriverVersion = "1.0.0"
--     c_network_port.SpeedMbps = 1000
--     c_network_port.AutoSpeedNegotiation = true
--     c_network_port.ncsi_config_obj = {}
--     c_network_port.t_prop_without_bma = {LinkStatus = 'N/A'}
    
--     c_network_port.restore_port_prop_without_bma = function(self)
--         self.LinkStatus = 'N/A'
--     end
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.OSLinkStatus, '', "OSLinkStatus should be reset to empty")
--     lu.assertEquals(c_network_port.Name, '', "Name should be reset to empty")
--     lu.assertEquals(c_network_port.WorkMode, 'NonLoop', "WorkMode should be reset to NonLoop")
--     lu.assertEquals(c_network_port.DriverName, '', "DriverName should be reset to empty")
--     lu.assertEquals(c_network_port.DriverVersion, '', "DriverVersion should be reset to empty")
--     lu.assertEquals(c_network_port.SpeedMbps, 0, "SpeedMbps should be reset to 0 when ncsi_config_obj is empty")
--     lu.assertEquals(c_network_port.AutoSpeedNegotiation, false, "AutoSpeedNegotiation should be reset to false when ncsi_config_obj is empty")
--     lu.assertEquals(#log_debug_calls, 1, "log:debug should be called once")
--     lu.assertEquals(log_debug_calls[1].fmt, 'NetworkPort(%sPort%s) reset from bma complete')
    
--     -- 测试场景2: ncsi_config_obj有LinkStatus时不重置SpeedMbps和AutoSpeedNegotiation
--     c_network_port.SpeedMbps = 1000
--     c_network_port.AutoSpeedNegotiation = true
--     c_network_port.ncsi_config_obj = {LinkStatus = {}}
--     log_debug_calls = {}
--     log_notice_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.SpeedMbps, 1000, "SpeedMbps should not be reset when ncsi_config_obj has LinkStatus")
--     lu.assertEquals(c_network_port.AutoSpeedNegotiation, true, "AutoSpeedNegotiation should not be reset when ncsi_config_obj has LinkStatus")
    
--     -- 测试场景3: PermanentMACAddress有效时从PermanentMACAddress恢复MACAddress
--     c_network_port.PermanentMACAddress = "AA:BB:CC:DD:EE:FF"
--     c_network_port.MACAddress = "11:22:33:44:55:66"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.MACAddress, "AA:BB:CC:DD:EE:FF", "MACAddress should be restored from PermanentMACAddress")
--     lu.assertEquals(#log_notice_calls, 1, "log:notice should be called once for MAC address reset")
--     lu.assertEquals(log_notice_calls[1].fmt, 'NetworkPort(%sPort%s) macaddr reset from permanent mac(%s) complete')
--     lu.assertEquals(log_notice_calls[1].args[1], "TestAdapter")
--     lu.assertEquals(log_notice_calls[1].args[2], 1)
--     lu.assertEquals(log_notice_calls[1].args[3], "AA:BB:CC:DD:EE:FF")
    
--     -- 测试场景4: PermanentMACAddress为默认值时不恢复MACAddress
--     c_network_port.PermanentMACAddress = "00:00:00:00:00:00"
--     c_network_port.MACAddress = "11:22:33:44:55:66"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.MACAddress, "11:22:33:44:55:66", "MACAddress should not be restored when PermanentMACAddress is default")
--     lu.assertEquals(#log_notice_calls, 0, "log:notice should not be called when PermanentMACAddress is default")
    
--     -- 测试场景5: PermanentMACAddress为无效值时不恢复MACAddress
--     c_network_port.PermanentMACAddress = "N/A"
--     c_network_port.MACAddress = "11:22:33:44:55:66"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.MACAddress, "11:22:33:44:55:66", "MACAddress should not be restored when PermanentMACAddress is invalid")
--     lu.assertEquals(#log_notice_calls, 0, "log:notice should not be called when PermanentMACAddress is invalid")
    
--     -- 测试场景6: PermanentMACAddress为nil时不恢复MACAddress
--     c_network_port.PermanentMACAddress = nil
--     c_network_port.MACAddress = "11:22:33:44:55:66"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.MACAddress, "11:22:33:44:55:66", "MACAddress should not be restored when PermanentMACAddress is nil")
--     lu.assertEquals(#log_notice_calls, 0, "log:notice should not be called when PermanentMACAddress is nil")
    
--     -- 测试场景7: FC卡相关属性重置
--     c_network_port.FCId = "FC123"
--     c_network_port.WWPN = "WWPN123"
--     c_network_port.WWNN = "WWNN123"
--     c_network_port.SpeedGbps = 16
--     c_network_port.FirmwareVersion = "2.0.0"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.WWPN, '', "WWPN should be reset to empty for FC card")
--     lu.assertEquals(c_network_port.WWNN, '', "WWNN should be reset to empty for FC card")
--     lu.assertEquals(c_network_port.SpeedGbps, 0, "SpeedGbps should be reset to 0 for FC card")
--     lu.assertEquals(c_network_port.FCId, '', "FCId should be reset to empty for FC card")
--     lu.assertEquals(c_network_port.FirmwareVersion, '', "FirmwareVersion should be reset to empty for FC card")
    
--     -- 测试场景8: FCId为空字符串时不重置FC相关属性
--     c_network_port.FCId = ""
--     c_network_port.WWPN = "WWPN123"
--     c_network_port.WWNN = "WWNN123"
--     c_network_port.SpeedGbps = 16
--     c_network_port.FirmwareVersion = "2.0.0"
--     log_notice_calls = {}
--     log_debug_calls = {}
    
--     c_network_port:reset_bma_info()
    
--     lu.assertEquals(c_network_port.WWPN, "WWPN123", "WWPN should not be reset when FCId is empty")
--     lu.assertEquals(c_network_port.WWNN, "WWNN123", "WWNN should not be reset when FCId is empty")
--     lu.assertEquals(c_network_port.SpeedGbps, 16, "SpeedGbps should not be reset when FCId is empty")
--     lu.assertEquals(c_network_port.FirmwareVersion, "2.0.0", "FirmwareVersion should not be reset when FCId is empty")
    
--     -- 还原原始状态
--     c_network_port.restore_port_prop_without_bma = original_restore_port_prop_without_bma
--     c_network_port.NetworkAdapterId = original_NetworkAdapterId
--     c_network_port.PortID = original_PortID
--     c_network_port.OSLinkStatus = original_OSLinkStatus
--     c_network_port.Name = original_Name
--     c_network_port.WorkMode = original_WorkMode
--     c_network_port.DriverName = original_DriverName
--     c_network_port.DriverVersion = original_DriverVersion
--     c_network_port.SpeedMbps = original_SpeedMbps
--     c_network_port.AutoSpeedNegotiation = original_AutoSpeedNegotiation
--     c_network_port.MACAddress = original_MACAddress
--     c_network_port.PermanentMACAddress = original_PermanentMACAddress
--     c_network_port.FCId = original_FCId
--     c_network_port.WWPN = original_WWPN
--     c_network_port.WWNN = original_WWNN
--     c_network_port.SpeedGbps = original_SpeedGbps
--     c_network_port.FirmwareVersion = original_FirmwareVersion
--     c_network_port.ncsi_config_obj = original_ncsi_config_obj
--     c_network_port.t_prop_without_bma = original_t_prop_without_bma
--     log_module.notice = original_log_notice
--     log_module.debug = original_log_debug
-- end

-- 覆盖 restore_*_after_bma_stop、restore_port_prop_without_bma、reset_bma_info 中非 BMA 属性恢复（network_port.lua 101-148, 817-858）
local function make_np_bma_stub(overrides)
    local p = {
        NetworkAdapterId = 'UT',
        PortID = 1,
        device_path = '/ut/pcie/p0',
        CreatedByDeviceObject = false,
        FCId = '',
        t_prop_without_bma = {},
        OSLinkStatus = 'Up',
        Name = 'eth0',
        WorkMode = 'Loop',
        DriverName = 'drv',
        DriverVersion = '1.0',
        SpeedMbps = 9999,
        AutoSpeedNegotiation = 1,
        FullDuplex = 1,
        BDF = 'bdf_bma',
        FirmwareVersion = 'fw_bma',
        MACAddress = '11:22:33:44:55:66',
        PermanentMACAddress = nil,
    }
    if overrides then
        for k, v in pairs(overrides) do
            p[k] = v
        end
    end
    setmetatable(p, { __index = c_network_port })
    return p
end

function TEST_network_port:test_restore_speed_mbps_after_bma_stop_uses_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.SpeedMbps = 2500
    c_network_port.restore_speed_mbps_after_bma_stop(p)
    lu.assertEquals(p.SpeedMbps, 2500)
end

function TEST_network_port:test_restore_speed_mbps_after_bma_stop_default_uint_max()
    local p = make_np_bma_stub()
    c_network_port.restore_speed_mbps_after_bma_stop(p)
    lu.assertEquals(p.SpeedMbps, 4294967295)
end

function TEST_network_port:test_restore_auto_speed_negotiation_after_bma_stop_uses_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.AutoSpeedNegotiation = 2
    c_network_port.restore_auto_speed_negotiation_after_bma_stop(p)
    lu.assertEquals(p.AutoSpeedNegotiation, 2)
end

function TEST_network_port:test_restore_auto_speed_negotiation_after_bma_stop_default_255()
    local p = make_np_bma_stub()
    p.AutoSpeedNegotiation = 1
    c_network_port.restore_auto_speed_negotiation_after_bma_stop(p)
    lu.assertEquals(p.AutoSpeedNegotiation, 255)
end

function TEST_network_port:test_restore_full_duplex_after_bma_stop_uses_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.FullDuplex = 0
    c_network_port.restore_full_duplex_after_bma_stop(p)
    lu.assertEquals(p.FullDuplex, 0)
end

function TEST_network_port:test_restore_full_duplex_after_bma_stop_default_255()
    local p = make_np_bma_stub()
    c_network_port.restore_full_duplex_after_bma_stop(p)
    lu.assertEquals(p.FullDuplex, 255)
end

function TEST_network_port:test_restore_bdf_after_bma_stop_uses_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.BDF = '0000:03:00.0'
    c_network_port.restore_bdf_after_bma_stop(p)
    lu.assertEquals(p.BDF, '0000:03:00.0')
end

function TEST_network_port:test_restore_bdf_after_bma_stop_default_empty()
    local p = make_np_bma_stub()
    c_network_port.restore_bdf_after_bma_stop(p)
    lu.assertEquals(p.BDF, '')
end

function TEST_network_port:test_restore_firmware_version_after_bma_stop_uses_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.FirmwareVersion = 'SR_1.2.3'
    c_network_port.restore_firmware_version_after_bma_stop(p)
    lu.assertEquals(p.FirmwareVersion, 'SR_1.2.3')
end

function TEST_network_port:test_restore_firmware_version_after_bma_stop_default_empty()
    local p = make_np_bma_stub()
    c_network_port.restore_firmware_version_after_bma_stop(p)
    lu.assertEquals(p.FirmwareVersion, '')
end

function TEST_network_port:test_restore_port_prop_without_bma_restores_all_from_cache()
    local p = make_np_bma_stub()
    p.t_prop_without_bma.SpeedMbps = 100
    p.t_prop_without_bma.AutoSpeedNegotiation = 2
    p.t_prop_without_bma.FullDuplex = 3
    p.t_prop_without_bma.BDF = 'bdf_ok'
    p.t_prop_without_bma.FirmwareVersion = 'fw_ok'
    c_network_port.restore_port_prop_without_bma(p)
    lu.assertEquals(p.SpeedMbps, 100)
    lu.assertEquals(p.AutoSpeedNegotiation, 2)
    lu.assertEquals(p.FullDuplex, 3)
    lu.assertEquals(p.BDF, 'bdf_ok')
    lu.assertEquals(p.FirmwareVersion, 'fw_ok')
end

function TEST_network_port:test_reset_bma_info_non_device_object_restores_firmware()
    local p = make_np_bma_stub({ CreatedByDeviceObject = false })
    p.t_prop_without_bma.FirmwareVersion = 'cached_fw'
    p.FirmwareVersion = 'bma_fw'
    c_network_port.reset_bma_info(p)
    lu.assertEquals(p.DriverName, '')
    lu.assertEquals(p.FirmwareVersion, 'cached_fw')
end

function TEST_network_port:test_reset_bma_info_device_object_missing_returns_early()
    local p = make_np_bma_stub({ CreatedByDeviceObject = true })
    p.FirmwareVersion = 'keep_fw'
    local orig_gi = c_object_manage.get_instance
    local orig_gdo = device_comm_fun.get_device_obj_by_path_interface
    c_object_manage.get_instance = function()
        return { bus = {} }
    end
    device_comm_fun.get_device_obj_by_path_interface = function()
        return nil
    end
    c_network_port.reset_bma_info(p)
    lu.assertEquals(p.Name, '')
    lu.assertEquals(p.FirmwareVersion, 'keep_fw')
    c_object_manage.get_instance = orig_gi
    device_comm_fun.get_device_obj_by_path_interface = orig_gdo
end

function TEST_network_port:test_reset_bma_info_device_object_restore_prop_pcall_fail()
    local p = make_np_bma_stub({ CreatedByDeviceObject = true })
    p.FirmwareVersion = 'keep_fw'
    local orig_gi = c_object_manage.get_instance
    local orig_gdo = device_comm_fun.get_device_obj_by_path_interface
    local mock_bus = {}
    c_object_manage.get_instance = function()
        return { bus = mock_bus }
    end
    device_comm_fun.get_device_obj_by_path_interface = function(bus, path, iface)
        lu.assertIs(bus, mock_bus)
        lu.assertEquals(path, p.device_path)
        lu.assertEquals(iface, device_comm_defs.NETWORK_PORT_OEM_INTERFACE)
        return {
            call_method = function()
                error('restore_fail')
            end,
        }
    end
    c_network_port.reset_bma_info(p)
    lu.assertEquals(p.FirmwareVersion, 'keep_fw')
    c_object_manage.get_instance = orig_gi
    device_comm_fun.get_device_obj_by_path_interface = orig_gdo
end

function TEST_network_port:test_reset_bma_info_device_object_restore_prop_ok()
    local restore_called = false
    local p = make_np_bma_stub({ CreatedByDeviceObject = true })
    p.FirmwareVersion = 'keep_fw'
    local orig_gi = c_object_manage.get_instance
    local orig_gdo = device_comm_fun.get_device_obj_by_path_interface
    local mock_bus = {}
    c_object_manage.get_instance = function()
        return { bus = mock_bus }
    end
    device_comm_fun.get_device_obj_by_path_interface = function(bus, path, iface)
        lu.assertIs(bus, mock_bus)
        lu.assertEquals(path, p.device_path)
        lu.assertEquals(iface, device_comm_defs.NETWORK_PORT_OEM_INTERFACE)
        return {
            call_method = function(_, method, sig)
                if method == 'RestorePropAfterBmaStop' and sig == '' then
                    restore_called = true
                    return true
                end
                return false
            end,
        }
    end
    c_network_port.reset_bma_info(p)
    lu.assertTrue(restore_called)
    lu.assertEquals(p.FirmwareVersion, 'keep_fw')
    c_object_manage.get_instance = orig_gi
    device_comm_fun.get_device_obj_by_path_interface = orig_gdo
end

local cnt = 0
function TEST_network_port:test_update_npu_heartbeat_status()
    local old_loop = skynet.fork_loop
    skynet.fork_loop = function(test, cb)
        cb()
    end
    local old_f1 = fructl.get_power_status
    fructl.get_power_status = function ()
        if cnt == 0 then
            cnt = cnt + 1
            return 'OFF'
        end
        mock_network_port.PowerOn = 1
        return 'ON'
    end
    local op = {}
    op.set_npu_heartbeat_loss_status = function ()
        
    end
    op.set_dmp_ready_status = function ()
        error("error")
    end
    mock_network_port.get_optical_module = function ()
        return op        
    end
    local old_check= npu_imu_cmd.heartbeat_check
    npu_imu_cmd.heartbeat_check = function ()
        return true, 0
    end
    mock_network_port.PowerOn = 1
    pcall(c_network_port.update_npu_heartbeat_status, mock_network_port)
    lu.assertIsTrue(c_network_port.is_dmp_status_ready)

    npu_imu_cmd.heartbeat_check = function ()
        return true, 3
    end
    mock_network_port.PowerOn = 0
    pcall(c_network_port.update_npu_heartbeat_status, mock_network_port)
    lu.assertIsTrue(c_network_port.is_dmp_status_ready)
    npu_imu_cmd.heartbeat_check = old_check
    skynet.fork_loop = old_loop
     fructl.get_power_status = old_f1
end

-- 测试 NpuID 处理逻辑：ProcessorPorts 前缀替换
function TEST_network_port:test_update_processor_ports()
    c_network_port.NpuID = "2"
    c_network_port.ProcessorPorts = {
        {"", "PortA"},
        {"", "PortB"},
        {"", "PortC"}
    }
    c_network_port:update_processor_ports()

    -- 断言：第一个字段全部变成 NPU2，第二个字段保持不变
    lu.assertEquals(c_network_port.ProcessorPorts[1][1], "NPU2")
    lu.assertEquals(c_network_port.ProcessorPorts[2][1], "NPU2")
    lu.assertEquals(c_network_port.ProcessorPorts[3][1], "NPU2")

    lu.assertEquals(c_network_port.ProcessorPorts[1][2], "PortA")
    lu.assertEquals(c_network_port.ProcessorPorts[2][2], "PortB")
    lu.assertEquals(c_network_port.ProcessorPorts[3][2], "PortC")

    c_network_port.NpuID = "255"
    c_network_port.ProcessorPorts = {
        {"", "PortA"}
    }
    c_network_port:update_processor_ports()

    -- 断言：保持原样
    lu.assertEquals(c_network_port.ProcessorPorts[1][1], "NPU255")

    c_network_port.NpuID = nil
    c_network_port.ProcessorPorts = {
        {"Original1", "PortA"}
    }
    c_network_port:update_processor_ports()
    lu.assertEquals(c_network_port.ProcessorPorts[1][1], "Original1")

    -- 测试 NpuID 非数字字符串：不处理
    c_network_port.NpuID = "abc"
    c_network_port.ProcessorPorts = {
        {"", "PortA"}
    }
    c_network_port:update_processor_ports()
    lu.assertEquals(c_network_port.ProcessorPorts[1][1], "")

    -- 测试空 ProcessorPorts：不崩溃
    c_network_port.NpuID = "3"
    c_network_port.ProcessorPorts = {}

    c_network_port:update_processor_ports()
    lu.assertEquals(#c_network_port.ProcessorPorts, 0)
end

-- 测试 NPU_UB 模型的心跳状态更新
function TEST_network_port:test_update_npu_heartbeat_status_npu_ub()
    local old_loop = skynet.fork_loop
    skynet.fork_loop = function(test, cb)
        cb()
    end
    local old_f1 = fructl.get_power_status
    fructl.get_power_status = function()
        return 'ON'
    end

    local op = {}
    op.set_npu_heartbeat_loss_status = function()
    end
    op.set_dmp_ready_status = function()
    end
    mock_network_port.get_optical_module = function()
        return op
    end

    -- 模拟 NPU_UB 模型
    mock_network_port.get_parent = function()
        return {
            Model = 'NPU_UB'
        }
    end
    mock_network_port.PowerOn = 1
    mock_network_port.PortID = 1

    -- 测试 NPU_UB 模型不调用 heartbeat_check
    local old_check = npu_imu_cmd.heartbeat_check
    npu_imu_cmd.heartbeat_check = function()
        error("heartbeat_check should not be called for NPU_UB")
    end
    pcall(c_network_port.update_npu_heartbeat_status, mock_network_port)

    npu_imu_cmd.heartbeat_check = old_check
    skynet.fork_loop = old_loop
    fructl.get_power_status = old_f1
    lu.assertTrue(true)
end

-- 测试 update_npu_heartbeat_status_by_presence 函数
function TEST_network_port:test_update_npu_heartbeat_status_by_presence()
    -- 创建模拟光模块
    local mock_op1 = {
        RelatedNetworkPorts = {1, 2, 3, 4},
        Presence = 1,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }
    local mock_op2 = {
        RelatedNetworkPorts = {5, 6, 7, 8},
        Presence = 0,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }
    local mock_op3 = {
        RelatedNetworkPorts = {1, 2},
        Presence = 1,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }

    -- 保存原始的 collection.fold
    local old_fold = c_optical_module.collection and c_optical_module.collection.fold

    -- 模拟 collection.fold
    c_optical_module.collection = {
        fold = function(self, callback)
            callback(nil, mock_op1)
            callback(nil, mock_op2)
            callback(nil, mock_op3)
        end
    }

    local test_port = {
        PortID = 1,
        is_npu_heartbeat_loss = false,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }

    -- 测试场景1: 有关联光模块在位 (Presence = 1)
    pcall(c_network_port.update_npu_heartbeat_status_by_presence, test_port)
    lu.assertEquals(test_port.is_npu_heartbeat_loss, false)

    -- 测试场景2: 有关联光模块不在位 (Presence = 0)
    test_port.PortID = 5
    test_port.is_npu_heartbeat_loss = true
    pcall(c_network_port.update_npu_heartbeat_status_by_presence, test_port)
    lu.assertEquals(test_port.is_npu_heartbeat_loss, true)

    -- 恢复原始
    if old_fold then
        c_optical_module.collection.fold = old_fold
    end
    lu.assertTrue(true)
end

-- 测试 update_npu_heartbeat_status_by_presence 函数 - 空光模块表场景
function TEST_network_port:test_update_npu_heartbeat_status_by_presence_empty_table()
    -- 保存原始的 collection.fold
    local old_fold = c_optical_module.collection and c_optical_module.collection.fold

    -- 模拟 collection.fold 不返回任何光模块 (callback不被调用)
    c_optical_module.collection.fold = function(self, callback)
    end

    local test_port = {
        PortID = 99,
        is_npu_heartbeat_loss = false,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }

    -- 测试场景: 没有关联光模块
    local ok, err = pcall(c_network_port.update_npu_heartbeat_status_by_presence, test_port)
    lu.assertTrue(ok, err)
    lu.assertEquals(test_port.is_npu_heartbeat_loss, true)

    -- 恢复原始
    if old_fold then
        c_optical_module.collection.fold = old_fold
    end
    lu.assertTrue(true)
end

-- 测试 update_npu_heartbeat_status_by_presence 函数 - 所有光模块都不在位
function TEST_network_port:test_update_npu_heartbeat_status_by_presence_all_absent()
    local mock_op1 = {
        RelatedNetworkPorts = {10, 11},
        Presence = 0,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }
    local mock_op2 = {
        RelatedNetworkPorts = {10, 12},
        Presence = 0,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }

    -- 保存原始的 collection.fold
    local old_fold = c_optical_module.collection and c_optical_module.collection.fold

    -- 模拟 collection.fold 返回多个不在位的光模块
    c_optical_module.collection.fold = function(self, callback)
        callback(nil, mock_op1)
        callback(nil, mock_op2)
    end

    local test_port = {
        PortID = 10,
        is_npu_heartbeat_loss = false,
        set_npu_heartbeat_loss_status = function(self, status)
        end
    }

    -- 测试场景: 所有关联光模块都不在位 (触发 heartbeat_loss_flag = true 分支)
    local ok, err = pcall(c_network_port.update_npu_heartbeat_status_by_presence, test_port)
    lu.assertTrue(ok, err)
    lu.assertEquals(test_port.is_npu_heartbeat_loss, true)

    -- 恢复原始
    if old_fold then
        c_optical_module.collection.fold = old_fold
    end
    lu.assertTrue(true)
end