-- 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 network_adapter_app = require 'network_adapter_app'
local c_network_adapter = require 'device.class.network_adapter'
local c_object_manage = require 'mc.orm.object_manage'
local c_tasks = require 'mc.orm.tasks'
local c_network_port = require 'device.class.network_port'
local c_network_adapter_db = require 'network_adapter.db'
local hook_tasks = require 'test_common.hook_tasks'
local test_utils = require 'test_utils'
local create_mds_object = test_utils.create_mds_object
local vos = require 'utils.vos'
local utils = require 'mc.utils'
local ipmi = require 'ipmi'
local cc = ipmi.types.Cc
local c_optical_module = require 'device.class.optical_module'
local hi1822 = require 'hardware_config.Hi182X'
local hi1822_fc = require 'hardware_config.Hi1822-FC'
local bf2 = require 'hardware_config.BF2'
local cx4 = require 'hardware_config.CX4'
local cx5 = require 'hardware_config.CX5'
local cx7 = require 'hardware_config.CX7'
local cx6_dx = require 'hardware_config.CX6-DX'
local bf3 = require 'hardware_config.BF3'
local bs = require 'mc.bitstring'
local c_network_bonding = require 'device.class.network_bonding'
local c_handler_eth = require 'bma.handles.handler_eth'
local c_handler_fc = require 'bma.handles.handler_fc'
local pmu_cmd = require 'imu'
local skynet = require 'skynet'
local mctp_lib = require 'mctp_lib'

TEST_network_adapter = {}

function TEST_network_adapter:test_RequireConfig()
    local table_1822 = hi1822.smbus().properties
    for _, func in pairs(table_1822) do
        pcall(func.response)
    end
    local table_1822_ncsi = hi1822.mctp().properties
    for _, func in pairs(table_1822_ncsi) do
        pcall(func.response)
    end
    local table_cx4 = cx4.mctp().properties
    for _, func in pairs(table_cx4) do
        pcall(func.response)
    end
    local table_cx5 = cx5.mctp().properties
    for _, func in pairs(table_cx5) do
        pcall(func.response)
    end
    local table_cx6_dx = cx6_dx.mctp().properties
    for _, func in pairs(table_cx6_dx) do
        pcall(func.response)
    end
    local table_cx7 = cx7.mctp().properties
    for _, func in pairs(table_cx7) do
        pcall(func.response)
    end
    local table_bf3 = bf3.mctp().properties
    for _, func in pairs(table_bf3) do
        pcall(func.response)
    end
end

function TEST_network_adapter:setUp()
    self.database = c_network_adapter_db(':memory:')

    hook_tasks.hook()
    self.object_manage = c_object_manage.new(self.database)
    self.object_manage.app = self

    self.position = '00010102'
    self.network_adapter_obj = test_utils.add_network_adapter(self.object_manage, 'NetworkAdapter_1', {
        BoardID = 0x123,
        DeviceLocator = 'some locator',
        Name = "",
        Position = self.position
    })
    self.object_manage.mc:before_add_object(c_network_adapter.__class_name, self.network_adapter_obj)
    self.object_manage.mc:add_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)
    self.object_manage.mc:prepare_ok()
    self:flush()
end

function TEST_network_adapter:CreateVLANs()
end

function TEST_network_adapter:tearDown()
    c_object_manage.destroy()
    self.database.db:close()
    hook_tasks.unhook()
end

function TEST_network_adapter:flush()
    c_tasks.get_instance():run_all_task()
    self.database.db:flush()
end

function TEST_network_adapter:test_can_construct_network_adapter()
    local adapter = c_network_adapter.collection:find({Position = self.position})

    lu.assertEquals(adapter.Position, self.position)
    lu.assertEquals(adapter.ID, self.position .. 'somelocator')
    lu.assertEquals(adapter.BoardIDHex, '0x0123')
end

function TEST_network_adapter:test_can_deconstruct_network_adapter()
    local adapter = c_network_adapter.collection:find({Position = self.position})

    local r = {ports_deconstruct_called = 0, scheduler_deconstruct_called = 0}
    c_network_port.stop = function()
        r.ports_deconstruct_called = 1
    end

    -- Override dtor to only invoke stop, avoiding VLANs cleanup logic which is
    -- outside the scope of this test.
    local origin_dtor = c_network_port.dtor
    c_network_port.dtor = function(self)
        self:stop()
    end

    -- 自发现添加 port 对象
    local mds_network_port = create_mds_object('NetworkPort_1', {PortID = 1, NetDevFuncType = 1},
        self.network_adapter_obj,
        '/bmc/kepler/Systems/1/NetworkAdapters/NetworkAdapter_1_0001010101')
    self.object_manage.mc:before_add_object(c_network_port.__class_name, mds_network_port)
    self.object_manage.mc:add_object(c_network_port.__class_name, mds_network_port, self.position)

    local scheduler_expect_call<const> = 10
    for i = 1, scheduler_expect_call do
        adapter.ncsi_schedulers[i] = {
            deconstruct = function()
                r.scheduler_deconstruct_called = r.scheduler_deconstruct_called + 1
            end
        }
    end
    self.object_manage.mc:add_object_complete(self.position)

    -- 自发现删除对象
    self.object_manage.mc:del_object(c_network_port.__class_name, mds_network_port, self.position)
    self.object_manage.mc:del_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)

    -- Restore original destructor
    c_network_port.dtor = origin_dtor
    lu.assertEquals(r.ports_deconstruct_called, 1)
    lu.assertEquals(r.scheduler_deconstruct_called, scheduler_expect_call)
end

function TEST_network_adapter:test_network_adapter_collect_dump_info()
    self.network_adapter_obj.Name = 'SP580'
    self.network_portobj = test_utils.add_network_port(self.object_manage, 'NetworkPort_1',
        {PortID = 0, PermanentMACAddress = '04:8C:16:59:C9:3E', MACAddress = '04:8C:16:59:C9:3E',
            LinkStatus = 'LinkDown'},
        self.position, self.network_adapter_obj)
    self.optical_module_obj = test_utils.add_optical_module(self.object_manage, 'OpticalModule_1',
        {FaultState = 0, Presence = 0, Manufacturer = 'Huawei', SerialNumber = 'INHAX4480013'},
        self.position, self.network_portobj)
    test_utils.add_object_complete(self.object_manage, self.position)
    local bma = {handles = {[3] = {objects = {[self.network_portobj] =
        '/redfish/v1/Sms/1/Systems/1/EthernetInterfaces/0000:bc:00.0_0000:bd:00.0'}}}}

    -- Avoid VLANs-related destructor logic which is not under test here.
    local origin_dtor = c_network_port.dtor
    c_network_port.dtor = function(self)
        self:stop()
    end

    c_network_adapter.collect_dump_info('/var/log', bma)
    lu.assertEquals(vos.get_file_accessible('/var/log/net_card_info'), true)
    self.object_manage.mc:del_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)
    self.object_manage.mc:del_object(c_network_port.__class_name, self.network_portobj, self.position)
    self.object_manage.mc:del_object(c_optical_module.__class_name, self.optical_module_obj, self.position)
    utils.remove_file('/var/log/net_card_info')

    c_network_adapter.collect_dump_info('/var/log', bma)
    lu.assertEquals(vos.get_file_accessible('/var/log/net_card_info'), true)

    -- Restore original destructor
    c_network_port.dtor = origin_dtor
end

function TEST_network_adapter:test_network_adapter_collect_static_optical_info()
    local col_tab = c_optical_module.collection 
    local opt_info = {
        PortID = 0,
        NpuID = '123',
        Manufacturer = "test",
        SerialNumber = "123",
        TransceiverType = '1',
        Presence = 1
    }
    c_optical_module.collection  = {
        fetch_by_position = function (...)
            return {
                test = {
                    get_parent = function (...)
                        return opt_info
                    end
                }
            }
        end       
    }
    local net = {
        get_position = function (...)
            return 'test'
        end
    }
    local ret = c_network_adapter.collect_optical_static_info(net)
    lu.assertEquals(#ret, 144)
    opt_info.NpuID = nil
    ret = c_network_adapter.collect_optical_static_info(net)
    lu.assertEquals(#ret, 0)
    opt_info.NpuID = '123'
    opt_info.PortID = 1
    ret = c_network_adapter.collect_optical_static_info(net)
    lu.assertEquals(#ret, 93)
    c_optical_module.collection = col_tab
end

-- 十六进制字符转ascii字符
local function hex_to_ascii(mac)
    local hex = string.gsub(mac, ":", "")
    -- 非法mac地址置零
    if string.find(hex, "[^%x]") then
        return "\0\0\0\0\0\0"
    end
    local str = {}
    local byte
    -- 十六进制字符转换
    for i = 1, #hex, 2 do
        byte = string.sub(hex, i, i + 1)
        table.insert(str, string.char(tonumber(byte, 16)))
    end
    return table.concat(str)
end

local function get_port_mac_hex(port)
    local str = "\0\0\0\0\0\0"
    if not port then
        return str
    end

    local mac = port.PermanentMACAddress
    -- 非法mac地址置零
    if mac and #mac == 17 then
        str = hex_to_ascii(mac)
    end
    return str
end

local function get_uuid_pack_str(ports)
    table.sort(ports, function(a, b)
        return a.PortID < b.PortID
    end)
    local mac1 = get_port_mac_hex(ports[1])
    local mac2 = get_port_mac_hex(ports[2])
    local mac3 = get_port_mac_hex(ports[3])
    local mac4 = get_port_mac_hex(ports[4])
    -- 网口个数小于等于2个时，取两个网口的mac共96位，剩余32位用0补齐
    -- 网口个数大于2个，取第一个网口的mac（48位），后三个网口的低24位共120位，
    -- 剩余8位用0补齐
    if #ports <= 2 then
        return string.pack("<c6c6c4", mac1, mac2, string.rep('\0', 4))
    end
    return string.pack("<c6c3c3c3c1", mac1, string.sub(mac2, 4, 6),
                string.sub(mac3, 4, 6), string.sub(mac4, 4, 6), '\0')
end

local function regenerate_uuid_str(pack_string)
    local partten_mac = bs.new([[<<
        custom_a:48/big,
        custom_b:4,
        custom_c:4,
        custom_d:4,
        custom_e:4,
        custom_f:56/big,
        custom_g:8/big>>]])
    local uuid_part = partten_mac:unpack(pack_string)
    local uuid_param = {
        custom_a = uuid_part.custom_a,
        custom_b = uuid_part.custom_c,
        ver = 8,
        custom_c = uuid_part.custom_e,
        custom_d = uuid_part.custom_b,
        custom_e = uuid_part.custom_d,
        var = 8,
        custom_f = uuid_part.custom_f
    }
    local partten_uuid = bs.new([[<<
        custom_a:48/big,
        custom_b:4,
        ver:4,
        custom_c:4,
        custom_d:4,
        custom_e:4,
        var:4,
        custom_f:56/big>>]])
    return partten_uuid:pack(uuid_param)
end

function TEST_network_adapter:test_network_adapter_uuid()
    local network_port_objs = {
        {
            PermanentMACAddress = "00:15:5d:34:80:59",
            PortID = 0
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5A",
            PortID = 1
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5B",
            PortID = 2
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5C",
            PortID = 3
        }
    }

    local uuid8_1 = regenerate_uuid_str(get_uuid_pack_str({network_port_objs[1]}))
    local uuid8_2 = regenerate_uuid_str(get_uuid_pack_str({network_port_objs[1], network_port_objs[2]}))
    local uuid8_3 = regenerate_uuid_str(
        get_uuid_pack_str({network_port_objs[1], network_port_objs[2], network_port_objs[3]}))
    local uuid8_4 = regenerate_uuid_str(get_uuid_pack_str(network_port_objs))
    local uuid8_4_2 = regenerate_uuid_str(get_uuid_pack_str(network_port_objs))
    lu.assertEquals(uuid8_4, uuid8_4_2)
    lu.assertEquals(uuid8_1, '\x00\x15\x5D\x34\x80\x59\x80\x00\x80\x00\x00\x00\x00\x00\x00\x00')
    lu.assertEquals(uuid8_2, '\x00\x15\x5D\x34\x80\x59\x80\x01\x85\x5d\x34\x80\x5a\x00\x00\x00')
    lu.assertEquals(uuid8_3, '\x00\x15\x5D\x34\x80\x59\x83\x48\x80\x5a\x34\x80\x5b\x00\x00\x00')
    lu.assertEquals(uuid8_4, '\x00\x15\x5D\x34\x80\x59\x83\x48\x80\x5a\x34\x80\x5b\x34\x80\x5c')
end

function TEST_network_adapter:test_network_bonding_update_ports()
    local value = {
        Id = 1,
        Ports = 1
    }
    local ok, _ = pcall(function()
        c_network_bonding.update_ports(value)
    end)
    lu.assertEquals(ok, true)
end

function TEST_network_adapter:test_update_network_port_linkstatus()
    local port = {}
    local data = {}
    local oem_huawei = {}

    port.update_ipv6_default_gateway = function()
    end
    port.set_link_status_numeric = function()
    end

    data.LinkStatus = ''
    local ok, _ = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.OSLinkStatus, 'N/A')

    data.LinkStatus = 'LinkUp'
    local ok, _ = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.OSLinkStatus, 'LinkUp')

    data.LinkStatus = 'LinkUp'
    port.NetDevFuncType = 32
    local ok, _ = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.OSLinkStatus, 'LinkUp')

    data.PhysicalPort = {}
    data.PhysicalPort.LinkStatus = 'Online'
    local ok, _ = pcall(function()
        c_handler_fc:update_fc_network_port(port, data)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.OSLinkStatus, 'LinkUp')

    data.PhysicalPort.LinkStatus = 'Offline'
    local ok, _ = pcall(function()
        c_handler_fc:update_fc_network_port(port, data)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.OSLinkStatus, 'LinkDown')
end

function TEST_network_adapter:test_mctp()
    c_network_adapter.ncsi_schedulers = {}
    local origin_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    c_network_adapter.ncsi_config_obj = {
        DisableHardwareArbitration = function()
            return {
                value = function() return true end
            }
        end,
        ChipTemp = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end,
                on_error = { on = function(data) end }
            }
        end,
        LinkAbility = function()
            return {
                value = function() return {} end
            }
        end,
        LinkInfo = function()
            return {
                value = function() return {} end
            }
        end,
        FaultStatCode = function()
            return {
                on_data_change = { on = function(_, cb) cb({health_status=0}) end },
                on_error = { on = function(data) end },
                start = function () end
            }
        end,
        VendorID = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
        GetOSStatus = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
    }
    c_network_adapter:disable_hardware_arbittration()
    cx5.mctp().properties.DisableHardwareArbitration.response('\x00\x00\x00\x00')
    local res = hi1822.mctp().properties.OpticalModuleInfo.response(
        string.pack('<c85','aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' .. '\x01' ..
            'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'))
    lu.assertEquals(res.MediumType, 'Optical')
    bf2.mctp().properties.GetOSStatus.response('\x00\x00\x00\x00\x05')
    c_network_adapter:update_chip_temp_by_ncsi()
    c_network_adapter:update_link_ability_by_ncsi()
    c_network_adapter:update_link_status_by_ncsi()
    c_network_adapter:update_error_code_by_ncsi()
    c_network_adapter:update_ncsi_firmware({})
    c_network_adapter:fetch_dpu_os_status_by_ncsi()
    c_network_adapter.ncsi_schedulers = origin_ncsi_schedulers
end

function TEST_network_adapter:test_get_max_npu_cdr_temp_from_imu()
    c_network_adapter:get_max_npu_cdr_temp_from_imu()
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu()
    local network_port_objs = {
        { PortID = 0 }, { PortID = 1 }, { PortID = 2 }, { PortID = 3 }
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 40
        end
    end
    c_network_adapter.TemperatureStatus = 0
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 40)

    network_port_objs = { { npu_port_cdr_info_abnormal = true } }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 40
        end
    end
    c_network_adapter.TemperatureStatus = 0
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 3)
end

c_network_adapter.cdr_overtemp_table = {}
function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu_0()
    local network_port_objs = {
        { PortID = 0 , NpuID = 0}
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 90, ''
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.cdr_overtemp_table.max_temp, 0)
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu_1()
    local network_port_objs = {
        { PortID = 0 , NpuID = 0}
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 101, ''
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.cdr_overtemp_table.max_temp, 101)
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu_2()
    local network_port_objs = {
        { PortID = 0 , NpuID = 0}
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 110, ''
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.cdr_overtemp_table.max_temp, 110)
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu_3()
    local network_port_objs = {
        { PortID = 0 , NpuID = 0}
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 105, ''
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.cdr_overtemp_table.max_temp, 105)
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu_4()
    local network_port_objs = {
        { PortID = 0 , NpuID = 0}
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 90, ''
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.cdr_overtemp_table.max_temp, 0)
end

function TEST_network_adapter:test_register_npu_listen_callback()
    local client = require 'network_adapter.client'
    local func = client.OnFruCtrlPropertiesChanged
    local original_values = client.values
    client.values = {
        SysResetDetected = {
            value = function()
                return 1
            end
        },
        PowerState = {
            value = function()
                return "OFF"
            end
        }
    }
    client.OnFruCtrlPropertiesChanged = function(self, cb)
        cb(self.values, 1, 2)
    end
    local collection = c_network_port.collection
    c_network_port.collection = {
        obj = {
            get_parent = function()
                return {
                    Model = "NPU"
                }
            end
        }
    }
    c_network_port.collection.fold = function(self, cb)
        cb(1, self.obj)
    end
    c_network_adapter.register_npu_listen_callback()
    client.OnFruCtrlPropertiesChanged = func
    client.values = original_values
    c_network_port.collection = collection
end

function TEST_network_adapter:test_firmware_version()
    local properties = hi1822.mctp().properties
    local vendor = properties.VendorID
    local response =vendor.response
    local tmp = bs.new
    bs.new = function()
        return {unpack = function(data, flag)
            return {
                ncsi_version_major=8,
                ncsi_version_minor=8, 
                ncsi_version_update=8, 
                ncsi_version_alpha1=8, 
                ncsi_version_alpha2=8,
                firmware_version_0=8,
                firmware_version_1=8,
                firmware_version_2=8,
                firmware_version_3=8,
                pci_vid=16,
                pci_did=16,
                pci_svid=16,
                pci_ssid=16
            }
        end, pack = function()
            return "test"
        end}
    end
    local ok, _ = pcall(function()
        return response(1)
    end)
    lu.assertEquals(ok, true)
    bs.new = tmp
end

function TEST_network_adapter:test_update_ncsi_properties()
    local original_sleep_ms = c_network_adapter.sleep_ms
    local original_disable_hardware_arbittration = c_network_adapter.disable_hardware_arbittration
    local original_package_id = c_network_adapter.package_id
    local original_Type = c_network_adapter.Type
    local original_Model = c_network_adapter.Model
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_update_network_adapter = c_network_adapter.update_network_adapter
    
    local ports = {
        ['BF3'] = {
            set_ncsi_config_obj = function ()
                return nil
            end,
            set_package_id = function(port, package_id)
                port.package_id = package_id
            end,
            set_smbus_schedulers = function ()
                return nil
            end,
            initialize_ncsi_channel = function ()
                return nil
            end,
            update_ncsi_properties = function ()
                return nil
            end,
            next_tick = function ()
                return nil
            end
        }
    }
    
    c_network_adapter.sleep_ms = function() end
    c_network_adapter.disable_hardware_arbittration = function() end
    c_network_adapter.update_network_adapter = function ()
        return nil
    end
    c_network_adapter.package_id = 0
    c_network_adapter.Type = 'test'
    c_network_adapter.Model = 'BF3'
    c_network_adapter.ncsi_config_obj = {}
    c_network_adapter.smbus_schedulers = {}
    
    pcall(c_network_adapter.update_ncsi_properties, c_network_adapter, ports)
    lu.assertEquals(ports['BF3'].package_id, 0)
    
    -- 还原
    if original_sleep_ms ~= nil then
        c_network_adapter.sleep_ms = original_sleep_ms
    end
    if original_disable_hardware_arbittration ~= nil then
        c_network_adapter.disable_hardware_arbittration = original_disable_hardware_arbittration
    end
    if original_package_id ~= nil then
        c_network_adapter.package_id = original_package_id
    end
    if original_Type ~= nil then
        c_network_adapter.Type = original_Type
    end
    if original_Model ~= nil then
        c_network_adapter.Model = original_Model
    end
    if original_ncsi_config_obj ~= nil then
        c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    end
    if original_smbus_schedulers ~= nil then
        c_network_adapter.smbus_schedulers = original_smbus_schedulers
    end
    if original_update_network_adapter ~= nil then
        c_network_adapter.update_network_adapter = original_update_network_adapter
    end
end

function TEST_network_adapter:test_set_npu_max_sfp_temp()
    local c_next_tick = c_network_adapter.next_tick
    local c_update_npu_max_sfp_temp = c_network_adapter.update_npu_max_sfp_temp
    c_network_adapter.update_npu_max_sfp_temp = function ()
        error('end')
    end
    c_network_adapter.next_tick = function(_, cb)
        return cb()
    end
    local ok, _ = pcall(function()
        return c_network_adapter:set_npu_max_sfp_temp()
    end)
    c_network_adapter.next_tick = c_next_tick
    c_network_adapter.update_npu_max_sfp_temp = c_update_npu_max_sfp_temp
    lu.assertEquals(ok, false)
end

function TEST_network_adapter:test_update_npu_max_sfp_temp()
    local ops = {
        { ReadingCelsius = 20 }, { ReadingCelsius = 40 }, { ReadingCelsius = 10 }, { ReadingCelsius = 0 }
    }
    c_network_adapter:update_npu_max_sfp_temp(ops)
    lu.assertEquals(c_network_adapter.SFPMaxTemperatureCelsius, 40)

    ops = { 
        { ReadingCelsius = 20, npu_om_not_present = false, npu_optical_module_info_abnormal = true }
    }
    c_network_adapter.TemperatureStatus = 0
    c_network_adapter:update_npu_max_sfp_temp(ops)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 12)
end

function TEST_network_adapter:test_update_bdf()
    c_network_adapter.Bus = 1
    c_network_adapter.Device = 2
    c_network_adapter.Function = 3
    c_network_adapter:update_root_bdf()
    lu.assertEquals(c_network_adapter.RootBDF, '0000:01:02.3')
end

local pcie_info = {
    system_id = 1,
    is_local = false,
    cpu_id = 1,
    bus_num = 1,
    device_num = 1,
    function_num = 1,
    did = 0,
    vid = 0,
    sdid = 0,
    svid = 0
}

-- 测试四元组信息解析成功
function TEST_network_adapter:test_parse_quater_info_succ()
    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertEquals(pcie_info.did, 0xd120)
    lu.assertEquals(pcie_info.svid, 0x19e5)
end

-- 测试四元组信息校验失败
function TEST_network_adapter:test_update_quater_info_fail()
    pmu_cmd.get_info_from_pmu = function()
        return {0xff, 0xff, 0xff, 0xff}
    end
    local ret = c_network_adapter:verify_vid_did_svid_sdid(pcie_info)
    lu.assertIsFalse(ret)
end

-- 测试获取四元组信息
function TEST_network_adapter:test_update_quater_info()
    pmu_cmd.get_info_from_pmu = function()
        return {0xff, 0xff, 0xff, 0xff}
    end
    local ret = c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertIsFalse(ret)

    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    ret = c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertEquals(pcie_info.did, 0xd120)
    lu.assertIsTrue(ret)
end

-- 测试四元组信息校验成功
function TEST_network_adapter:test_verify_quater_info()
    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    pcie_info.sdid = 0
    local ret = c_network_adapter:verify_vid_did_svid_sdid(pcie_info)
    lu.assertIsTrue(ret)
    lu.assertNotEquals(pcie_info.sdid, 0xd120)
end

function TEST_network_adapter:test_record_nic_os_status_change()
    -- 状态变化上报日志
    local current_os_status = 5
    local last_os_status = 0
    c_network_adapter:record_nic_os_status_change(current_os_status, last_os_status)
    -- 状态未变化无操作
    local current_os_status = 5
    local last_os_status = 5
    c_network_adapter:record_nic_os_status_change(current_os_status, last_os_status)
end

function TEST_network_adapter:test_set_bios_boot_wait_flag()
    local bios_boot_wait_flag = 1
    c_network_adapter:set_bios_boot_wait_flag(bios_boot_wait_flag)
end

function TEST_network_adapter:test_get_dpu_os_status()
    c_network_adapter:get_dpu_os_status()
end

function TEST_network_adapter:test_fetch_dpu_os_status_by_ncsi_failed()
    c_network_adapter.ncsi_config_obj = nil
    c_network_adapter:fetch_dpu_os_status_by_ncsi()
end

mock_func = {}

local function mock_backup()
    mock_func.get_info_from_pmu = pmu_cmd.get_info_from_pmu
    mock_func.fork = skynet.fork
    mock_func.bdf_to_phy_addr = mctp_lib.bdf_to_phy_addr
    mock_func.get_endpoint_and_transport = mctp_lib.get_endpoint_and_transport
    mock_func.get_instance = c_object_manage.get_instance
end

local function mock_restore()
    pmu_cmd.get_info_from_pmu = mock_func.get_info_from_pmu
    skynet.fork = mock_func.fork
    mctp_lib.bdf_to_phy_addr = mock_func.bdf_to_phy_addr
    mctp_lib.get_endpoint_and_transport = mock_func.get_endpoint_and_transport
    c_object_manage.get_instance = mock_func.get_instance
end

local function mock_prepare_1()
    pmu_cmd.get_info_from_pmu = function()
        return {0x0c, 0xc1, 0xc2, 0x04}
    end
    skynet.fork = function(cb)
        cb()
    end
    mctp_lib.bdf_to_phy_addr = function(dev_bus)
        return dev_bus
    end
    mctp_lib.get_endpoint_and_transport = function()
        return 'endpoint', 'transport'
    end
    c_object_manage.get_instance = function()
        return {}
    end
end

local function mock_prepare_2()
    pmu_cmd.get_info_from_pmu = function()
        return {0x0c, nil, 0xc2, 0x04}
    end
end

local function mock_prepare_3()
    pmu_cmd.get_info_from_pmu = function()
        return 
    end
end

local function mock_prepare_4()
    pmu_cmd.get_info_from_pmu = function()
        return {0x0c, 0xc1, 0xc2, 0x04}
    end
    mctp_lib.get_endpoint_and_transport = function()
        return nil, nil
    end
end

function TEST_network_adapter:test_loop_bus_create_mctp_endpoint()
    mock_backup()
    local original_Model = c_network_adapter.Model
    local original_transports = c_network_adapter.transports
    local original_mctp_endpoints = c_network_adapter.mctp_endpoints
    mock_prepare_1()

    c_network_adapter.Model = 'BF3'
    c_network_adapter.transports = {}
    c_network_adapter.mctp_endpoints = {}
    local msg_type = 3
    local endpoint = c_network_adapter:create_mctp_endpoint(msg_type)
    lu.assertEquals(endpoint, 'endpoint')

    mock_prepare_2()
    local ok_2, endpoint_2 = pcall(function()
        return c_network_adapter:create_mctp_endpoint(msg_type)
    end)
    lu.assertEquals(ok_2, false)
    lu.assertNotEquals(endpoint_2, 'endpoint')

    mock_prepare_3()
    local ok_3, endpoint_3 = pcall(function()
        return c_network_adapter:create_mctp_endpoint(msg_type)
    end)
    lu.assertEquals(ok_3, false)
    lu.assertNotEquals(endpoint_3, 'endpoint')

    mock_prepare_4()
    local ok_4, endpoint_4 = pcall(function()
        return c_network_adapter:create_mctp_endpoint(msg_type)
    end)
    lu.assertEquals(ok_4, false)
    lu.assertNotEquals(endpoint_4, 'endpoint')

    mock_restore()
    if original_Model ~= nil then
        c_network_adapter.Model = original_Model
    end
    if original_transports ~= nil then
        c_network_adapter.transports = original_transports
    end
    if original_mctp_endpoints ~= nil then
        c_network_adapter.mctp_endpoints = original_mctp_endpoints
    end
end

-- 测试通过pldm协议更新信息
function TEST_network_adapter:test_init_pldm()
    c_network_adapter:ctor()
    local original_Model = c_network_adapter.Model
    local original_SupportedMctp = c_network_adapter.SupportedMctp
    local original_DevBus = c_network_adapter.DevBus
    local original_DevDevice = c_network_adapter.DevDevice
    local original_DevFunction = c_network_adapter.DevFunction
    c_network_adapter.Model = 'QLE2742'
    c_network_adapter.SupportedMctp = true
    c_network_adapter.DevBus = 1
    c_network_adapter.DevDevice = 0
    c_network_adapter.DevFunction = 0
    local create_endpoint = c_network_adapter.create_mctp_endpoint
    c_network_adapter.create_mctp_endpoint = function ()
        return {}
    end
    local next_tick = c_network_adapter.next_tick
    c_network_adapter.next_tick = function (_, cb, ...)
        cb(...)
    end
    c_network_adapter:synchronize_port_bdf(2, 2)
    for _, v in pairs(c_network_adapter.pldm_schedulers) do
        pcall(function ()
            v.on_data_change:emit(0x00)
        end)
    end
    for _, v in pairs(c_network_adapter.pldm_schedulers) do
        pcall(function ()
            v.on_data_change:emit()
        end
    )
    end
    utils.msleep(1000)
    lu.assertNotEquals(c_network_adapter.pldm_config_obj, nil)
    c_network_adapter.next_tick = next_tick
    c_network_adapter.create_mctp_endpoint = create_endpoint
    if original_Model ~= nil then
        c_network_adapter.Model = original_Model
    end
    if original_SupportedMctp ~= nil then
        c_network_adapter.SupportedMctp = original_SupportedMctp
    end
    if original_DevBus ~= nil then
        c_network_adapter.DevBus = original_DevBus
    end
    if original_DevDevice ~= nil then
        c_network_adapter.DevDevice = original_DevDevice
    end
    if original_DevFunction ~= nil then
        c_network_adapter.DevFunction = original_DevFunction
    end
end

-- 测试停止pldm协议例测任务
function TEST_network_adapter:test_stop_schedulers()
    c_network_adapter.ncsi_schedulers = {
        schedulers = {
            deconstruct = function()
            end
        }
    }
    c_network_adapter:stop()
    lu.assertEquals(c_network_adapter.pldm_schedulers, {})
    lu.assertEquals(c_network_adapter.smbus_schedulers, {})
    lu.assertEquals(c_network_adapter.ncsi_schedulers, {})
end

function TEST_network_adapter:test_update_port_mac()
    local port = {
        WorkloadType = 1,
        PortID = 1,
        MACAddress = "11:11:11:11:11:11"
    }
    local mac_info = {
        {
            [1] = 1,
            [3] = "22:22:22:22:22:22"
        }
    }
    c_network_adapter.update_port_mac_action(mac_info, port)
end

-- 测试 update_port_wwnn_by_pldm 日志输出
function TEST_network_adapter:test_update_port_wwnn_by_pldm_log()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_pldm_schedulers = c_network_adapter.pldm_schedulers
    local original_pldm_config_obj = c_network_adapter.pldm_config_obj
    local log_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(log_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.pldm_schedulers = {}
    c_network_adapter.pldm_config_obj = {
        DefaultWWNN = function()
            return {
                on_data_change = { on = function() end },
                start = function() end
            }
        end,
        WorkWWNN = function()
            return {
                on_data_change = { on = function() end },
                start = function() end
            }
        end
    }
    
    c_network_adapter:update_port_wwnn_by_pldm({})
    
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[1].fmt, '%s update default_wwnn by pldm start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    lu.assertEquals(log_calls[2].fmt, '%s update work_wwnn by pldm start')
    lu.assertEquals(log_calls[2].args[1], "TestAdapter")
    
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.pldm_schedulers = original_pldm_schedulers
    c_network_adapter.pldm_config_obj = original_pldm_config_obj
end

-- 测试 update_port_wwpn_by_pldm 日志输出
function TEST_network_adapter:test_update_port_wwpn_by_pldm_log()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_pldm_schedulers = c_network_adapter.pldm_schedulers
    local original_pldm_config_obj = c_network_adapter.pldm_config_obj
    local log_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(log_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.pldm_schedulers = {}
    c_network_adapter.pldm_config_obj = {
        Port1DefaultWWPN = function()
            return {
                on_data_change = { on = function() end },
                start = function() end
            }
        end,
        Port1WorkWWPN = function()
            return {
                on_data_change = { on = function() end },
                start = function() end
            }
        end
    }
    
    c_network_adapter:update_port_wwpn_by_pldm(1, {})
    
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[1].fmt, '%s update default_wwpn by pldm start, port_id:%s')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    lu.assertEquals(log_calls[1].args[2], 1)
    lu.assertEquals(log_calls[2].fmt, '%s update work_wwpn by pldm start, port_id:%s')
    lu.assertEquals(log_calls[2].args[1], "TestAdapter")
    lu.assertEquals(log_calls[2].args[2], 1)
    
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.pldm_schedulers = original_pldm_schedulers
    c_network_adapter.pldm_config_obj = original_pldm_config_obj
end

-- 测试 update_port_link_status_by_pldm 函数
function TEST_network_adapter:test_update_port_link_status_by_pldm()
    local log = require('mc.logging')
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_pldm_schedulers = c_network_adapter.pldm_schedulers
    local original_pldm_config_obj = c_network_adapter.pldm_config_obj
    local error_called = false
    local mock_scheduler = {}
    
    log.error = function(self, fmt, ...)
        error_called = true
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.pldm_schedulers = {}
    c_network_adapter.pldm_config_obj = {
        Port1LinkStatus = function() return mock_scheduler end
    }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    local test_port = {
        PortID = 1,
        get_properties_res = {},
        set_link_status = function() end,
        set_link_status_numeric = function() end
    }
    
    c_network_adapter:update_port_link_status_by_pldm(1, {test_port})
    
    -- 测试数据变化
    mock_scheduler.data_cb({LinkStatus = "Up", SpeedMbps = 1000, SpeedGbps = 1})
    lu.assertTrue(test_port.get_properties_res.LinkStatusFromPldm)
    
    -- 测试错误回调
    test_port.get_properties_res.LinkStatusFromPldm = true
    mock_scheduler.error_cb()
    lu.assertFalse(test_port.get_properties_res.LinkStatusFromPldm)
    lu.assertTrue(error_called)
    
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.pldm_schedulers = original_pldm_schedulers
    c_network_adapter.pldm_config_obj = original_pldm_config_obj
end

-- 测试 update_chip_temp_by_pldm 函数
function TEST_network_adapter:test_update_chip_temp_by_pldm()
    local log = require('mc.logging')
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_pldm_schedulers = c_network_adapter.pldm_schedulers
    local original_pldm_config_obj = c_network_adapter.pldm_config_obj
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_smbios_status = c_network_adapter.smbios_status
    local original_TemperatureCelsius = c_network_adapter.TemperatureCelsius
    local original_TemperatureStatus = c_network_adapter.TemperatureStatus
    local error_called = false
    local mock_scheduler = {}
    
    log.error = function(self, fmt, ...)
        error_called = true
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.pldm_schedulers = {}
    c_network_adapter.get_properties_res = {}
    c_network_adapter.pldm_config_obj = {
        ChipTemp = function() return mock_scheduler end
    }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    c_network_adapter:update_chip_temp_by_pldm()
    
    -- 测试数据变化
    mock_scheduler.data_cb(45)
    lu.assertTrue(c_network_adapter.get_properties_res.ChipTempFromPldm)
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 45)
    
    -- 测试错误回调 - smbios_status == 3 时 TemperatureStatus = 1
    c_network_adapter.get_properties_res.ChipTempFromPldm = true
    c_network_adapter.smbios_status = 3
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertFalse(c_network_adapter.get_properties_res.ChipTempFromPldm)
    lu.assertTrue(error_called)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 0, "TemperatureStatus should be 0 when smbios_status == 3 and debounce not meet")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    -- 测试错误回调 - smbios_status != 3 时 TemperatureStatus = 2
    c_network_adapter.get_properties_res.ChipTempFromPldm = true
    c_network_adapter.smbios_status = 2
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertEquals(c_network_adapter.TemperatureStatus, 2, "TemperatureStatus should be 2 when smbios_status != 3")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.pldm_schedulers = original_pldm_schedulers
    c_network_adapter.pldm_config_obj = original_pldm_config_obj
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.TemperatureCelsius = original_TemperatureCelsius
    c_network_adapter.TemperatureStatus = original_TemperatureStatus
end

-- 测试 update_link_status_by_pldm 函数
function TEST_network_adapter:test_update_link_status_by_pldm()
    local log = require('mc.logging')
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_pldm_schedulers = c_network_adapter.pldm_schedulers
    local original_pldm_config_obj = c_network_adapter.pldm_config_obj
    local error_called = false
    local mock_scheduler = {}
    
    log.error = function(self, fmt, ...)
        error_called = true
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.pldm_schedulers = {}
    c_network_adapter.pldm_config_obj = {
        Port1LinkStatus = function() return mock_scheduler end
    }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    local test_port = {
        PortID = 1,
        NodeId = "TestPort1",
        get_properties_res = {}
    }
    
    c_network_adapter:update_link_status_by_pldm(1, {test_port})
    
    -- 测试数据变化
    mock_scheduler.data_cb("Up")
    lu.assertTrue(test_port.get_properties_res.LinkStatusFromPldm)
    lu.assertEquals(test_port.LinkStatus, "Up")
    
    -- 测试错误回调
    test_port.get_properties_res.LinkStatusFromPldm = true
    mock_scheduler.error_cb()
    lu.assertFalse(test_port.get_properties_res.LinkStatusFromPldm)
    lu.assertTrue(error_called)
    
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.pldm_schedulers = original_pldm_schedulers
    c_network_adapter.pldm_config_obj = original_pldm_config_obj
end

-- 测试 update_chip_temp 函数 (smbus版本)
function TEST_network_adapter:test_update_chip_temp_smbus()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_smbios_status = c_network_adapter.smbios_status
    local original_TemperatureCelsius = c_network_adapter.TemperatureCelsius
    local original_TemperatureStatus = c_network_adapter.TemperatureStatus
    local log_calls = {}
    local mock_scheduler = {}
    
    log.notice = function(self, fmt, ...) table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) end
    log.error = function(self, fmt, ...) table.insert(log_calls, {type='error', fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.smbus_schedulers = {}
    c_network_adapter.get_properties_res = {}
    c_network_adapter.smbus_config_obj = { ChipTemp = function() return mock_scheduler end }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    c_network_adapter:update_chip_temp()
    
    -- 测试启动日志
    lu.assertEquals(log_calls[1].fmt, '%s update chip temp by smbus start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    
    -- 测试数据变化（没有日志输出）
    mock_scheduler.data_cb(55)
    lu.assertTrue(c_network_adapter.get_properties_res.ChipTempFromSmbus)
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 55)
    
    -- 测试错误回调 - smbios_status == 3 时 TemperatureStatus = 1
    c_network_adapter.get_properties_res.ChipTempFromSmbus = true
    c_network_adapter.smbios_status = 3
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertFalse(c_network_adapter.get_properties_res.ChipTempFromSmbus)
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[2].fmt, '%s update chip temp by smbus on_error')
    lu.assertEquals(c_network_adapter.TemperatureStatus, 0, "TemperatureStatus should be 0 when smbios_status == 3 and debounce not meet")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    -- 测试错误回调 - smbios_status != 3 时 TemperatureStatus = 2
    c_network_adapter.get_properties_res.ChipTempFromSmbus = true
    c_network_adapter.smbios_status = 2
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertEquals(c_network_adapter.TemperatureStatus, 2, "TemperatureStatus should be 2 when smbios_status != 3")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    log.notice = original_notice
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.TemperatureCelsius = original_TemperatureCelsius
    c_network_adapter.TemperatureStatus = original_TemperatureStatus
end

-- 测试 update_optical_temp 函数 (smbus版本)
function TEST_network_adapter:test_update_optical_temp_smbus()
    local log = require('mc.logging')
    local original_debug = log.debug
    local original_error = log.error
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local log_calls = {}
    local mock_scheduler = {}
    
    log.debug = function(self, fmt, ...) table.insert(log_calls, {type='debug', fmt=fmt, args={...}}) end
    log.error = function(self, fmt, ...) table.insert(log_calls, {type='error', fmt=fmt, args={...}}) end
    log.notice = function(self, fmt, ...) table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.smbus_schedulers = {}
    c_network_adapter.smbus_config_obj = { OpticalTemp = function() return mock_scheduler end }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    local test_ports = {{
        NodeId = "Port1", 
        get_properties_res = {}, 
        set_op_temp_hi182x = function() end, 
        connect_signal = function() end
    }}
    c_network_adapter:update_optical_temp(test_ports)
    
    -- 测试启动日志
    lu.assertEquals(log_calls[1].fmt, '%s update optical temp by smbus start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    
    -- 测试错误回调
    test_ports[1].get_properties_res.OpticalTempFromSmbus = true
    mock_scheduler.error_cb()
    lu.assertFalse(test_ports[1].get_properties_res.OpticalTempFromSmbus)
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[2].fmt, '%s update optical temp by smbus on_error')
    
    log.debug = original_debug
    log.error = original_error
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
end

-- 测试 update_smbus_link_status 函数
function TEST_network_adapter:test_update_smbus_link_status()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local log_calls = {}
    local mock_scheduler = {}
    
    log.notice = function(self, fmt, ...) table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) end
    log.error = function(self, fmt, ...) table.insert(log_calls, {type='error', fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.smbus_schedulers = {}
    c_network_adapter.smbus_config_obj = { LinkStatus = function() return mock_scheduler end }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    local test_ports = {{
        NodeId = "Port1", 
        get_properties_res = {}, 
        set_link_status = function() end, 
        set_link_status_numeric = function() end, 
        connect_signal = function() end
    }}
    c_network_adapter:update_smbus_link_status(test_ports)
    
    -- 测试启动日志
    lu.assertEquals(log_calls[1].fmt, '%s update link status by smbus start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    
    -- 测试错误回调
    test_ports[1].get_properties_res.LinkStatusFromSmbus = true
    mock_scheduler.error_cb()
    lu.assertFalse(test_ports[1].get_properties_res.LinkStatusFromSmbus)
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[2].fmt, '%s update link status by smbus on_error')
    
    log.notice = original_notice
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
end

-- 测试 update_smbus_mac_address 函数
function TEST_network_adapter:test_update_smbus_mac_address()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local log_calls = {}
    local mock_scheduler = {}
    local mock_signal = {}
    local signal_handlers = {}
    
    log.notice = function(self, fmt, ...) table.insert(log_calls, {fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.smbus_schedulers = {}
    c_network_adapter.smbus_config_obj = { MacAddress = function() return mock_scheduler end }
    
    -- 模拟信号系统
    mock_signal.emit = function(self, port_id, val)
        for _, handler in ipairs(signal_handlers) do
            handler(port_id, val)
        end
    end
    
    mock_scheduler.on_data_change = { 
        on = function(self, cb) 
            mock_scheduler.data_cb = function(data)
                for i, val in ipairs(data) do
                    mock_signal:emit(i, val)
                end
            end
        end 
    }
    mock_scheduler.start = function() end
    
    -- 测试场景1: MACAddress 为 '00:00:00:00:00:00' 时应该调用 set_mac_addr
    local set_mac_called_1 = false
    local set_permanent_mac_called_1 = false
    local set_mac_addr_1 = function(self, addr)
        set_mac_called_1 = true
        lu.assertEquals(addr, "AA:BB:CC:DD:EE:FF")
    end
    local set_permanent_mac_addr_1 = function(self, addr)
        set_permanent_mac_called_1 = true
        lu.assertEquals(addr, "AA:BB:CC:DD:EE:FF")
    end
    
    local test_ports_1 = {{
        NodeId = "Port1",
        MACAddress = '00:00:00:00:00:00',
        set_mac_addr = set_mac_addr_1,
        set_permanent_mac_addr = set_permanent_mac_addr_1,
        connect_signal = function(self, sig, handler)
            table.insert(signal_handlers, handler)
        end
    }}
    
    signal_handlers = {}
    c_network_adapter:update_smbus_mac_address(test_ports_1)
    
    -- 触发回调测试条件判断
    if mock_scheduler.data_cb then
        mock_scheduler.data_cb({"AA:BB:CC:DD:EE:FF"})
    end
    
    lu.assertTrue(set_mac_called_1, "当 MACAddress 为 '00:00:00:00:00:00' 时应该调用 set_mac_addr")
    lu.assertTrue(set_permanent_mac_called_1, "应该调用 set_permanent_mac_addr")
    
    -- 测试场景2: MACAddress 为 'N/A' 时应该调用 set_mac_addr
    local set_mac_called_2 = false
    local set_permanent_mac_called_2 = false
    local set_mac_addr_2 = function(self, addr)
        set_mac_called_2 = true
        lu.assertEquals(addr, "11:22:33:44:55:66")
    end
    local set_permanent_mac_addr_2 = function(self, addr)
        set_permanent_mac_called_2 = true
        lu.assertEquals(addr, "11:22:33:44:55:66")
    end
    
    local test_ports_2 = {{
        NodeId = "Port2",
        MACAddress = 'N/A',
        set_mac_addr = set_mac_addr_2,
        set_permanent_mac_addr = set_permanent_mac_addr_2,
        connect_signal = function(self, sig, handler)
            table.insert(signal_handlers, handler)
        end
    }}
    
    signal_handlers = {}
    c_network_adapter:update_smbus_mac_address(test_ports_2)
    
    if mock_scheduler.data_cb then
        mock_scheduler.data_cb({"11:22:33:44:55:66"})
    end
    
    lu.assertTrue(set_mac_called_2, "当 MACAddress 为 'N/A' 时应该调用 set_mac_addr")
    lu.assertTrue(set_permanent_mac_called_2, "应该调用 set_permanent_mac_addr")
    
    -- 测试场景3: MACAddress 为其他值时不应该调用 set_mac_addr，但应该调用 set_permanent_mac_addr
    local set_mac_called_3 = false
    local set_permanent_mac_called_3 = false
    local set_mac_addr_3 = function(self, addr)
        set_mac_called_3 = true
    end
    local set_permanent_mac_addr_3 = function(self, addr)
        set_permanent_mac_called_3 = true
        lu.assertEquals(addr, "99:88:77:66:55:44")
    end
    
    local test_ports_3 = {{
        NodeId = "Port3",
        MACAddress = '12:34:56:78:90:AB',
        set_mac_addr = set_mac_addr_3,
        set_permanent_mac_addr = set_permanent_mac_addr_3,
        connect_signal = function(self, sig, handler)
            table.insert(signal_handlers, handler)
        end
    }}
    
    signal_handlers = {}
    c_network_adapter:update_smbus_mac_address(test_ports_3)
    
    if mock_scheduler.data_cb then
        mock_scheduler.data_cb({"99:88:77:66:55:44"})
    end
    
    lu.assertFalse(set_mac_called_3, "当 MACAddress 为其他值时不应该调用 set_mac_addr")
    lu.assertTrue(set_permanent_mac_called_3, "应该调用 set_permanent_mac_addr")
    
    -- 测试启动日志
    lu.assertEquals(log_calls[1].fmt, '%s update mac address by smbus start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
end

-- 测试 update_ncsi_firmware 函数
function TEST_network_adapter:test_update_ncsi_firmware()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    local original_VendorID = c_network_adapter.VendorID
    local log_calls = {}
    local mock_scheduler = {}
    
    log.notice = function(self, fmt, ...) table.insert(log_calls, {fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.package_id = 1
    c_network_adapter.VendorID = ""
    c_network_adapter.ncsi_config_obj = { 
        VendorID = function(params) return mock_scheduler end 
    }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.start = function() end
    
    local test_ports = {{set_firmware_version = function() end}}
    c_network_adapter:update_ncsi_firmware(test_ports)
    
    -- 测试启动日志
    lu.assertEquals(log_calls[1].fmt, '%s update firmware version by ncsi start')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    
    -- 测试数据变化
    mock_scheduler.data_cb({FirmwareVersion = "v1.2.3"})
    lu.assertEquals(#log_calls, 2)
    lu.assertEquals(log_calls[2].fmt, '%s update firmware version to %s by ncsi')
    lu.assertEquals(log_calls[2].args[1], "TestAdapter")
    lu.assertEquals(log_calls[2].args[2], "v1.2.3")
    lu.assertEquals(c_network_adapter.FirmwareVersion, "v1.2.3")
    
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
    c_network_adapter.VendorID = original_VendorID
end

-- 测试 update_chip_temp_by_ncsi 函数
function TEST_network_adapter:test_update_chip_temp_by_ncsi()
    local log = require('mc.logging')
    local original_error = log.error
    local original_NodeId = c_network_adapter.NodeId
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_smbios_status = c_network_adapter.smbios_status
    local original_TemperatureCelsius = c_network_adapter.TemperatureCelsius
    local original_TemperatureStatus = c_network_adapter.TemperatureStatus
    local error_called = false
    local mock_scheduler = {}
    
    log.error = function(self, fmt, ...)
        error_called = true
    end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.get_properties_res = {}
    c_network_adapter.ncsi_config_obj = { ChipTemp = function() return mock_scheduler end }
    
    mock_scheduler.on_data_change = { on = function(self, cb) mock_scheduler.data_cb = cb end }
    mock_scheduler.on_error = { on = function(self, cb) mock_scheduler.error_cb = cb end }
    mock_scheduler.start = function() end
    
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 测试数据变化
    mock_scheduler.data_cb(50)
    lu.assertTrue(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    
    -- 测试错误回调 - smbios_status == 3 时 TemperatureStatus = 1
    c_network_adapter.get_properties_res.ChipTempFromNcsi = true
    c_network_adapter.smbios_status = 3
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertFalse(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    lu.assertTrue(error_called)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 0, "TemperatureStatus should be 1 when smbios_status == 3 and debounce not meet")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    -- 测试错误回调 - smbios_status != 3 时 TemperatureStatus = 2
    c_network_adapter.get_properties_res.ChipTempFromNcsi = true
    c_network_adapter.smbios_status = 2
    c_network_adapter.TemperatureCelsius = 50
    c_network_adapter.TemperatureStatus = 0
    mock_scheduler.error_cb()
    lu.assertEquals(c_network_adapter.TemperatureStatus, 2, "TemperatureStatus should be 2 when smbios_status != 3")
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 0x4000, "TemperatureCelsius should be 0x4000 on error")
    
    log.error = original_error
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.TemperatureCelsius = original_TemperatureCelsius
    c_network_adapter.TemperatureStatus = original_TemperatureStatus
end

-- 测试 update_max_operating_temp_by_ncsi 函数
function TEST_network_adapter:test_update_max_operating_temp_by_ncsi()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_NodeId = c_network_adapter.NodeId
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    local log_calls = {}
    
    log.notice = function(self, fmt, ...) table.insert(log_calls, {fmt=fmt, args={...}}) end
    
    c_network_adapter.NodeId = "TestAdapter"
    c_network_adapter.package_id = 1
    c_network_adapter.ncsi_config_obj = { 
        MaxOpTemp = function(params) 
            return { value = function() return 90 end }
        end 
    }
    
    c_network_adapter:update_max_operating_temp_by_ncsi()
    
    -- 测试日志输出
    lu.assertEquals(log_calls[1].fmt, '%s update max operating temperature to %s by ncsi')
    lu.assertEquals(log_calls[1].args[1], "TestAdapter")
    lu.assertEquals(log_calls[1].args[2], 90)
    lu.assertEquals(c_network_adapter.MaxOperatingTemperatureCelsius, 90)
    
    log.notice = original_notice
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试 update_health 函数
function TEST_network_adapter:test_update_health()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_info = log.info
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local original_FaultState = c_network_adapter.FaultState
    local original_FaultCode = c_network_adapter.FaultCode
    local original_smbios_status = c_network_adapter.smbios_status
    local original_smbus_collect_status = c_network_adapter.smbus_collect_status
    local original_smbus_has_collected = c_network_adapter.smbus_has_collected
    local original_last_fault_code = c_network_adapter.last_fault_code
    local original_collect_log_by_smbus_task = c_network_adapter.collect_log_by_smbus_task
    local original_package_id = c_network_adapter.package_id
    local log_calls = {}
    local collect_log_called = false
    local mock_scheduler = {}
    local mock_fault_code_scheduler = {}
    local fault_code_value = {1, 2, 0}
    
    -- 使用pcall确保即使出错也能还原
    local ok, err = pcall(function()
        log.notice = function(self, fmt, ...) 
            table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) 
        end
        log.info = function(self, fmt, ...) 
            table.insert(log_calls, {type='info', fmt=fmt, args={...}}) 
        end
        
        c_network_adapter.NodeId = "TestAdapter"
        c_network_adapter.smbus_schedulers = {}
        c_network_adapter.smbios_status = 2
        c_network_adapter.smbus_collect_status = 0  -- LOG_DUMP_IDLE
        c_network_adapter.smbus_has_collected = false
        c_network_adapter.last_fault_code = {}
        c_network_adapter.FaultState = 0
        c_network_adapter.FaultCode = 'N/A'
        
        -- 模拟collect_log_by_smbus_task
        c_network_adapter.collect_log_by_smbus_task = function()
            collect_log_called = true
        end
        
        -- 模拟FaultCode调度器
        mock_fault_code_scheduler.value = function()
            return fault_code_value
        end
        
        -- 模拟SMBUS配置对象
        c_network_adapter.smbus_config_obj = {
            Health = function() 
                return mock_scheduler 
            end,
            FaultCode = function() 
                return mock_fault_code_scheduler 
            end
        }
        
        mock_scheduler.on_data_change = { 
            on = function(self, cb) 
                mock_scheduler.data_cb = cb 
            end 
        }
        mock_scheduler.on_error = { 
            on = function(self, cb) 
                mock_scheduler.error_cb = cb 
            end 
        }
        mock_scheduler.start = function() end
        
        -- 执行函数
        c_network_adapter:update_health()
        
        -- 测试1: data == 0时，重置FaultState和FaultCode
        mock_scheduler.data_cb(0)
        lu.assertEquals(c_network_adapter.FaultState, 0)
        
        -- 测试2: data != 0时，更新FaultState和FaultCode
        fault_code_value = {1, 2, 0}
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultState, 1)
        lu.assertEquals(c_network_adapter.FaultCode, '1, 2')
        lu.assertTrue(collect_log_called)
        collect_log_called = false
        
        -- 测试3: 故障代码未变化时，不更新FaultCode
        c_network_adapter.last_fault_code = {1, 2, 0}
        local old_fault_code = c_network_adapter.FaultCode
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, old_fault_code)
        
        -- 测试4: 故障代码变化时，更新FaultCode
        fault_code_value = {3, 4, 0}
        c_network_adapter.last_fault_code = {1, 2, 0}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, '3, 4')
        
        -- 测试5: 空故障代码时，设置为'N/A'
        fault_code_value = {}
        c_network_adapter.last_fault_code = {3, 4, 0}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
        
        -- 测试6: smbus_collect_status == LOG_DUMP_BUSY时，不收集日志
        c_network_adapter.smbus_collect_status = 1  -- LOG_DUMP_BUSY
        c_network_adapter.smbus_has_collected = false  -- 重置状态
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.smbus_collect_status, 1)
        lu.assertFalse(c_network_adapter.smbus_has_collected)
        
        -- 测试7: 错误回调 - smbios_status != 3时，重置FaultState和FaultCode
        c_network_adapter.smbios_status = 2
        c_network_adapter.FaultState = 1
        c_network_adapter.FaultCode = '1, 2'
        mock_scheduler.error_cb()
        lu.assertEquals(c_network_adapter.FaultState, 0)
        
        -- 测试8: 错误回调 - smbios_status == 3时，不重置
        c_network_adapter.smbios_status = 3
        c_network_adapter.FaultState = 1
        c_network_adapter.FaultCode = '1, 2'
        mock_scheduler.error_cb()
        lu.assertEquals(c_network_adapter.FaultState, 1)
        lu.assertEquals(c_network_adapter.FaultCode, '1, 2')
        
        -- 测试9: 真实调用 FaultCode response 函数，使用给定的测试数据 '\x00\x01\x00\x04\x2B\x00\x00\x04'
        local libmgmt_protocol = require 'libmgmt_protocol'
        local original_create_array_parser = libmgmt_protocol.create_array_parser
        local fault_code_response = hi1822.smbus().properties.FaultCode.response
        local test_data = '\x00\x01\x00\x04\x2B\x00\x00\x04'
        libmgmt_protocol.create_array_parser = function(template, length)
            lu.assertEquals(template, [[<<code:16>>]], "template should match")
            lu.assertEquals(length, 2, "length should be 2")       
            return function(data)
                lu.assertEquals(data, test_data, "data should match test_data")
                return {
                    {code = 0x0100},
                    {code = 0x0400},
                    {code = 0x002B},
                    {code = 0x0400}
                }
            end
        end
        -- 在 update_health 中真实调用 FaultCode: 让 value() 方法调用真实的 response 函数
        local fault_code_result_from_response
        mock_fault_code_scheduler.value = function()
            -- 模拟从硬件读取数据，然后调用真实的 response 函数处理
            fault_code_result_from_response = fault_code_response(test_data)
            return fault_code_result_from_response
        end
        local expected_fault_codes = {256, 1024, 43, 1024}
        c_network_adapter.last_fault_code = {}
        c_network_adapter.FaultCode = 'N/A'  -- 初始值
        c_network_adapter.smbus_collect_status = 0  -- LOG_DUMP_IDLE
        mock_scheduler.data_cb(1)  -- 触发 update_health 中的 FaultCode 调用
        
        -- 验证1: response 函数返回的结果
        lu.assertNotNil(fault_code_result_from_response, "fault_code_result_from_response should not be nil")
        lu.assertEquals(type(fault_code_result_from_response), "table", "fault_code_result_from_response should be a table")
        lu.assertEquals(#fault_code_result_from_response, 4, "fault_code_result_from_response should have 4 elements")
        lu.assertEquals(fault_code_result_from_response[1], 256, "first fault code from response should be 256")
        lu.assertEquals(fault_code_result_from_response[2], 1024, "second fault code from response should be 1024")
        lu.assertEquals(fault_code_result_from_response[3], 43, "third fault code from response should be 43")
        lu.assertEquals(fault_code_result_from_response[4], 1024, "fourth fault code from response should be 1024")
        
        -- 验证2: last_fault_code 被正确赋值
        lu.assertNotNil(c_network_adapter.last_fault_code, "last_fault_code should not be nil")
        lu.assertEquals(#c_network_adapter.last_fault_code, 4, "last_fault_code should have 4 elements")
        lu.assertEquals(c_network_adapter.last_fault_code[1], 256, "last_fault_code[1] should be 256")
        lu.assertEquals(c_network_adapter.last_fault_code[2], 1024, "last_fault_code[2] should be 1024")
        lu.assertEquals(c_network_adapter.last_fault_code[3], 43, "last_fault_code[3] should be 43")
        lu.assertEquals(c_network_adapter.last_fault_code[4], 1024, "last_fault_code[4] should be 1024")
        
        -- 验证3: join_fault_codes 处理后的字符串（手动计算期望值）
        local expected_fault_code_str = "256, 1024, 43, 1024"
        
        -- 验证4: 最终赋值给 FaultCode 的值
        lu.assertEquals(c_network_adapter.FaultState, 1, "FaultState should be 1")
        lu.assertEquals(c_network_adapter.FaultCode, expected_fault_code_str, 
            string.format("FaultCode should be '%s', but got '%s'", expected_fault_code_str, c_network_adapter.FaultCode))
        -- 还原 create_array_parser
        libmgmt_protocol.create_array_parser = original_create_array_parser
        
        -- 测试10: 验证日志收集状态变化
        c_network_adapter.smbus_collect_status = 0
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.smbus_collect_status, 0)
        lu.assertTrue(c_network_adapter.smbus_has_collected)
    end)
    
    -- 还原（无论成功还是失败都要还原）
    log.notice = original_notice
    log.info = original_info
    c_network_adapter.NodeId = original_NodeId
    -- 先清除测试过程中可能添加的 health_task
    if c_network_adapter.smbus_schedulers then
        c_network_adapter.smbus_schedulers.health_task = nil
    end
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
    c_network_adapter.FaultState = original_FaultState
    c_network_adapter.FaultCode = original_FaultCode
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.smbus_collect_status = original_smbus_collect_status
    c_network_adapter.smbus_has_collected = original_smbus_has_collected
    c_network_adapter.last_fault_code = original_last_fault_code
    c_network_adapter.collect_log_by_smbus_task = original_collect_log_by_smbus_task
    c_network_adapter.package_id = original_package_id
    
    -- 如果测试失败，重新抛出错误
    if not ok then
        error(err)
    end
end

-- 测试 update_error_code_by_ncsi 函数
function TEST_network_adapter:test_update_error_code_by_ncsi()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_error = log.error
    local original_info = log.info
    local original_NodeId = c_network_adapter.NodeId
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_smbios_status = c_network_adapter.smbios_status
    local original_FaultState = c_network_adapter.FaultState
    local original_FaultCode = c_network_adapter.FaultCode
    local original_smbus_collect_status = c_network_adapter.smbus_collect_status
    local original_smbus_has_collected = c_network_adapter.smbus_has_collected
    local original_collect_log_by_smbus_task = c_network_adapter.collect_log_by_smbus_task
    local log_calls = {}
    local collect_log_called = false
    local mock_scheduler = {}
    
    -- 使用pcall确保即使出错也能还原
    local ok, err = pcall(function()
        log.notice = function(self, fmt, ...) 
            table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) 
        end
        log.error = function(self, fmt, ...) 
            table.insert(log_calls, {type='error', fmt=fmt, args={...}}) 
        end
        log.info = function(self, fmt, ...) 
            table.insert(log_calls, {type='info', fmt=fmt, args={...}}) 
        end
        
        c_network_adapter.NodeId = "TestAdapter"
        c_network_adapter.ncsi_schedulers = {}
        c_network_adapter.smbus_schedulers = {}
        c_network_adapter.get_properties_res = {}
        c_network_adapter.smbios_status = 2
        c_network_adapter.smbus_collect_status = 0  -- LOG_DUMP_IDLE
        c_network_adapter.smbus_has_collected = false
        c_network_adapter.FaultState = 0
        c_network_adapter.FaultCode = 'N/A'
        c_network_adapter.package_id = 1
        
        -- 模拟collect_log_by_smbus_task
        c_network_adapter.collect_log_by_smbus_task = function()
            collect_log_called = true
        end
        
        -- 模拟SMBUS health任务
        local smbus_health_task_paused = false
        local smbus_health_task_resumed = false
        c_network_adapter.smbus_schedulers.health_task = {
            is_paused = false,
            pause = function() 
                smbus_health_task_paused = true
                c_network_adapter.smbus_schedulers.health_task.is_paused = true
            end,
            resume = function() 
                smbus_health_task_resumed = true
                c_network_adapter.smbus_schedulers.health_task.is_paused = false
            end,
            data = "some_data"
        }
        
        -- 模拟NCSI配置对象
        c_network_adapter.ncsi_config_obj = {
            FaultStatCode = function(self, params) 
                lu.assertNotNil(params, "params should not be nil")
                lu.assertEquals(params.package_id, 1)
                return mock_scheduler 
            end
        }
        
        mock_scheduler.on_data_change = { 
            on = function(self, cb) 
                mock_scheduler.data_cb = cb 
            end 
        }
        mock_scheduler.on_error = { 
            on = function(self, cb) 
                mock_scheduler.error_cb = cb 
            end 
        }
        mock_scheduler.start = function() end
        
        -- 执行函数
        c_network_adapter:update_error_code_by_ncsi()
        
        -- 测试1: NCSI成功时，更新FaultState和FaultCode，暂停SMBUS备选
        local test_data = {
            health_status = 1,
            error_codes = {1, 2, 0}
        }
        mock_scheduler.data_cb(test_data)
        lu.assertEquals(c_network_adapter.FaultState, 1)
        lu.assertEquals(c_network_adapter.FaultCode, '1, 2')
        lu.assertTrue(c_network_adapter.get_properties_res.FaultCodeFromNcsi)
        lu.assertTrue(smbus_health_task_paused)
        lu.assertNil(c_network_adapter.smbus_schedulers.health_task.data)
        lu.assertTrue(collect_log_called)
        collect_log_called = false
        
        -- 测试2: NCSI成功时，FaultState == 0时不收集日志
        test_data = {
            health_status = 0,
            error_codes = {}
        }
        c_network_adapter.smbus_collect_status = 0
        mock_scheduler.data_cb(test_data)
        lu.assertEquals(c_network_adapter.FaultState, 0)
        
        -- 测试3: NCSI成功时，如果SMBUS任务已暂停，不重复暂停
        c_network_adapter.smbus_schedulers.health_task.is_paused = true
        test_data = {
            health_status = 0,
            error_codes = {}
        }
        mock_scheduler.data_cb(test_data)
        
        -- 测试4: NCSI失败时，未达到阈值不启动SMBUS备选
        c_network_adapter.get_properties_res.FaultCodeFromNcsi = true
        c_network_adapter.smbus_schedulers.health_task.is_paused = true
        smbus_health_task_resumed = false
        -- 失败2次（阈值是3）
        for i = 1, 2 do
            mock_scheduler.error_cb()
        end
        lu.assertFalse(c_network_adapter.get_properties_res.FaultCodeFromNcsi)
        lu.assertFalse(smbus_health_task_resumed)
        
        -- 测试5: NCSI失败达到阈值时，启动SMBUS备选
        -- 再失败1次，总共3次
        mock_scheduler.error_cb()
        lu.assertTrue(smbus_health_task_resumed)
        lu.assertFalse(c_network_adapter.smbus_schedulers.health_task.is_paused)
        
        -- 测试6: NCSI失败后成功，重置失败计数并暂停SMBUS备选
        c_network_adapter.smbus_schedulers.health_task.is_paused = false
        test_data = {
            health_status = 0,
            error_codes = {}
        }
        mock_scheduler.data_cb(test_data)
        -- 再次失败，应该需要重新计数到3次才启动
        c_network_adapter.smbus_schedulers.health_task.is_paused = true
        for i = 1, 2 do
            mock_scheduler.error_cb()
        end
        mock_scheduler.error_cb()
        lu.assertTrue(smbus_health_task_resumed)
        
        -- 测试7: 错误回调 - smbios_status != 3时，重置FaultState和FaultCode
        c_network_adapter.smbios_status = 2
        c_network_adapter.FaultState = 1
        mock_scheduler.error_cb()
        lu.assertEquals(c_network_adapter.FaultState, 0)
        
        -- 测试8: 错误回调 - smbios_status == 3时，不重置
        c_network_adapter.smbios_status = 3
        c_network_adapter.FaultState = 1
        mock_scheduler.error_cb()
        lu.assertEquals(c_network_adapter.FaultState, 1)
        
        -- 测试9: 空故障代码时，设置为'N/A'
        test_data = {
            health_status = 1,
            error_codes = {}
        }
        mock_scheduler.data_cb(test_data)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
        
        -- 测试10: smbus_collect_status == LOG_DUMP_BUSY时，不收集日志
        c_network_adapter.smbus_collect_status = 1  -- LOG_DUMP_BUSY
        test_data = {
            health_status = 1,
            error_codes = {1, 2, 0}
        }
        mock_scheduler.data_cb(test_data)
        
        -- 验证scheduler被添加到ncsi_schedulers（在还原前验证）
        lu.assertEquals(#c_network_adapter.ncsi_schedulers, 1)
        lu.assertEquals(c_network_adapter.ncsi_schedulers[1], mock_scheduler)
    end)
    
    -- 还原（无论成功还是失败都要还原）
    log.notice = original_notice
    log.error = original_error
    log.info = original_info
    c_network_adapter.NodeId = original_NodeId
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    -- 先清除测试过程中可能添加的 health_task
    if c_network_adapter.smbus_schedulers then
        c_network_adapter.smbus_schedulers.health_task = nil
    end
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.FaultState = original_FaultState
    c_network_adapter.FaultCode = original_FaultCode
    c_network_adapter.smbus_collect_status = original_smbus_collect_status
    c_network_adapter.smbus_has_collected = original_smbus_has_collected
    c_network_adapter.collect_log_by_smbus_task = original_collect_log_by_smbus_task
    
    -- 如果测试失败，重新抛出错误
    if not ok then
        error(err)
    end
end

function TEST_network_adapter:test_update_port_bdf()
    local skynet = require 'skynet'
    local fructl = require 'infrastructure.fructl'
    local origin_sleep = skynet.sleep
    local origin_get_power_status = fructl.get_power_status
    local origin_verify_vid_did_svid_sdid = c_network_adapter.verify_vid_did_svid_sdid
    local original_DevBus = c_network_adapter.DevBus
    local original_Type = c_network_adapter.Type
    local original_SocketId = c_network_adapter.SocketId
    local original_Bus = c_network_adapter.Bus
    local original_Device = c_network_adapter.Device

    c_network_adapter.verify_vid_did_svid_sdid = function()
        return true
    end

    fructl.get_power_status = function()
        return 'ON'
    end
    
    skynet.sleep = function()
        return
    end

    c_network_adapter.DevBus = 0
    c_network_adapter.Type = 1
    c_network_adapter.SocketId = 1
    c_network_adapter.Bus = 5
    c_network_adapter.Device= 16
    local ports = {{PortID = 1, BDF = ''}}
    c_network_adapter:update_port_bdf(ports)
    lu.assertEquals(ports[1].BDF, '0000:06:10.1')
    c_network_adapter.verify_vid_did_svid_sdid = origin_verify_vid_did_svid_sdid
    skynet.sleep = origin_sleep
    fructl.get_power_status = origin_get_power_status
    if original_DevBus ~= nil then
        c_network_adapter.DevBus = original_DevBus
    end
    if original_Type ~= nil then
        c_network_adapter.Type = original_Type
    end
    if original_SocketId ~= nil then
        c_network_adapter.SocketId = original_SocketId
    end
    if original_Bus ~= nil then
        c_network_adapter.Bus = original_Bus
    end
    if original_Device ~= nil then
        c_network_adapter.Device = original_Device
    end
end

function TEST_network_adapter:test_update_smbios_status()
    local temp_collection = c_optical_module.collection
    c_optical_module.collection = {
        fetch = function(test)
            return {
                {
                    update_smbios_status = function(self, value)
                        c_network_adapter.smbios_status = value
                    end
                }    
            }
        end
    }

    c_network_adapter:update_smbios_status('TestStaus')
    c_optical_module.collection = temp_collection
    lu.assertEquals(c_network_adapter.smbios_status, 'TestStaus')
end

function TEST_network_adapter:test_gc_task()
    local skynet = require('skynet')
    local log = require('mc.logging')

    -- 备份原始函数
    local original_notice = log.notice
    local original_fork_loop = skynet.fork_loop
    local original_sleep = skynet.sleep

    local log_calls = {}

    -- 模拟log.notice来捕获调用
    log.notice = function(self, fmt, ...)
        table.insert(log_calls, { type = 'notice', fmt = fmt, args = { ... } })
    end

    -- 模拟skynet.sleep立即返回
    skynet.sleep = function()
        return
    end

    -- 模拟skynet.fork_loop，执行回调但避免无限循环
    skynet.fork_loop = function(config, callback)
        if callback then
            -- 创建模拟环境，防止while循环无限运行
            local sleep_count = 0
            local original_inner_sleep = skynet.sleep

            skynet.sleep = function(time)
                sleep_count = sleep_count + 1
                -- 第一次sleep后（触发我们要测试的日志消息），
                -- 抛出错误来跳出任何while循环
                if sleep_count > 1 then
                    error("Test break - avoiding infinite loop")
                end
            end

            -- 在保护模式下执行回调以捕获我们的测试中断
            local success, err = pcall(callback)

            -- 恢复内部的sleep函数
            skynet.sleep = original_inner_sleep

            -- 如果失败不是由于我们的测试中断，则重新抛出错误
            if not success and not string.find(err, "Test break") then
                error(err)
            end
        end
    end

    network_adapter_app:gc_task()

    local found_log = false
    for _, call in ipairs(log_calls) do
        if call.type == 'notice' and string.find(call.fmt, 'gc task start') then
            found_log = true
            break
        end
    end

    -- 恢复原始函数
    log.notice = original_notice
    skynet.fork_loop = original_fork_loop
    skynet.sleep = original_sleep

    lu.assertTrue(found_log)
end

-- 测试 reset_ncsi_port_type 函数，在重启前将 NcsiNCPortInfo.Type 置为 ExternalPCIe 并调用 save
function TEST_network_adapter:test_reset_ncsi_port_type()
    -- 构造两个假的 NcsiNCPortInfo 对象
    local saved_objs = {}
    local port1 = { Type = 'OtherType1' }
    function port1:save()
        table.insert(saved_objs, self)
    end

    local port2 = { Type = 'OtherType2' }
    function port2:save()
        table.insert(saved_objs, self)
    end

    -- 构造假的 db.select 接口，返回带有 fold 方法的对象
    local select_called = false
    local mock_db = {
        NcsiNCPortInfo = {}
    }
    function mock_db:select(class)
        lu.assertEquals(class, self.NcsiNCPortInfo)
        select_called = true
        return {
            fold = function(_, f)
                f(port1)
                f(port2)
            end
        }
    end

    -- 用 network_adapter_app 作为 metatable，挂载 reset_ncsi_port_type 方法
    local app = { db = mock_db }
    setmetatable(app, { __index = network_adapter_app })

    -- 调用待测函数
    app:reset_ncsi_port_type()

    -- 校验每个端口对象的 Type 被置为 ExternalPCIe，且 save 被调用
    lu.assertTrue(select_called)
    lu.assertEquals(port1.Type, 'ExternalPCIe')
    lu.assertEquals(port2.Type, 'ExternalPCIe')
    lu.assertEquals(#saved_objs, 2)
    lu.assertEquals(saved_objs[1], port1)
    lu.assertEquals(saved_objs[2], port2)
end

-- 测试update_chip_temp_by_ncsi函数的NCSI恢复时停止SMBUS fallback逻辑
function TEST_network_adapter:test_update_chip_temp_by_ncsi_ncsi_recovered_stops_smbus_fallback()
    -- 备份原始状态
    local original_smbios_status = c_network_adapter.smbios_status
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    
    -- 初始化测试状态
    c_network_adapter.smbios_status = 3
    c_network_adapter.get_properties_res = {}
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.smbus_schedulers = {}
    
    -- 模拟SMBUS任务正在运行
    local smbus_task_paused = false
    c_network_adapter.smbus_schedulers.chip_temp_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_adapter.ncsi_config_obj = {
        ChipTemp = function() return scheduler_mock end
    }
    c_network_adapter.package_id = 0
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 模拟NCSI数据变化（恢复正常）
    local temp_value = 45
    scheduler_mock.on_data_change.callback(temp_value)
    
    -- 验证结果
    lu.assertEquals(c_network_adapter.TemperatureCelsius, temp_value)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 0)
    lu.assertTrue(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    lu.assertTrue(smbus_task_paused)
    lu.assertNil(c_network_adapter.smbus_schedulers.chip_temp_task.data)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试update_chip_temp_by_ncsi函数的NCSI失败时启动SMBUS fallback逻辑
function TEST_network_adapter:test_update_chip_temp_by_ncsi_ncsi_failed_activates_smbus_fallback()
    -- 备份原始状态
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    
    -- 初始化测试状态
    c_network_adapter.get_properties_res = {ChipTempFromNcsi = true}
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.smbus_schedulers = {}
    
    -- 模拟SMBUS任务已暂停
    local smbus_task_resumed = false
    c_network_adapter.smbus_schedulers.chip_temp_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_adapter.ncsi_config_obj = {
        ChipTemp = function() return scheduler_mock end
    }
    c_network_adapter.package_id = 0
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 模拟NCSI连续失败达到阈值（NCSI_FAILED_THRESHOLD = 3）
    for i = 1, 3 do
        scheduler_mock.on_error.error_callback()
    end
    
    -- 验证结果
    lu.assertFalse(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    lu.assertTrue(smbus_task_resumed)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试当SMBUS任务已暂停时NCSI恢复不会重复暂停
function TEST_network_adapter:test_ncsi_recovered_when_smbus_already_paused()
    -- 备份原始状态
    local original_smbios_status = c_network_adapter.smbios_status
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    
    -- 初始化测试状态
    c_network_adapter.smbios_status = 3
    c_network_adapter.get_properties_res = {}
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.smbus_schedulers = {}
    
    -- 模拟SMBUS任务已暂停
    local pause_call_count = 0
    c_network_adapter.smbus_schedulers.chip_temp_task = {
        is_paused = true,
        pause = function() pause_call_count = pause_call_count + 1 end,
        data = "existing_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_adapter.ncsi_config_obj = {
        ChipTemp = function() return scheduler_mock end
    }
    c_network_adapter.package_id = 0
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 模拟NCSI数据变化（恢复正常）
    local temp_value = 50
    scheduler_mock.on_data_change.callback(temp_value)
    
    -- 验证结果：不应该重复调用pause，因为任务已经暂停
    lu.assertEquals(pause_call_count, 0)
    lu.assertEquals(c_network_adapter.TemperatureCelsius, temp_value)
    lu.assertEquals(c_network_adapter.TemperatureStatus, 0)
    lu.assertTrue(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试当NCSI失败次数未达到阈值时不会启动SMBUS fallback
function TEST_network_adapter:test_ncsi_failed_below_threshold_no_smbus_fallback()
    -- 备份原始状态
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    
    -- 初始化测试状态
    c_network_adapter.get_properties_res = {ChipTempFromNcsi = true}
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.smbus_schedulers = {}
    
    -- 模拟SMBUS任务已暂停
    local resume_call_count = 0
    c_network_adapter.smbus_schedulers.chip_temp_task = {
        is_paused = true,
        resume = function() resume_call_count = resume_call_count + 1 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_adapter.ncsi_config_obj = {
        ChipTemp = function() return scheduler_mock end
    }
    c_network_adapter.package_id = 0
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 模拟NCSI失败但未达到阈值（只失败2次，阈值是3次）
    for i = 1, 2 do
        scheduler_mock.on_error.error_callback()
    end
    
    -- 验证结果：应该记录错误但不启动fallback
    lu.assertFalse(c_network_adapter.get_properties_res.ChipTempFromNcsi)
    lu.assertEquals(resume_call_count, 0) -- 不应该调用resume
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试pause后data被正确清空的逻辑
function TEST_network_adapter:test_pause_clears_data_correctly()
    -- 备份原始状态
    local original_smbios_status = c_network_adapter.smbios_status
    local original_get_properties_res = c_network_adapter.get_properties_res
    local original_ncsi_schedulers = c_network_adapter.ncsi_schedulers
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_package_id = c_network_adapter.package_id
    
    -- 初始化测试状态
    c_network_adapter.smbios_status = 3
    c_network_adapter.get_properties_res = {}
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.smbus_schedulers = {}
    
    -- 模拟SMBUS任务正在运行，包含数据
    local original_data = {temp = 25, status = 'active'}
    c_network_adapter.smbus_schedulers.chip_temp_task = {
        is_paused = false,
        pause = function() 
            c_network_adapter.smbus_schedulers.chip_temp_task.is_paused = true 
        end,
        data = original_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_adapter.ncsi_config_obj = {
        ChipTemp = function() return scheduler_mock end
    }
    c_network_adapter.package_id = 0
    
    -- 打桩外部依赖
    local original_table_insert = table.insert
    table.insert = function() end
    
    -- 执行函数
    c_network_adapter:update_chip_temp_by_ncsi()
    
    -- 模拟NCSI数据变化（恢复正常）
    local temp_value = 35
    scheduler_mock.on_data_change.callback(temp_value)
    
    -- 验证任务被暂停
    lu.assertTrue(c_network_adapter.smbus_schedulers.chip_temp_task.is_paused)
    
    -- 验证data被清空
    lu.assertNil(c_network_adapter.smbus_schedulers.chip_temp_task.data)
    
    -- 还原打桩
    table.insert = original_table_insert
    
    -- 还原全局状态
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.get_properties_res = original_get_properties_res
    c_network_adapter.ncsi_schedulers = original_ncsi_schedulers
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    c_network_adapter.package_id = original_package_id
end

-- 测试 update_power_status 函数 - PowerScanner无效的情况
function TEST_network_adapter:test_update_power_status_invalid_powerscanner()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_PowerScanner = c_network_adapter.PowerScanner
    local original_smbios_status = c_network_adapter.smbios_status
    local original_PowerStatus = c_network_adapter.PowerStatus
    local original_NodeId = c_network_adapter.NodeId
    local notice_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(notice_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.PowerScanner = 2  -- 无效值（不是0或1）
    c_network_adapter.smbios_status = 3
    c_network_adapter.PowerStatus = 255
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:update_power_status()
    
    -- 应该记录notice日志
    lu.assertEquals(#notice_calls, 1)
    lu.assertEquals(notice_calls[1].fmt, 'PowerScanner is %s, BiosStatus is %s, NodeId:%s')
    lu.assertEquals(notice_calls[1].args[1], 2)
    lu.assertEquals(notice_calls[1].args[2], 3)
    lu.assertEquals(notice_calls[1].args[3], 'TestAdapter')
    
    -- PowerStatus应该设置为255（因为PowerScanner无效）
    lu.assertEquals(c_network_adapter.PowerStatus, 255)
    
    log.notice = original_notice
    c_network_adapter.PowerScanner = original_PowerScanner
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.PowerStatus = original_PowerStatus
    c_network_adapter.NodeId = original_NodeId
end

-- 测试 update_power_status 函数 - PowerScanner有效且smbios_status==3的情况
function TEST_network_adapter:test_update_power_status_valid_powerscanner_smbios_3()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_PowerScanner = c_network_adapter.PowerScanner
    local original_smbios_status = c_network_adapter.smbios_status
    local original_PowerStatus = c_network_adapter.PowerStatus
    local original_NodeId = c_network_adapter.NodeId
    local notice_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(notice_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.PowerScanner = 1  -- 有效值
    c_network_adapter.smbios_status = 3
    c_network_adapter.PowerStatus = 255
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:update_power_status()
    
    -- 应该记录notice日志
    lu.assertEquals(#notice_calls, 1)
    lu.assertEquals(notice_calls[1].fmt, 'PowerScanner is %s, BiosStatus is %s, NodeId:%s')
    lu.assertEquals(notice_calls[1].args[1], 1)
    lu.assertEquals(notice_calls[1].args[2], 3)
    lu.assertEquals(notice_calls[1].args[3], 'TestAdapter')
    
    -- PowerStatus应该设置为PowerScanner的值
    lu.assertEquals(c_network_adapter.PowerStatus, 1)
    
    log.notice = original_notice
    c_network_adapter.PowerScanner = original_PowerScanner
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.PowerStatus = original_PowerStatus
    c_network_adapter.NodeId = original_NodeId
end

-- 测试 update_power_status 函数 - PowerScanner有效但smbios_status!=3的情况
function TEST_network_adapter:test_update_power_status_valid_powerscanner_smbios_not_3()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_PowerScanner = c_network_adapter.PowerScanner
    local original_smbios_status = c_network_adapter.smbios_status
    local original_PowerStatus = c_network_adapter.PowerStatus
    local original_NodeId = c_network_adapter.NodeId
    local notice_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(notice_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.PowerScanner = 1  -- 有效值
    c_network_adapter.smbios_status = 4  -- 不等于3，BIOS未启动值（下电状态）
    c_network_adapter.PowerStatus = 255
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:update_power_status()
    
    -- 应该记录notice日志
    lu.assertEquals(#notice_calls, 1)
    lu.assertEquals(notice_calls[1].fmt, 'PowerScanner is %s, BiosStatus is %s, NodeId:%s')
    lu.assertEquals(notice_calls[1].args[1], 1)
    lu.assertEquals(notice_calls[1].args[2], 4)
    lu.assertEquals(notice_calls[1].args[3], 'TestAdapter')
    
    -- PowerStatus应该设置为255
    lu.assertEquals(c_network_adapter.PowerStatus, 255)
    
    log.notice = original_notice
    c_network_adapter.PowerScanner = original_PowerScanner
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.PowerStatus = original_PowerStatus
    c_network_adapter.NodeId = original_NodeId
end

-- 测试 update_power_status 函数 - PowerScanner为0的情况
function TEST_network_adapter:test_update_power_status_powerscanner_zero()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_PowerScanner = c_network_adapter.PowerScanner
    local original_smbios_status = c_network_adapter.smbios_status
    local original_PowerStatus = c_network_adapter.PowerStatus
    local original_NodeId = c_network_adapter.NodeId
    local notice_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(notice_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.PowerScanner = 0  -- 有效值
    c_network_adapter.smbios_status = 3
    c_network_adapter.PowerStatus = 255
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:update_power_status()
    
    -- 应该记录notice日志
    lu.assertEquals(#notice_calls, 1)
    lu.assertEquals(notice_calls[1].fmt, 'PowerScanner is %s, BiosStatus is %s, NodeId:%s')
    lu.assertEquals(notice_calls[1].args[1], 0)
    lu.assertEquals(notice_calls[1].args[2], 3)
    lu.assertEquals(notice_calls[1].args[3], 'TestAdapter')
    
    -- PowerStatus应该设置为PowerScanner的值（0）
    lu.assertEquals(c_network_adapter.PowerStatus, 0)
    
    log.notice = original_notice
    c_network_adapter.PowerScanner = original_PowerScanner
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.PowerStatus = original_PowerStatus
    c_network_adapter.NodeId = original_NodeId
end

-- 测试 set_smbios_status 函数 - GetSmBiosObjects返回空的情况
function TEST_network_adapter:test_set_smbios_status_empty_list()
    local client = require 'network_adapter.client'
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    
    client.GetSmBiosObjects = function()
        return nil
    end
    
    c_network_adapter.smbios_status = 0
    
    c_network_adapter:set_smbios_status()
    
    -- smbios_status应该保持不变
    lu.assertEquals(c_network_adapter.smbios_status, 0)
    
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
end

-- 测试 set_smbios_status 函数 - GetSmBiosObjects返回空表的情况
function TEST_network_adapter:test_set_smbios_status_empty_table()
    local client = require 'network_adapter.client'
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    
    client.GetSmBiosObjects = function()
        return {}
    end
    
    c_network_adapter.smbios_status = 0
    
    c_network_adapter:set_smbios_status()
    
    -- smbios_status应该保持不变
    lu.assertEquals(c_network_adapter.smbios_status, 0)
    
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
end

-- 测试 set_smbios_status 函数 - GetSmBiosObjects返回有对象的情况
function TEST_network_adapter:test_set_smbios_status_with_object()
    local client = require 'network_adapter.client'
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    
    local mock_smbios_obj = {
        SmBiosStatus = 3
    }
    
    client.GetSmBiosObjects = function()
        return {
            [1] = mock_smbios_obj
        }
    end
    
    c_network_adapter.smbios_status = 0
    
    c_network_adapter:set_smbios_status()
    
    -- smbios_status应该被设置为对象的SmBiosStatus
    lu.assertEquals(c_network_adapter.smbios_status, 3)
    
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
end

-- 测试 set_smbios_status 函数 - GetSmBiosObjects返回多个对象的情况（只取第一个）
function TEST_network_adapter:test_set_smbios_status_multiple_objects()
    local client = require 'network_adapter.client'
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    
    local mock_smbios_obj1 = {
        SmBiosStatus = 3
    }
    local mock_smbios_obj2 = {
        SmBiosStatus = 1
    }
    
    client.GetSmBiosObjects = function()
        return {
            [1] = mock_smbios_obj1,
            [2] = mock_smbios_obj2
        }
    end
    
    c_network_adapter.smbios_status = 0
    
    c_network_adapter:set_smbios_status()
    
    -- smbios_status应该被设置为第一个对象的SmBiosStatus
    lu.assertEquals(c_network_adapter.smbios_status, 3)
    
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
end

-- 测试 set_smbios_status 函数 - GetSmBiosObjects抛出异常的情况
function TEST_network_adapter:test_set_smbios_status_pcall_failed()
    local log = require('mc.logging')
    local client = require 'network_adapter.client'
    local original_error = log.error
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    local original_NodeId = c_network_adapter.NodeId
    local error_calls = {}
    
    log.error = function(self, fmt, ...)
        table.insert(error_calls, {fmt = fmt, args = {...}})
    end
    
    client.GetSmBiosObjects = function()
        error('GetSmBiosObjects failed')
    end
    
    c_network_adapter.smbios_status = 0
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:set_smbios_status()
    
    -- 应该记录错误日志
    lu.assertEquals(#error_calls, 1)
    lu.assertEquals(error_calls[1].fmt, 'get SmBios objects failed: %s')
    lu.assertEquals(error_calls[1].args[1], 'TestAdapter')
    
    -- smbios_status应该保持不变
    lu.assertEquals(c_network_adapter.smbios_status, 0)
    
    log.error = original_error
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.NodeId = original_NodeId
end

-- 测试 set_smbios_status 函数 - smbios_obj没有SmBiosStatus字段的情况
function TEST_network_adapter:test_set_smbios_status_no_smbiosstatus_field()
    local client = require 'network_adapter.client'
    local original_GetSmBiosObjects = client.GetSmBiosObjects
    local original_smbios_status = c_network_adapter.smbios_status
    
    local mock_smbios_obj = {}
    
    client.GetSmBiosObjects = function()
        return {
            [1] = mock_smbios_obj
        }
    end
    
    c_network_adapter.smbios_status = 0
    
    -- 使用pcall捕获可能的错误
    local ok, _ = pcall(function()
        c_network_adapter:set_smbios_status()
    end)
    
    -- 如果出错，smbios_status应该保持不变
    if not ok then
        lu.assertEquals(c_network_adapter.smbios_status, 0)
    end
    
    client.GetSmBiosObjects = original_GetSmBiosObjects
    c_network_adapter.smbios_status = original_smbios_status
end

-- 测试 update_power_status 函数 - PowerScanner为255（默认值）的情况
function TEST_network_adapter:test_update_power_status_powerscanner_default_255()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_PowerScanner = c_network_adapter.PowerScanner
    local original_smbios_status = c_network_adapter.smbios_status
    local original_PowerStatus = c_network_adapter.PowerStatus
    local original_NodeId = c_network_adapter.NodeId
    local notice_calls = {}
    
    log.notice = function(self, fmt, ...)
        table.insert(notice_calls, {fmt = fmt, args = {...}})
    end
    
    c_network_adapter.PowerScanner = 255  -- 默认值，无效值
    c_network_adapter.smbios_status = 3
    c_network_adapter.PowerStatus = 255
    c_network_adapter.NodeId = 'TestAdapter'
    
    c_network_adapter:update_power_status()
    
    -- 应该记录notice日志
    lu.assertEquals(#notice_calls, 1)
    lu.assertEquals(notice_calls[1].fmt, 'PowerScanner is %s, BiosStatus is %s, NodeId:%s')
    lu.assertEquals(notice_calls[1].args[1], 255)
    lu.assertEquals(notice_calls[1].args[2], 3)
    lu.assertEquals(notice_calls[1].args[3], 'TestAdapter')
    
    -- PowerStatus应该设置为255（因为PowerScanner无效）
    lu.assertEquals(c_network_adapter.PowerStatus, 255)
    
    log.notice = original_notice
    c_network_adapter.PowerScanner = original_PowerScanner
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.PowerStatus = original_PowerStatus
    c_network_adapter.NodeId = original_NodeId
end

function TEST_network_adapter:test_update_related_port_info()
    local om_obj = {
        RelatedNetworkPorts = {1},
        get_parent = function()
            return {
                get_parent = function()
                    return {
                        BoardName = 'Ethernet Adapter Board'
                    }
                end
            }
        end,
        SilkText = 'ETH 8'
    }
    pcall(function()
        c_network_adapter:update_related_port_info(om_obj)
    end)
    lu.assertEquals(om_obj.cache_location, '[Ethernet Adapter Board][ETH 8]')
end

function TEST_network_adapter:test_set_prop_on_removing()
    local pcie_slot1 = {
        path = "/bmc/kepler/Systems/1/PCIeSlots/PCIeSlot_1_000101",
        SlotId = 1,
        SupportedComponentTypes = string.pack('I', 8),
        ReadyToRemove = 0,
        PowerControl_PACKED = function(obj, value)
            obj.ReadyToRemove = value == 'Off' and 1 or 0
        end
    }
    local pcie_slot2 = {
        path = "/bmc/kepler/Systems/1/PCIeSlots/PCIeSlot_1_000101",
        SlotId = 2,
        SupportedComponentTypes = string.pack('I', 83),
        ReadyToRemove = 0,
        PowerControl_PACKED = function(obj, value)
            obj.ReadyToRemove = value == 'Off' and 1 or 0
        end
    }
    local pcie_slot3 = {
        path = "/bmc/kepler/Systems/1/PCIeSlots/PCIeSlot_1_000101",
        SlotId = 1,
        SupportedComponentTypes = string.pack('I', 83),
        ReadyToRemove = 0,
        PowerControl_PACKED = function(obj, value)
            obj.ReadyToRemove = value == 'Off' and 1 or 0
        end
    }
    local client = require 'network_adapter.client'
    local origin_func = client.GetPCIeSlotObjects
    client.GetPCIeSlotObjects = function()
        return {pcie_slot1, pcie_slot2, pcie_slot3}
    end
    local origin_foreach = client.ForeachComponentObjects
    client.ForeachComponentObjects = function(_, cb)
        local component_obj = {
            Name = 'test_name',
            Type = 83
        }
        cb(component_obj)
    end
    local adapter = {
        SlotNumber = 1,
        component_type = 0,
        DeviceLocator = 'test_name',
        ObjectName = 'NetworkAdapter_1_00010101',
        set_prop_on_removing = c_network_adapter.set_prop_on_removing,
        is_matched_slot = c_network_adapter.is_matched_slot,
        update_component_type = c_network_adapter.update_component_type
    }
    adapter:set_prop_on_removing(true)
    lu.assertEquals(pcie_slot1.ReadyToRemove, 0)
    lu.assertEquals(pcie_slot2.ReadyToRemove, 0)
    lu.assertEquals(pcie_slot3.ReadyToRemove, 1)
    client.GetPCIeSlotObjects = origin_func
    client.ForeachComponentObjects = origin_foreach
end

function TEST_network_adapter:test_register_slot_power_state_listen()
    local pcie_slot3 = {
        path = "/bmc/kepler/Systems/1/PCIeSlots/PCIeSlot_1_000101",
        SlotId = 1,
        SupportedComponentTypes = string.pack('I', 83),
        PowerState = 'Off'
    }
    local client = require 'network_adapter.client'
    local origin_func = client.GetPCIeSlotObjects
    client.GetPCIeSlotObjects = function()
        return {pcie_slot3}
    end
    local origin_properties_changed = client.OnPCIeSlotPropertiesChanged
    client.OnPCIeSlotPropertiesChanged = function(_, cb)
        local values = {
            PowerState = {value = function() return 'Off' end}
        }
        local path = "/bmc/kepler/Systems/1/PCIeSlots/PCIeSlot_1_000101"
        cb(values, path, _)
    end
    local adapter = {
        SlotNumber = 1,
        SlotPowerState = '',
        component_type = 83,
        ObjectName = 'NetworkAdapter_1_00010101',
        register_slot_power_state_listen = c_network_adapter.register_slot_power_state_listen,
        is_matched_slot = c_network_adapter.is_matched_slot
    }
    adapter:register_slot_power_state_listen()
    lu.assertEquals(adapter.SlotPowerState, 'Off')
    client.GetPCIeSlotObjects = origin_func
    client.OnPCIeSlotPropertiesChanged = origin_properties_changed
end

-- 测试 get_dpu_host_access 函数
function TEST_network_adapter:test_get_dpu_host_access()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj

    -- 测试用例1: 没有GetHostAccess方法
    c_network_adapter.ncsi_config_obj = {}
    local compcode, data = c_network_adapter:get_dpu_host_access()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例2: 有GetHostAccess方法但返回nil
    c_network_adapter.ncsi_config_obj = {
        GetHostAccess = function(params)
            return {value = function() return nil end}
        end
    }
    compcode, data = c_network_adapter:get_dpu_host_access()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)
    
    -- 测试用例3: 返回值为 0xff（无效值）→ 返回 cc.CommandDisabled, 0
    c_network_adapter.ncsi_config_obj = {
        GetHostAccess = function(params)
            return {value = function() return 0xff end}
        end
    }
    compcode, data = c_network_adapter:get_dpu_host_access()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例4: 返回值为 0xfe（未指定错误）→ 返回 cc.UnspecifiedError, 0
    c_network_adapter.ncsi_config_obj = {
        GetHostAccess = function(params)
            return {value = function() return 0xfe end}
        end
    }
    compcode, data = c_network_adapter:get_dpu_host_access()
    lu.assertEquals(compcode, cc.UnspecifiedError)
    lu.assertEquals(data, 0)

    -- 测试用例5: 正常返回有效值（0x01）→ 返回 cc.Success, 0x01
    local test_value = 0x01
    c_network_adapter.ncsi_config_obj = {
        GetHostAccess = function(params)
            return {value = function() return test_value end}
        end
    }
    compcode, data = c_network_adapter:get_dpu_host_access()
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(data, test_value)

    -- 还原原始值
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
end

-- 测试 set_dpu_host_access 函数
function TEST_network_adapter:test_set_dpu_host_access()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_ipmi_operation_log = ipmi.ipmi_operation_log
    local ctx = {ChanType = 1}
    local ipmi_logs = {}
    ipmi.ipmi_operation_log = function(ctx, module, msg)
        table.insert(ipmi_logs, {module = module, msg = msg})
    end

    -- 测试用例1: 没有两个方法 → cc.CommandDisabled
    c_network_adapter.ncsi_config_obj = {}
    local compcode = c_network_adapter:set_dpu_host_access(0x00, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例2: 传入无效req_status → cc.InvalidCommand（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0x00 end} end,
        SetHostAccessEnable = function() return {value = function() return 0x01 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x02, ctx)
    lu.assertEquals(compcode, cc.InvalidCommand)

    -- 测试用例3: 调用Disable返回nil → cc.CommandDisabled（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return nil end} end,
        SetHostAccessEnable = function() return {value = function() return 0x01 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x00, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例4: 调用Disable返回0xff → cc.CommandDisabled（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0xff end} end,
        SetHostAccessEnable = function() return {value = function() return 0x01 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x00, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例5: 返回值不匹配 → cc.UnspecifiedError（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0x00 end} end,
        SetHostAccessEnable = function() return {value = function() return 0x00 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x01, ctx)
    lu.assertEquals(compcode, cc.UnspecifiedError)

    -- 测试用例6: 调用Enable返回0xfe → cc.UnspecifiedError（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0x00 end} end,
        SetHostAccessEnable = function() return {value = function() return 0xfe end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x01, ctx)
    lu.assertEquals(compcode, cc.UnspecifiedError)

    -- 测试用例7: 正常设置Disable → cc.Success（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0x00 end} end,
        SetHostAccessEnable = function() return {value = function() return 0x01 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x00, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu host access Disable successfully")

    -- 测试用例8: 正常设置Enable → cc.Success（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetHostAccessDisable = function() return {value = function() return 0x00 end} end,
        SetHostAccessEnable = function() return {value = function() return 0x01 end} end
    }
    compcode = c_network_adapter:set_dpu_host_access(0x01, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu host access Enable successfully")

    -- 还原
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    ipmi.ipmi_operation_log = original_ipmi_operation_log
end

-- 测试 get_dpu_extended_host_privilege 函数
function TEST_network_adapter:test_get_dpu_extended_host_privilege()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj

    -- 测试用例1: 没有GetExtendedHostPrivilege方法 → cc.CommandDisabled, 0
    c_network_adapter.ncsi_config_obj = {}
    local compcode, data = c_network_adapter:get_dpu_extended_host_privilege()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例2: 有方法但返回nil → cc.CommandDisabled, 0
    c_network_adapter.ncsi_config_obj = {
        GetExtendedHostPrivilege = function() return {value = function() return nil end} end
    }
    compcode, data = c_network_adapter:get_dpu_extended_host_privilege()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例3: 返回0xff → cc.CommandDisabled, 0
    c_network_adapter.ncsi_config_obj = {
        GetExtendedHostPrivilege = function()
        return {value = function() return 0xff end} end
    }
    compcode, data = c_network_adapter:get_dpu_extended_host_privilege()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例4: 返回0xfe → cc.UnspecifiedError, 0
    c_network_adapter.ncsi_config_obj = {
        GetExtendedHostPrivilege = function()
        return {value = function() return 0xfe end} end
    }
    compcode, data = c_network_adapter:get_dpu_extended_host_privilege()
    lu.assertEquals(compcode, cc.UnspecifiedError)
    lu.assertEquals(data, 0)

    -- 测试用例5: 正常返回有效值 → cc.Success, 0x02
    local test_value = 0x02
    c_network_adapter.ncsi_config_obj = {
        GetExtendedHostPrivilege = function()
        return {value = function() return test_value end} end
    }
    compcode, data = c_network_adapter:get_dpu_extended_host_privilege()
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(data, test_value)

    -- 还原
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
end

-- 测试 set_dpu_extended_host_privilege 函数
function TEST_network_adapter:test_set_dpu_extended_host_privilege()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_ipmi_operation_log = ipmi.ipmi_operation_log
    local ctx = {ChanType = 1}
    local ipmi_logs = {}
    ipmi.ipmi_operation_log = function(ctx, module, msg)
        table.insert(ipmi_logs, {module = module, msg = msg})
    end

    -- 测试用例1: 没有两个方法 → cc.CommandDisabled
    c_network_adapter.ncsi_config_obj = {}
    local compcode = c_network_adapter:set_dpu_extended_host_privilege(0, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例2: 传入无效req_status → cc.InvalidCommand（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetExtendedHostPrivilegeDisable = function() return {value = function() return true end} end,
        SetExtendedHostPrivilegeEnable = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_extended_host_privilege(2, ctx)
    lu.assertEquals(compcode, cc.InvalidCommand)

    -- 测试用例3: 调用Disable返回nil → cc.CommandDisabled（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetExtendedHostPrivilegeDisable = function() return {value = function() return false end} end,
        SetExtendedHostPrivilegeEnable = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_extended_host_privilege(0, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例4: 正常设置Enable → cc.Success（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetExtendedHostPrivilegeDisable = function() return {value = function() return true end} end,
        SetExtendedHostPrivilegeEnable = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_extended_host_privilege(1, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu extended host privilege Enable successfully")

    -- 测试用例5: 正常设置Disable → cc.Success（同时设定两个方法）
    compcode = c_network_adapter:set_dpu_extended_host_privilege(0, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu extended host privilege Disable successfully")

    -- 还原
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    ipmi.ipmi_operation_log = original_ipmi_operation_log
end

-- 测试 get_dpu_mode 函数
function TEST_network_adapter:test_get_dpu_mode()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj

    -- 测试用例1: 没有GetMode方法 → cc.CommandDisabled, 0
    c_network_adapter.ncsi_config_obj = {}
    local compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例2: 有方法但返回nil → cc.CommandDisabled, 0（统一模板）
    c_network_adapter.ncsi_config_obj = {
        GetMode = function() return {value = function() return nil end} end
    }
    compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例3: 返回0xff → cc.CommandDisabled, 0（统一模板）
    c_network_adapter.ncsi_config_obj = {
        GetMode = function() return {value = function() return 0xff end} end
    }
    compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.CommandDisabled)
    lu.assertEquals(data, 0)

    -- 测试用例4: 返回0xfe → cc.UnspecifiedError, 0（统一模板）
    c_network_adapter.ncsi_config_obj = {
        GetMode = function() return {value = function() return 0xfe end} end
    }
    compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.UnspecifiedError)
    lu.assertEquals(data, 0)

    -- 测试用例5: 正常返回0x00 → cc.Success, 0x00（统一模板）
    local test_value = 0x00
    c_network_adapter.ncsi_config_obj = {
        GetMode = function() return {value = function() return test_value end} end
    }
    compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(data, test_value)

    -- 测试用例6: 正常返回0x01 → cc.Success, 0x01（统一模板）
    test_value = 0x01
    c_network_adapter.ncsi_config_obj = {
        GetMode = function() return {value = function() return test_value end} end
    }
    compcode, data = c_network_adapter:get_dpu_mode()
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(data, test_value)

    -- 还原
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
end

-- 测试 set_dpu_mode 函数
function TEST_network_adapter:test_set_dpu_mode()
    local original_ncsi_config_obj = c_network_adapter.ncsi_config_obj
    local original_ipmi_operation_log = ipmi.ipmi_operation_log
    local ctx = {ChanType = 1}
    local ipmi_logs = {}
    ipmi.ipmi_operation_log = function(ctx, module, msg)
        table.insert(ipmi_logs, {module = module, msg = msg})
    end

    -- 测试用例1: 没有两个方法 → cc.CommandDisabled
    c_network_adapter.ncsi_config_obj = {}
    local compcode = c_network_adapter:set_dpu_mode(0x00, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例2: 传入无效req_status → cc.InvalidCommand（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetModeFromDPUToNIC = function() return {value = function() return true end} end,
        SetModeFromNICToDPU = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_mode(0x02, ctx)
    lu.assertEquals(compcode, cc.InvalidCommand)

    -- 测试用例3: 调用DPUToNIC返回nil → cc.CommandDisabled（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetModeFromDPUToNIC = function() return {value = function() return false end} end,
        SetModeFromNICToDPU = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_mode(0x00, ctx)
    lu.assertEquals(compcode, cc.CommandDisabled)

    -- 测试用例4: 正常设置NICToDPU → cc.Success（同时设定两个方法）
    c_network_adapter.ncsi_config_obj = {
        SetModeFromDPUToNIC = function() return {value = function() return true end} end,
        SetModeFromNICToDPU = function() return {value = function() return true end} end
    }
    compcode = c_network_adapter:set_dpu_mode(0x01, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu mode to DPU successfully")

    -- 测试用例5: 正常设置DPUToNIC → cc.Success（同时设定两个方法）
    compcode = c_network_adapter:set_dpu_mode(0x00, ctx)
    lu.assertEquals(compcode, cc.Success)
    lu.assertEquals(ipmi_logs[#ipmi_logs].msg, "set dpu mode to NIC successfully")

    -- 还原
    c_network_adapter.ncsi_config_obj = original_ncsi_config_obj
    ipmi.ipmi_operation_log = original_ipmi_operation_log
end

-- 测试 Hi182X OpticalTemp response 函数
function TEST_network_adapter:test_hi182x_optical_temp_response()
    local libmgmt_protocol = require 'libmgmt_protocol'
    local log = require 'mc.logging'
    local original_create_array_parser = libmgmt_protocol.create_array_parser
    local original_debug = log.debug
    local debug_calls = {}
    
    -- Mock log:debug
    log.debug = function(self, fmt, ...)
        table.insert(debug_calls, {fmt = fmt, args = {...}})
    end
    
    local optical_temp_response = hi1822.smbus().properties.OpticalTemp.response
    
    -- 测试场景1: 正常温度值
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 60}
            }
        end
    end
    
    local result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 60, "second port temp should be 60")
    lu.assertEquals(#debug_calls, 0, "no debug log for normal temp")
    
    -- 测试场景2: temp == 0x7ffd (无效温度)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 0x7ffd},
                {temp = 60}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 0x7ffd, "first port temp should be 0x7ffd")
    lu.assertEquals(result[2], 60, "second port temp should be 60")
    lu.assertEquals(#debug_calls, 1, "should have one debug log")
    lu.assertEquals(debug_calls[1].fmt, 'optical temp for port %d is invalid')
    lu.assertEquals(debug_calls[1].args[1], 1)
    
    -- 测试场景3: temp == 0x7ffe (光模块不可用)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 0x7ffe}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 0x7ffe, "second port temp should be 0x7ffe")
    lu.assertEquals(#debug_calls, 1, "should have one debug log")
    lu.assertEquals(debug_calls[1].fmt, 'optical module is not available for port %d')
    lu.assertEquals(debug_calls[1].args[1], 2)
    
    -- 测试场景4: temp == 0x7fff (读取失败)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 0x7fff},
                {temp = 0x7fff}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 0x7fff, "first port temp should be 0x7fff")
    lu.assertEquals(result[2], 0x7fff, "second port temp should be 0x7fff")
    lu.assertEquals(#debug_calls, 2, "should have two debug logs")
    lu.assertEquals(debug_calls[1].fmt, 'Failed to read optical temp from port %d')
    lu.assertEquals(debug_calls[1].args[1], 1)
    lu.assertEquals(debug_calls[2].fmt, 'Failed to read optical temp from port %d')
    lu.assertEquals(debug_calls[2].args[1], 2)
    
    -- 测试场景5: parser 返回 nil
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return nil
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNil(result, "result should be nil when parser returns nil")
    lu.assertEquals(#debug_calls, 0, "no debug log when parser returns nil")
    
    -- 测试场景6: 混合场景 - 包含所有特殊值
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 0x7ffd},
                {temp = 0x7ffe},
                {temp = 0x7fff}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 0x7ffd, "second port temp should be 0x7ffd")
    lu.assertEquals(result[3], 0x7ffe, "third port temp should be 0x7ffe")
    lu.assertEquals(result[4], 0x7fff, "fourth port temp should be 0x7fff")
    lu.assertEquals(#debug_calls, 3, "should have three debug logs")
    
    -- 还原
    libmgmt_protocol.create_array_parser = original_create_array_parser
    log.debug = original_debug
end

-- 测试 Hi1822-FC OpticalTemp response 函数
function TEST_network_adapter:test_hi1822_fc_optical_temp_response()
    local libmgmt_protocol = require 'libmgmt_protocol'
    local log = require 'mc.logging'
    local original_create_array_parser = libmgmt_protocol.create_array_parser
    local original_debug = log.debug
    local debug_calls = {}
    
    -- Mock log:debug
    log.debug = function(self, fmt, ...)
        table.insert(debug_calls, {fmt = fmt, args = {...}})
    end
    
    local optical_temp_response = hi1822_fc.smbus().properties.OpticalTemp.response
    
    -- 测试场景1: 正常温度值
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 60}
            }
        end
    end
    
    local result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 60, "second port temp should be 60")
    lu.assertEquals(#debug_calls, 0, "no debug log for normal temp")
    
    -- 测试场景2: temp == 0x7ffd (无效温度)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 0x7ffd},
                {temp = 60}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 0x7ffd, "first port temp should be 0x7ffd")
    lu.assertEquals(result[2], 60, "second port temp should be 60")
    lu.assertEquals(#debug_calls, 1, "should have one debug log")
    lu.assertEquals(debug_calls[1].fmt, 'optical temp for port %d is invalid')
    lu.assertEquals(debug_calls[1].args[1], 1)
    
    -- 测试场景3: temp == 0x7ffe (光模块不可用)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 0x7ffe}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 0x7ffe, "second port temp should be 0x7ffe")
    lu.assertEquals(#debug_calls, 1, "should have one debug log")
    lu.assertEquals(debug_calls[1].fmt, 'optical module is not available for port %d')
    lu.assertEquals(debug_calls[1].args[1], 2)
    
    -- 测试场景4: temp == 0x7fff (读取失败)
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 0x7fff},
                {temp = 0x7fff}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 0x7fff, "first port temp should be 0x7fff")
    lu.assertEquals(result[2], 0x7fff, "second port temp should be 0x7fff")
    lu.assertEquals(#debug_calls, 2, "should have two debug logs")
    lu.assertEquals(debug_calls[1].fmt, 'Failed to read optical temp from port %d')
    lu.assertEquals(debug_calls[1].args[1], 1)
    lu.assertEquals(debug_calls[2].fmt, 'Failed to read optical temp from port %d')
    lu.assertEquals(debug_calls[2].args[1], 2)
    
    -- 测试场景5: parser 返回 nil
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return nil
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNil(result, "result should be nil when parser returns nil")
    lu.assertEquals(#debug_calls, 0, "no debug log when parser returns nil")
    
    -- 测试场景6: 混合场景 - 包含所有特殊值
    debug_calls = {}
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return {
                {temp = 50},
                {temp = 0x7ffd},
                {temp = 0x7ffe},
                {temp = 0x7fff}
            }
        end
    end
    
    result = optical_temp_response('test_data')
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(result[1], 50, "first port temp should be 50")
    lu.assertEquals(result[2], 0x7ffd, "second port temp should be 0x7ffd")
    lu.assertEquals(result[3], 0x7ffe, "third port temp should be 0x7ffe")
    lu.assertEquals(result[4], 0x7fff, "fourth port temp should be 0x7fff")
    lu.assertEquals(#debug_calls, 3, "should have three debug logs")
    
    -- 还原
    libmgmt_protocol.create_array_parser = original_create_array_parser
    log.debug = original_debug
end

-- 测试 join_fault_codes 函数（通过 update_health 方法间接测试）
function TEST_network_adapter:test_join_fault_codes()
    local log = require('mc.logging')
    local original_notice = log.notice
    local original_info = log.info
    local original_NodeId = c_network_adapter.NodeId
    local original_smbus_schedulers = c_network_adapter.smbus_schedulers
    local original_smbus_config_obj = c_network_adapter.smbus_config_obj
    local original_FaultState = c_network_adapter.FaultState
    local original_FaultCode = c_network_adapter.FaultCode
    local original_smbios_status = c_network_adapter.smbios_status
    local original_smbus_collect_status = c_network_adapter.smbus_collect_status
    local original_smbus_has_collected = c_network_adapter.smbus_has_collected
    local original_last_fault_code = c_network_adapter.last_fault_code
    local original_collect_log_by_smbus_task = c_network_adapter.collect_log_by_smbus_task
    local original_package_id = c_network_adapter.package_id
    local log_calls = {}
    local mock_scheduler = {}
    local mock_fault_code_scheduler = {}
    
    -- 使用pcall确保即使出错也能还原
    local ok, err = pcall(function()
        log.notice = function(self, fmt, ...) 
            table.insert(log_calls, {type='notice', fmt=fmt, args={...}}) 
        end
        log.info = function(self, fmt, ...) 
            table.insert(log_calls, {type='info', fmt=fmt, args={...}}) 
        end
        
        c_network_adapter.NodeId = "TestAdapter"
        c_network_adapter.smbus_schedulers = {}
        c_network_adapter.smbios_status = 2
        c_network_adapter.smbus_collect_status = 0  -- LOG_DUMP_IDLE
        c_network_adapter.smbus_has_collected = false
        c_network_adapter.last_fault_code = {}
        c_network_adapter.FaultState = 0
        c_network_adapter.FaultCode = 'N/A'
        
        -- 模拟collect_log_by_smbus_task
        c_network_adapter.collect_log_by_smbus_task = function()
        end
        
        -- 模拟SMBUS配置对象
        c_network_adapter.smbus_config_obj = {
            Health = function() 
                return mock_scheduler 
            end,
            FaultCode = function() 
                return mock_fault_code_scheduler 
            end
        }
        
        mock_scheduler.on_data_change = { 
            on = function(self, cb) 
                mock_scheduler.data_cb = cb 
            end 
        }
        mock_scheduler.on_error = { 
            on = function(self, cb) 
                mock_scheduler.error_cb = cb 
            end 
        }
        mock_scheduler.start = function() end
        
        -- 执行函数
        c_network_adapter:update_health()
        
        -- 测试用例1: nil 输入（type(nil) ~= "table" 会返回 ""，导致设置为 'N/A'）
        mock_fault_code_scheduler.value = function() return nil end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
        
        -- 测试用例2: true 输入（type(true) ~= "table" 会返回 ""，导致设置为 'N/A'）
        mock_fault_code_scheduler.value = function() return true end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
        
        -- 测试用例3: 空数组
        mock_fault_code_scheduler.value = function() return {} end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
        
        -- 测试用例4: 单个错误码
        mock_fault_code_scheduler.value = function() return {1} end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, '1')
        
        -- 测试用例5: 多个错误码
        mock_fault_code_scheduler.value = function() return {1, 2, 3} end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, '1, 2, 3')
        
        -- 测试用例6: 错误码中包含0，遇到0就停止
        mock_fault_code_scheduler.value = function() return {1, 2, 0, 3, 4} end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, '1, 2')
        
        -- 测试用例7: 第一个元素就是0
        mock_fault_code_scheduler.value = function() return {0, 1, 2} end
        c_network_adapter.last_fault_code = {}
        mock_scheduler.data_cb(1)
        lu.assertEquals(c_network_adapter.FaultCode, 'N/A')
    end)
    
    -- 还原（无论成功还是失败都要还原）
    log.notice = original_notice
    log.info = original_info
    c_network_adapter.NodeId = original_NodeId
    -- 先清除测试过程中可能添加的 health_task
    if c_network_adapter.smbus_schedulers then
        c_network_adapter.smbus_schedulers.health_task = nil
    end
    c_network_adapter.smbus_schedulers = original_smbus_schedulers
    c_network_adapter.smbus_config_obj = original_smbus_config_obj
    c_network_adapter.FaultState = original_FaultState
    c_network_adapter.FaultCode = original_FaultCode
    c_network_adapter.smbios_status = original_smbios_status
    c_network_adapter.smbus_collect_status = original_smbus_collect_status
    c_network_adapter.smbus_has_collected = original_smbus_has_collected
    c_network_adapter.last_fault_code = original_last_fault_code
    c_network_adapter.collect_log_by_smbus_task = original_collect_log_by_smbus_task
    c_network_adapter.package_id = original_package_id
    
    -- 如果测试失败，重新抛出错误
    if not ok then
        error(err)
    end
end

-- 测试 Hi182X FaultCode response 函数
function TEST_network_adapter:test_hi182x_fault_code_response()
    local libmgmt_protocol = require 'libmgmt_protocol'
    local original_create_array_parser = libmgmt_protocol.create_array_parser
    
    local fault_code_response = hi1822.smbus().properties.FaultCode.response
    
    -- 测试场景: 使用给定的测试数据 '\x00\x01\x00\x04\x2B\x00\x00\x04'
    -- 数据解析: 每2字节为一个16位值（小端序）
    -- \x00\x01 -> 0x0100 = 256
    -- \x00\x04 -> 0x0400 = 1024
    -- \x2B\x00 -> 0x002B = 43
    -- \x00\x04 -> 0x0400 = 1024
    local test_data = '\x00\x01\x00\x04\x2B\x00\x00\x04'
    
    libmgmt_protocol.create_array_parser = function(template, length)
        -- 验证模板和长度参数
        lu.assertEquals(template, [[<<code:16>>]], "template should match")
        lu.assertEquals(length, 2, "length should be 2")
        
        return function(data)
            -- 模拟解析器返回结果
            -- 数据会被解析为4个元素，每个元素包含code字段
            return {
                {code = 0x0100},  -- \x00\x01 小端序
                {code = 0x0400},  -- \x00\x04 小端序
                {code = 0x002B},  -- \x2B\x00 小端序
                {code = 0x0400}   -- \x00\x04 小端序
            }
        end
    end
    
    local result = fault_code_response(test_data)
    lu.assertNotNil(result, "result should not be nil")
    lu.assertEquals(type(result), "table", "result should be a table")
    lu.assertEquals(#result, 4, "result should have 4 elements")
    lu.assertEquals(result[1], 256, "first fault code should be 256")
    lu.assertEquals(result[2], 1024, "second fault code should be 1024")
    lu.assertEquals(result[3], 43, "third fault code should be 43")
    lu.assertEquals(result[4], 1024, "fourth fault code should be 1024")
    
    -- 测试场景2: parser 返回 nil
    libmgmt_protocol.create_array_parser = function(template, length)
        return function(data)
            return nil
        end
    end
    
    result = fault_code_response(test_data)
    lu.assertNil(result, "result should be nil when parser returns nil")
    
    -- 还原
    libmgmt_protocol.create_array_parser = original_create_array_parser
end

-- BIOS已启动，失败次数未达到阈值
function TEST_network_adapter:test_update_temp_status_on_err_bios_started_below_threshold()
    local adapter = {}
    adapter.smbios_status = 3
    adapter.update_temperature_fail_count = 3
    
    c_network_adapter.update_temp_status_on_err(adapter)
    
    lu.assertEquals(adapter.update_temperature_fail_count, 4, "失败次数应该增加1")
    lu.assertEquals(adapter.TemperatureStatus, 0, "未达到阈值，不应告警")
end

-- BIOS已启动，失败次数达到阈值
function TEST_network_adapter:test_update_temp_status_on_err_bios_started_at_threshold()
    local adapter = {}
    adapter.smbios_status = 3
    adapter.update_temperature_fail_count = 60
    
    c_network_adapter.update_temp_status_on_err(adapter)
    
    lu.assertEquals(adapter.update_temperature_fail_count, 61, "失败次数应该增加1")
    lu.assertEquals(adapter.TemperatureStatus, 1, "达到阈值，应该告警")
end

-- BIOS未启动
function TEST_network_adapter:test_update_temp_status_on_err_bios_not_started()
    local adapter = {}
    adapter.smbios_status = 2
    adapter.update_temperature_fail_count = 7
    
    c_network_adapter.update_temp_status_on_err(adapter)
    
    lu.assertEquals(adapter.update_temperature_fail_count, 0, "失败次数应该重置为0")
    lu.assertEquals(adapter.TemperatureStatus, 2, "BIOS未启动，状态应为2")
end