-- 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 dpu_object = require 'dpu_service.dpu_object'
local dpu_service = require 'dpu_service.dpu_service'
local client = require 'general_hardware.client'
local log_collector = require 'log_collector'
local utils = require 'mc.utils'
local skynet = require 'skynet'
local utils_core = require 'utils.core'
local file_sec = require 'utils.file'
local vos = require 'utils.vos'
local ipmi = require 'ipmi'
local cc = ipmi.types.Cc
local event = require 'infrastructure.event'
local context = require 'mc.context'
local log = require 'mc.logging'
local sd_bus = require 'sd_bus'
local mdb = require 'mc.mdb'
local cjson = require 'cjson'

TestDpuObj = {}

local DPUCARD_INTF<const> = 'bmc.kepler.Systems.DPUCard'
local PCIECARD_INTF<const> = 'bmc.kepler.Systems.PCIeDevices.PCIeCard'
local DPUCARD_FAULT_INTF<const> = 'bmc.kepler.Systems.DPUCard.Fault'
local DPUCARD_METRICS_INTF<const> = 'bmc.kepler.Systems.DPUCard.Metrics'
local DPUCARD_SYSTEM_INTF<const> = 'bmc.kepler.Systems.DPUCard.System'
local TLV_OS_STATUS<const> = 6
local TLV_POWER_STATUS<const> = 1 -- 获取DPU卡电源状态
local FUNCTION_CLASS_FOR_DPU<const> = 11
local PCIE_SENSOR_INVALID_READING<const> = 0x7fff
local SENSOR_INVALID_READING<const> = 0x8000
local SENSOR_NA_READING<const> = 0x4000
local CONFIG_BMC<const> = 0

local board_name = 'IT21SHSU'
local description = 'test description information'
local board_id = 1

local mds_obj = {
    BoardID = 165,
    [DPUCARD_METRICS_INTF] = {
        PowerWatts = 0,
        CPUTemperatureCelsius = 0,
        SFP1TemperatureCelsius = 0,
        SFP2TemperatureCelsius = 0,
        Inlet1TemperatureCelsius = 0,
        Outlet1TemperatureCelsius = 0,
    },
    [DPUCARD_FAULT_INTF] = {
        HeartBeatLoss = 0
    },
    [DPUCARD_INTF] = {
        MCUVersion = '0.0.0',
        LogicVersion = '0.0',
        MultiLogicVersion = {},
        VrdVersion = '0',
        BootSourceOverrideMode = 0,
        BootSourceOverrideEnabled = 0,
        PxeOption = '0',
        UUID = '0',
        M2SlotPresence = 0,
        PfMacInfo = {},
        StorageIpAddr = '0.0.0.0',
        StorageIpVlan = 0,
        SystemLoadedStatus = 0,
        CPLDCount = 2
    },
    [PCIECARD_INTF] = {
        SlotID = 1,
        BoardName = board_name,
        BoardID = board_id,
        Description = description,
        NodeID = 'PCIeCard0',
        FunctionClass = FUNCTION_CLASS_FOR_DPU,
        property_changed = {
            on = function(self, cb)
                cb("SerialNumber", "SN123456")
                cb('FirmwareVersion', '0.0.0')
            end
        }
    },
    [DPUCARD_SYSTEM_INTF] = {
        BIOSVersion = 'NA',
        SecureBootOptionEnabled = 1
    },
    MRCLogLevel = 1,
    UEFILogLevel = 1,
    PowerState = '',
    MPUBusyStatus = 1,
    RefChip = require 'test_data.dpu_ref_chip',
    -- 资产清单接口结果打桩
    AssetType = '',
    AssetName = '',
    InventoryModel = '',
    InventorySerialNumber = nil,
    InventoryFirmwareVersion = nil,
    -- 资产清单接口数据源打桩
    Model = 'Model1',
    SlotID = 1,
    CPUBoardName = 'CPU',
    NICBoardName = 'NIC',
    SerialNumber = "SN123456",
    FirmwareVersion = '0.0.0'
}

function TestDpuObj:setupClass()
    self.dpu_object = dpu_object.new(mds_obj, '010101')
end

function TestDpuObj:teardownClass()
end

function TestDpuObj:test_change_system_loaded_status()
    local obj = {
        fetch_bios_version = function ()
            return 'version'
        end,
        get_slot_id = function ()
            return 'slot_id'
        end,
        fetch_cpu_arch = function ()
            return 'cpu_arch'
        end,
        fetch_cpu_cores = function ()
            return 'cpu_cores'
        end,
        fetch_disk_size = function ()
            return 'disk_size'
        end,
        fetch_mem_size = function ()
            return 'mem_size'
        end
    }
    lu.assertEquals(true, self.dpu_object.mock_change_system_loaded_status(obj, 'name', 3))
end

function TestDpuObj:test_basic_info()
    lu.assertEquals(self.dpu_object.pciecard.BoardName, board_name)
    lu.assertEquals(self.dpu_object:get_description(), description)
    lu.assertEquals(self.dpu_object:get_board_id(), board_id)
end

function TestDpuObj:test_inventory_info()
    lu.assertEquals(mds_obj.AssetType, 'SDI Card')
    lu.assertEquals(mds_obj.AssetName, 'PCIeCard' .. mds_obj[PCIECARD_INTF].SlotID)
    lu.assertEquals(mds_obj.InventoryModel, mds_obj.Model)
    lu.assertEquals(mds_obj.InventorySerialNumber, mds_obj.SerialNumber)
    lu.assertEquals(mds_obj.InventoryFirmwareVersion, mds_obj.FirmwareVersion)
end

function TestDpuObj:test_set_sys_time()
    local sys_time = {
        year = 2024,
        month = 8,
        day = 1,
        hour = 23,
        min = 16,
        sec = 29,
        wday = 5
    }
    local ret = self.dpu_object:set_sys_time(sys_time)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_fetch_mcu_fw_ver()
    self.dpu_object.std_smbus.GetMcuFwVer = function()
        return '1.2.3'
    end
    self.dpu_object:fetch_mcu_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.MCUVersion, '1.2.3')
end

function TestDpuObj:test_update_prop_retry()
    self.dpu_object:update_prop_retry()
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 2)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 2)
end

function TestDpuObj:test_set_property()
    self.dpu_object:set_property('MRCLogLevel', 2)
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 2)
    self.dpu_object:set_property('UEFILogLevel', 3)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 3)
end

function TestDpuObj:test_set_bios_log_prop()
    self.dpu_object:set_bios_log_prop(1, 3)
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 3)
    self.dpu_object:set_bios_log_prop(2, 4)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 4)
end

function TestDpuObj:test_set_bios_log_level()
    local type = 1
    local level = 1
    local ret = self.dpu_object:set_bios_log_level(type, level)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_mrc_log_level()
    self.dpu_object.std_smbus.GetBiosLogLevel = function()
        return {
            level = 1
        }
    end
    local mrc_log_level = self.dpu_object:get_mrc_log_level()
    lu.assertEquals(mrc_log_level.level, 1)
end

function TestDpuObj:test_get_uefi_log_level()
    self.dpu_object.std_smbus.GetBiosLogLevel = function()
        return {
            level = 2
        }
    end
    local uefi_log_level = self.dpu_object:get_uefi_log_level()
    lu.assertEquals(uefi_log_level.level, 2)
end

function TestDpuObj:test_set_sol_switch()
    local sol_switch_type = 7
    local ret = self.dpu_object:set_sol_switch(sol_switch_type)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_sol_switch()
    self.dpu_object.std_smbus.GetSolSwitch = function()
        return {
            type = 5,
            value = 7
        }
    end
    local data = self.dpu_object:get_sol_switch()
    lu.assertEquals(data.type, 5)
    lu.assertEquals(data.value, 7)
end

function TestDpuObj:test_update_pciecard_nodeid()
    self.dpu_object:update_pciecard_nodeid()
    lu.assertEquals(self.dpu_object.pciecard.NodeID, 'PCIeCard1')
end

function TestDpuObj:test_check_heartbeat_loss()
    self.dpu_object:check_heartbeat_loss()
    lu.assertEquals(self.dpu_object.dpucard_fault.HeartBeatLoss, 0)
end

function TestDpuObj:test_fetch_health()
    local ret = self.dpu_object:fetch_health()
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_generate_alarm()
    self.dpu_object:generate_alarm('HeartBeatLoss', 3)
    lu.assertEquals(self.dpu_object.dpucard_fault.HeartBeatLoss, 3)
end

function TestDpuObj:test_fetch_power()
    self.dpu_object.std_smbus.GetPower = function()
        return {
            power = 1800
        }
    end
    self.dpu_object:fetch_power()
    lu.assertEquals(self.dpu_object.dpucard_metrics.PowerWatts, 1800)
    lu.assertEquals(self.dpu_object.dpucard_metrics.ActualPowerWatts, 180)
end

function TestDpuObj:test_fetch_power_fail()
    self.dpu_object.std_smbus.GetPower = function()
        return {
            power = PCIE_SENSOR_INVALID_READING
        }
    end
    self.dpu_object:fetch_power()
    lu.assertEquals(self.dpu_object.dpucard_metrics.PowerWatts, SENSOR_NA_READING)
    lu.assertEquals(self.dpu_object.dpucard_metrics.ActualPowerWatts, SENSOR_INVALID_READING)
end

function TestDpuObj:test_fetch_bios_version()
    self.dpu_object.std_smbus.GetBIOSVersion = function()
        return 'B055'
    end
    self.dpu_object:fetch_bios_version()
    lu.assertEquals(self.dpu_object.dpucard_system.BIOSVersion, 'B055')
end

function TestDpuObj:test_fetch_secure_boot()
    self.dpu_object.is_dpu_card = function()
        return true
    end
    self.dpu_object.std_smbus.GetSecureBoot = function()
        return {
            type = 2,
            secure_boot = 1
        }
    end
    self.dpu_object:fetch_secure_boot()
    lu.assertEquals(self.dpu_object.dpucard_system.SecureBootOptionEnabled, true)
end

function TestDpuObj:test_fetch_mac_address()
    self.dpu_object.std_smbus.GetMacAddress = function()
        return {
            {
                0,
                24,
                'a1b2c3d4e5f6'
            },
            {
                1,
                24,
                'a1b2c3d4e5f6'
            }
        }
    end
    self.dpu_object:fetch_mac_address()
    lu.assertEquals(self.dpu_object.dpucard.PfMacInfo, {{0, 24, "a1b2c3d4e5f6"}, {1, 24, "a1b2c3d4e5f6"}})
end

function TestDpuObj:test_fetch_cpld_fw_ver()
    self.dpu_object:fetch_cpld_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.LogicVersion, '0.0')

    self.dpu_object.std_smbus.GetCpldFwVer = function()
        return '1.02'
    end
    self.dpu_object:fetch_cpld_fw_ver()
    
    lu.assertEquals(self.dpu_object.dpucard.LogicVersion, '1.02')
    lu.assertEquals(self.dpu_object.dpucard.MultiLogicVersion['CPLD1'], '1.02')
    lu.assertEquals(self.dpu_object.dpucard.MultiLogicVersion['CPLD2'], '1.02')
end

function TestDpuObj:test_fetch_vrd_fw_ver()
    self.dpu_object.std_smbus.GetVrdFwVer = function()
        return '1'
    end
    self.dpu_object:fetch_vrd_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.VrdVersion, '1')
end

function TestDpuObj:test_update_fw_ver()
    local cpld1_fw_obj = {}
    local cpld2_fw_obj = {}
    self.dpu_object.cpld_fw_obj = {cpld1_fw_obj, cpld2_fw_obj}
    self.dpu_object.mcu_fw_obj = {}
    self.dpu_object:update_fw_ver()
    lu.assertEquals(self.dpu_object.mcu_fw_obj.Version, '1.2.3')
    lu.assertEquals(self.dpu_object.cpld_fw_obj[1].Version, '1.02')
    lu.assertEquals(self.dpu_object.cpld_fw_obj[2].Version, '1.02')
end

function TestDpuObj:test_get_cpld_manufacture_id()
    self.dpu_object.std_smbus.GetCpldManufacture = function()
        return {
            manufacture_id = 2
        }
    end
    self.dpu_object:get_cpld_manufacture_id()
    lu.assertEquals(self.dpu_object.cpld_manufacture_ids, {2, 2})
end

function TestDpuObj:test_get_mcu_manufacture_id()
    self.dpu_object.std_smbus.GetMcuManufacture = function()
        return 0xA0
    end
    self.dpu_object.mcu_fw_raw = {}
    self.dpu_object:get_mcu_manufacture_id()
    lu.assertEquals(self.dpu_object.mcu_manufacture_id, 0xA0)
end

function TestDpuObj:test_get_slot_id()
    local slot_id = self.dpu_object:get_slot_id()
    lu.assertEquals(slot_id, 1)
    local dpu_service_obj = {
        dpu_objects = {
            [1] = {
                get_slot_id = function()
                    return 2
                end
            }
        }
    }
    slot_id = dpu_service.get_slot_id(dpu_service_obj, 1)
    lu.assertEquals(slot_id, 2)
end

function TestDpuObj:test_query_upgrade_status()
    self.dpu_object.std_smbus.GetMcuUpgradeStatus = function()
        return {
            status = 1
        }
    end
    local status = self.dpu_object:query_upgrade_status()
    lu.assertEquals(status, 1)
end

function TestDpuObj:test_fetch_mcu_reset_time()
    self.dpu_object.std_smbus.GetMcuResetTime = function()
        return 16
    end
    self.dpu_object:fetch_mcu_reset_time()
    lu.assertEquals(self.dpu_object.mcu_reset_time, 16)
end

function TestDpuObj:test_get_dpu_boot_option()
    self.dpu_object.std_smbus.GetBootOption = function()
        return {
            boot_option = 2,
            save_option = 1
        }
    end
    local ret, data = self.dpu_object:get_dpu_boot_option()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.save_option, 1)
    lu.assertEquals(data.boot_option, 2)
end

function TestDpuObj:test_fetch_boot_option()
    self.dpu_object.std_smbus.GetBootOption = function()
        return {
            type1 = 1,
            type2 = 2,
            boot_option = 2,
            save_option = 1
        }
    end
    self.dpu_object:fetch_boot_option()
    lu.assertEquals(self.dpu_object.dpucard.BootSourceOverrideMode, 2)
    lu.assertEquals(self.dpu_object.dpucard.BootSourceOverrideEnabled, 1)
end

function TestDpuObj:test_get_dpu_pxe_option()
    self.dpu_object.std_smbus.GetPxeOption = function()
        return {
            pxe_option = 4
        }
    end
    local ret, data = self.dpu_object:get_dpu_pxe_option()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.pxe_option, 4)
end

function TestDpuObj:test_set_dpu_pxe_option()
    local pxe_option = 5
    local ret = self.dpu_object:set_dpu_pxe_option(pxe_option)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_dpu_uuid()
    self.dpu_object.std_smbus.GetUUID = function()
        return '010203'
    end
    local ret, uuid = self.dpu_object:get_dpu_uuid()
    lu.assertEquals(ret, true)
    lu.assertEquals(uuid, '010203')
end

function TestDpuObj:test_fetch_pxe_option()
    self.dpu_object.std_smbus.GetPxeOption = function()
        return {
            type = 4,
            pxe_option = 4
        }
    end
    self.dpu_object:fetch_pxe_option()
    lu.assertEquals(self.dpu_object.dpucard.PxeOption, 'UefiIpv6')
end

function TestDpuObj:test_set_pxe_option()
    local pxe_opt = 'UefiIpv4AndIpv6'
    local ret = self.dpu_object:set_pxe_option(pxe_opt)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_fetch_dpu_uuid()
    self.dpu_object.std_smbus.GetUUID = function()
        return '010203'
    end
    self.dpu_object:fetch_dpu_uuid()
    lu.assertEquals(self.dpu_object.dpucard.UUID, '010203')
end

function TestDpuObj:test_set_presence()
    self.dpu_object.dpucard.M2SlotPresence = 0
    self.dpu_object:set_presence('M2SlotPresence', 1, 1)
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 1)
    self.dpu_object:set_presence('M2SlotPresence', 2, 0)
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 1)
end

function TestDpuObj:test_fetch_m2_presence()
    self.dpu_object.std_smbus.GetPresence = function()
        return {
            4, 3
        }
    end
    self.dpu_object:fetch_m2_presence()
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 0)
end

function TestDpuObj:test_get_dpu_ip()
    self.dpu_object.std_smbus.GetSdiIp = function()
        return {
            ipv4 = '7392390',
            mask = '7392390',
            vlan = 1
        }
    end
    local ret, data = self.dpu_object:get_dpu_ip()
    local ipv4 = string.format('%d.%d.%d.%d', string.unpack('BBBB',string.pack('I', data.ipv4)))
    local mask = string.format('%d.%d.%d.%d', string.unpack('BBBB',string.pack('I', data.mask)))
    lu.assertEquals(ret, true)
    lu.assertEquals(ipv4, '134.204.112.0')
    lu.assertEquals(mask, '134.204.112.0')
    lu.assertEquals(data.vlan, 1)
end

function TestDpuObj:test_fetch_sdi_ip()
    self.dpu_object.std_smbus.GetSdiIp = function()
        return {
            ipv4 = '7392390',
            vlan = 1
        }
    end
    self.dpu_object:fetch_sdi_ip()
    lu.assertEquals(self.dpu_object.dpucard.StorageIpAddr, '134.204.112.0')
    lu.assertEquals(self.dpu_object.dpucard.StorageIpVlan, 1)
end

function TestDpuObj:test_set_sdi_ipv4()
    local ip_type = 0
    local ipv4 = '192.168.2.1'
    local mask = '192.168.2.2'
    local vlan = 1
    local ret = self.dpu_object:set_sdi_ipv4(ip_type, ipv4, mask, vlan)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_sdi_ipv6()
    local ip_type = 0
    local ipv6 = '0000:0000:0000:0000:0000:0000:0000:0001'
    local prefix_length = 1
    local vlan = 1
    local ret = self.dpu_object:set_sdi_ipv6(ip_type, ipv6, prefix_length, vlan)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_sdi_slot()
    local ret = self.dpu_object:set_sdi_slot(1)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_update_info_task()
    local fork_bak = skynet.fork_loop 
    skynet.fork_loop = function (params, cb)
        return cb
    end
    local bus = {}
    local ipv4 = '192.168.2.1'
    local ipv6 = '2001:0db8:85a3:0000:0000:8a2e:0370:7334'
    local mask = '192.168.2.2'
    local vlan = 1

    dpu_service.get_eth_ipv4 = function(bus)
        return ipv4, mask
    end
    dpu_service.get_eth_vlan = function(bus)
        return vlan
    end
    dpu_service.pre_process_ipv6 = function(ipv6)
        return ipv6
    end
    dpu_service.update_info_task_internal = function(bus)
        return
    end
    dpu_service.update_sdi_slot = function()
        return
    end
    dpu_object.fetch_sdi_ip = function()
        return
    end
    dpu_object.fetch_power = function()
        return
    end
    dpu_object.fetch_boot_option = function()
        return
    end
    dpu_object.fetch_dpu_uuid = function()
        return
    end
    dpu_object.fetch_m2_presence = function()
        return
    end
    dpu_object.fetch_os_status = function()
        return
    end
    dpu_object.fetch_mpu_busy_status = function()
        return
    end
    dpu_object.establish_serial_record_connection = function(bus)
        return
    end

    client.OnIpv4PropertiesChanged = function()
        return true
    end
    client.OnIpv6PropertiesChanged = function()
        return true
    end
    client.OnEthernetInterfacesPropertiesChanged = function()
        return true
    end
    dpu_service:update_info_task(self.dpu_object)
    dpu_service:update_sdi_property(bus, self.dpu_object)
    skynet.fork_loop = fork_bak
end

function TestDpuObj:test_set_sdi_host_os_status()
    local ret = self.dpu_object:set_sdi_host_os_status(0)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_dpu_boot_option()
    local type = 1
    local value = 2
    local ret = self.dpu_object:set_dpu_boot_option(type, value)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_reset_sdi_card()
    local ret = self.dpu_object:reset_sdi_card()
    lu.assertEquals(ret, true)
end
local function construct_ctx()
    local ctx = {}
    ctx.ChanType = 1
    ctx.get_initiator = function()
        return {}
    end

    return ctx
end

function TestDpuObj:test_set_dpu_nmi()
    local req = {}
    req.ManufactureId = 0x0007DB
    dpu_service.get_dpu_obj_by_device_no = function()
        return false
    end
    local _, res = pcall(function()
        return dpu_service:set_dpu_nmi(req, construct_ctx())
    end)
    -- 1、测试 self:get_dpu_obj_by_device_no(req.DeviceNo) 为空
    lu.assertEquals(res.CompletionCode, cc.DataNotAvailable)

    dpu_service.get_dpu_obj_by_device_no = function()
        return {pciecard = {SlotID = 1, Name = "test"}, set_dpu_nmi = function()
            return false, "test"
        end}
    end
    local _, res1 = pcall(function()
        return dpu_service:set_dpu_nmi(req, construct_ctx())
    end)
    -- 2、测试 dpu:set_dpu_nmi() 为空
    lu.assertEquals(res1.CompletionCode, cc.UnspecifiedError)
end

function TestDpuObj:test_set_dpu_nmi_rpc()
    dpu_service.get_dpu_obj_by_slot_id = function()
        return false
    end
    local ok, _ = pcall(function()
        return dpu_service:set_dpu_nmi_rpc(construct_ctx(), 1)
    end)
    -- 1、测试 self:get_dpu_obj_by_slot_id(slot_id) 为空
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_slot_id = function()
        return {pciecard = {SlotID = 1, Name = "test"}, set_dpu_nmi = function()
            return false, "test"
        end}
    end
    local ok1, _ = pcall(function()
        return dpu_service:set_dpu_nmi_rpc(construct_ctx(), 1)
    end)
    -- 2、测试 dpu:set_dpu_nmi() 为空
    lu.assertEquals(ok1, false)
end

function TestDpuObj:test_set_dpu_power_state()
    local tmp_client_ForeachPcieAddrInfoObjects = client.ForeachPcieAddrInfoObjects
    client.ForeachPcieAddrInfoObjects = function(obj, cb)
        cb(obj)
    end

    local ret = self.dpu_object:set_dpu_power_state(1)
    lu.assertEquals(ret, true)

    self.dpu_object.hotplug = {
        HotplugNotify = 0
    }

    local tmp_SlotID = self.dpu_object.pciecard['SlotID']
    local tmp_pciecard = self.dpu_object.pciecard
    self.dpu_object.pciecard = {
        ['SlotID'] = 5
    }

    self.dpu_object.port_id = 16
    self.dpu_object.socket_id = 0

    local tmp_SetSystemAC = self.dpu_object.std_smbus.SetSystemAC
    self.dpu_object.std_smbus.SetSystemAC = function ()
        return true
    end

    local tmp_sleep = skynet.sleep
    skynet.sleep = function(_)
    end

    local tmp_fork_once = skynet.fork_once
    skynet.fork_once = function(cb, ...)
        cb(...)
    end

    ret = self.dpu_object:set_dpu_power_state(1)
    lu.assertEquals(ret, true)
    lu.assertEquals(self.dpu_object.hotplug.HotplugNotify, 4098)

    ret = self.dpu_object:set_dpu_power_state(0)
    lu.assertEquals(ret, true)
    lu.assertEquals(self.dpu_object.hotplug.HotplugNotify, 4097)

    self.dpu_object.port_id = 255
    local tmp_try_send_silk_info = self.dpu_object.try_send_silk_info
    self.dpu_object.try_send_silk_info = function()
        return true
    end
    ret = self.dpu_object:set_dpu_power_state(0)
    lu.assertEquals(ret, true)
    self.dpu_object.port_id = 16
    self.dpu_object.try_send_silk_info = tmp_try_send_silk_info

    local tmp_set_dpu_power_state = self.dpu_object._set_dpu_power_state
    self.dpu_object._set_dpu_power_state = function ()
        return false
    end
    ret = self.dpu_object:set_dpu_power_state(0)
    lu.assertEquals(ret, true)

    self.dpu_object.pciecard['SlotID'] = tmp_SlotID
    self.dpu_object.std_smbus.SetSystemAC = tmp_SetSystemAC
    self.dpu_object._set_dpu_power_state = tmp_set_dpu_power_state
    skynet.fork_once = tmp_fork_once
    skynet.sleep = tmp_sleep
    client.ForeachPcieAddrInfoObjects = tmp_client_ForeachPcieAddrInfoObjects
    self.dpu_object.pciecard = tmp_pciecard
    self.dpu_object.hotplug = nil
end

function TestDpuObj:test_set_dpu_reset_linkage()
    local ret = self.dpu_object:set_dpu_reset_linkage(1)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_reset_linkage()
    self.dpu_object.std_smbus.GetResetLinkage = function()
        return {
            linkage = 1
        }
    end
    local ret, data = self.dpu_object:get_reset_linkage()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.linkage, 1)
end


function TestDpuObj:test_get_mpu_busy_status()
    local ret, status = self.dpu_object:get_mpu_busy_status()
    lu.assertEquals(ret, true)
    lu.assertEquals(status, 1)
end

function TestDpuObj:test_fetch_power_status()
    self.dpu_object.std_smbus.GetSystemStatus = function()
        return {
            status = 1
        }
    end
    self.dpu_object:fetch_power_status()
    lu.assertEquals(self.dpu_object.inner.PowerState, 'On')
end

function TestDpuObj:test_fetch_os_status()
    self.dpu_object.std_smbus.GetSystemStatus = function()
        return {
            status = 1
        }
    end
    self.dpu_object:fetch_os_status()
    lu.assertEquals(self.dpu_object.dpucard.SystemLoadedStatus, 1)
end

function TestDpuObj:test_fetch_power_status_set_property_failed()
    local set_propterty_bak = self.dpu_object.set_property
    self.dpu_object.set_property = function(...)
        return false
    end
    self.dpu_object:fetch_power_status()
    self.dpu_object.set_property = set_propterty_bak
end

function TestDpuObj:test_fetch_power_status_failed()
    local get_dpu_status_bak = self.dpu_object.get_dpu_status
    self.dpu_object.get_dpu_status = function()
        return false
    end
    self.dpu_object:fetch_power_status()
    self.dpu_object.get_dpu_status = get_dpu_status_bak
end

function TestDpuObj:test_get_dpu_power_status()
    self.dpu_object.fetch_power_status = function()
        return
    end
    local sleep = skynet.sleep 
    skynet.sleep = function (times)
        if times == 200 then
            log:raise('break')
        end
    end
    pcall(dpu_service.get_dpu_power_status, dpu_service, self.dpu_object)
    skynet.sleep = sleep
end

function TestDpuObj:test_fetch_mpu_busy_status()
    self.dpu_object.std_smbus.GetMPUBusyStatus = function()
        return 1
    end
    self.dpu_object:fetch_mpu_busy_status()
    lu.assertEquals(self.dpu_object.inner.MPUBusyStatus, 0)
end

function TestDpuObj:test_get_error_code()
    local resp = self.dpu_object:get_error_code()
    lu.assertEquals(resp, {})
end

function TestDpuObj:test_fetch_device_temperature()
    self.dpu_object.std_smbus.GetDeviceTemperature = function()
        return {
            CPU = 26,
            SFP1 = 32,
            SFP2 = 48,
            Inlet1 = 50,
            Outlet1 = 43
        }
    end
    self.dpu_object:fetch_device_temperature()
    lu.assertEquals(self.dpu_object.dpucard_metrics.CPUTemperatureCelsius, 26)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP1TemperatureCelsius, 32)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP2TemperatureCelsius, 48)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Inlet1TemperatureCelsius, 50)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Outlet1TemperatureCelsius, 43)

    self.dpu_object.std_smbus.GetDeviceTemperature = function()
        error()
    end

    self.dpu_object:fetch_device_temperature()
    lu.assertEquals(self.dpu_object.dpucard_metrics.CPUTemperatureCelsius, PCIE_SENSOR_INVALID_READING)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP1TemperatureCelsius, PCIE_SENSOR_INVALID_READING)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP2TemperatureCelsius, PCIE_SENSOR_INVALID_READING)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Inlet1TemperatureCelsius, PCIE_SENSOR_INVALID_READING)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Outlet1TemperatureCelsius, PCIE_SENSOR_INVALID_READING)
end

function TestDpuObj:test_generate_mcu_event()
    local bus = {}
    self.dpu_object:generate_mcu_event(bus, 8358, 'true')
end

function TestDpuObj:test_log_clear()
    log_collector.init()
    log_collector.log_timeout = -1
    -- 用例开始前确保目录干净
    utils.remove_file(log_collector.pcie_card_log_base_dir)
    local file_names = {
        'PCIeCard5(RAID)_RAID',
        'PCIeCard5(SDIV5.0)_SDI5.0',
        'PCIeCard0(SDIV6.0)_SDI5.0',
        'PCIeCard$(SDI5.0)_SDI5.0',
        '_SDI5.0'
    }
    for _, file_name in ipairs(file_names) do
        local log_dir = log_collector.pcie_card_log_base_dir .. file_name
        log_collector.create_dir(log_dir)
        local f_log = file_sec.open_s(log_dir .. '/error_log.bin', 'w+')
        utils.close(f_log, pcall(f_log.write, f_log, 'xxx'))
    end
    -- 目录清理
    log_collector.clear_invaild_log()
    -- 检查目录是否被正确清理
    local expectd_exist_files = {
        'PCIeCard5(RAID)_RAID'
    }
    local files = utils_core.dir(log_collector.pcie_card_log_base_dir)
    lu.assertEquals(#files, #expectd_exist_files)
    table.sort(file_names, function(a, b) return a < b end)
    table.sort(expectd_exist_files, function(a, b) return a < b end)
    for index, _ in pairs(files) do
        lu.assertEquals(files[index], expectd_exist_files[index])
    end

    -- 用例执行结束需要清理残留文件
    utils.remove_file(log_collector.pcie_card_log_base_dir)
end

function TestDpuObj:test_calc_power_read_fail_duration()
    local time = self.dpu_object:calc_power_read_fail_duration()
    lu.assertEquals(time, 0)
end

function TestDpuObj:test_get_latest_alarm_list()
    log.error = function()
        return true
    end
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 1、测试 Event.event_obj为空
    lu.assertEquals(ok, true)
    lu.assertEquals(res, false)

    event.event_obj = {}
    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return false, "error"
        end
    }
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 2、测试 Event.event_obj不为空GetAlarmList_PACKED为空
    lu.assertEquals(ok, true)
    lu.assertEquals(res, false)

    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return true, "error"
        end
    }
    local ok, _ = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 3、测试 GetAlarmList_PACKED返回失败
    lu.assertEquals(ok, true)

    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return true, {EventList = {}}
        end
    }
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 4、测试 GetAlarmList_PACKED返回成功
    lu.assertEquals(ok, true)
    lu.assertEquals(res, true)
end

function TestDpuObj:test_update_serial_disconnect_alarm()
    local bus = {}
    log.debug = function()
        return true
    end
    dpu_object.pciecard = {name = "test"}
    local ok, _ = pcall(function()
        return dpu_object:update_serial_disconnect_alarm(bus)
    end)
    lu.assertEquals(ok, true)
end

function TestDpuObj:test_update_latest_alarm_list()
    local bus = {}
    log.debug = function()
        return true
    end
    dpu_object.pciecard = {name = "test"}
    local ok, _ = pcall(function()
        return dpu_object:update_latest_alarm_list(bus)
    end)
    lu.assertEquals(ok, true)
end

function TestDpuObj:test_get_slot_id()
    local file_path = '/tmp/test.cer'
    local routine_type = 'URI'
    local get_dpu_obj = dpu_service.get_dpu_obj_by_path
    local get_file = vos.get_file_accessible
    dpu_service.get_dpu_obj_by_path = function(...)
        return 1
    end
    vos.get_file_accessible = function(...)
        return true
    end
    local obj = {path = ''}

    -- 测试dpu_service.lua 行1413
    file_sec.check_real_path_s = function(...)
        return -1
    end
    local ok
    ok = pcall(dpu_service.import_public_key, dpu_service, obj, nil, routine_type, file_path)
    lu.assertEquals(ok, false)

    -- 测试dpu_service.lua 行1430
    file_sec.check_real_path_s = function(...)
        return 0
    end
    file_sec.move_file_s = function(...)
        return -1
    end
    ok = pcall(dpu_service.import_public_key, dpu_service, obj, nil, routine_type, file_path)
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_path = get_dpu_obj
    vos.get_file_accessible = get_file
end

function TestDpuObj:test_get_serial_record_connections()
    local c_get_serial_mgmt_objects = dpu_object.get_serial_mgmt_objects
    local c_inner = dpu_object.inner
    local c_exact_conn = dpu_object.exact_conn
    local c_mutex_conn = dpu_object.mutex_conn

    dpu_object.exact_conn = {}
    dpu_object.mutex_conn = {}

    dpu_object.inner = {
        SerialRecordConnect = { {0, 0, 1, 2} }
    }

    local obj = {
        SrcSerial = 1,
        DestSerial = 2,
        Id = "Id1"
    }

    dpu_object.get_serial_mgmt_objects = function()
        return {
            fold = function(_, cb)
                cb(obj)
            end
        }
    end

    dpu_object:get_serial_record_connections()
    lu.assertEquals(dpu_object.exact_conn.Id1.SrcSerial, 1)
    lu.assertEquals(dpu_object.exact_conn.Id1.DestSerial, 2)
    
    obj = {
        SrcSerial = 3,
        DestSerial = 2,
        Id = "Id2"
    }

    dpu_object.get_serial_mgmt_objects = function()
        return {
            fold = function(_, cb)
                cb(obj)
            end
        }
    end

    dpu_object:get_serial_record_connections()
    lu.assertEquals(dpu_object.mutex_conn.Id2.SrcSerial, 3)
    lu.assertEquals(dpu_object.mutex_conn.Id2.DestSerial, 2)

    dpu_object.get_serial_mgmt_objects = c_get_serial_mgmt_objects 
    dpu_object.inner = c_inner
    dpu_object.exact_conn = c_exact_conn
    dpu_object.mutex_conn = c_mutex_conn
end

function TestDpuObj:test_get_extend_card_presence()
    local tmp_card = dpu_object.is_dpu_card
    dpu_object.is_dpu_card = function()
        return false
    end
    local tmp = dpu_object.set_presence
    dpu_object.set_presence = function()
        return
    end
    dpu_object.pciecard = {BoardID = 165}
    local ok, _ = pcall(function()
        return dpu_object:is_sdi_5_0_card()
    end)
    lu.assertEquals(ok, true)
    local ok, _ = pcall(function()
        return dpu_object:get_extend_card_presence()
    end)
    lu.assertEquals(ok, true)
    dpu_object.set_presence = tmp
    dpu_object.is_dpu_card = tmp_card
end

function TestDpuObj:test_set_dpu_LLDP_status()
    dpu_object.dpucard = {
        CPLDCount = 1
    }
    local ret, _ = dpu_object:set_dpu_LLDP_status("")
    lu.assertEquals(ret, false)

    dpu_object.dpucard.CPLDCount = 2
    dpu_object.std_smbus = {
        SetLLDPStatus = function()
            return true
        end
    }
    lu.assertEquals(dpu_object:set_dpu_LLDP_status(""), true)
end

function TestDpuObj:Test_fetch_LLDP_status()
    local mock1 = dpu_object.get_dpu_LLDP_status
    dpu_object.get_dpu_LLDP_status = function()
        return false, {type = 5}
    end
    dpu_object.fetch_LLDP_status_switch = nil
    dpu_object:fetch_LLDP_status()
    lu.assertEquals(dpu_object.fetch_LLDP_status_switch, nil)
    
    dpu_object.fetch_LLDP_status_switch = 111
    dpu_object:fetch_LLDP_status()
    lu.assertEquals(dpu_object.fetch_LLDP_status_switch, false)

    dpu_object.get_dpu_LLDP_status = function()
        return true, {type = 4, status = 6}
    end
    dpu_object:fetch_LLDP_status()
    lu.assertEquals(dpu_object.fetch_LLDP_status_switch, true)
    lu.assertEquals(dpu_object.LLDP_status, nil)

    dpu_object.get_dpu_LLDP_status = function()
        return true, {type = 5, status = 6}
    end
    dpu_object:fetch_LLDP_status()
    lu.assertEquals(dpu_object.fetch_LLDP_status_switch, true)
    lu.assertEquals(dpu_object.LLDP_status, 6)

    dpu_object.get_dpu_LLDP_status = mock1
end

function TestDpuObj:Test_get_dpu_LLDP_status()
    dpu_object.dpucard = {
        CPLDCount = 1
    }

    local ret, _ = dpu_object:get_dpu_LLDP_status()
    lu.assertEquals(ret, false)

    dpu_object.dpucard = {
        CPLDCount = 2
    }
    dpu_object.std_smbus = {
        GetLLDPStatus = function(...)
            return "resp"
        end
    }

    local _, value = dpu_object:get_dpu_LLDP_status()
    lu.assertEquals(value, "resp")
end

function TestDpuObj:Test_get_sdi_cpld_info()

    dpu_object.dpucard = {
        CPLDCount = 1
    }

    dpu_object.std_smbus = {
        GetCpldInfo = function(...)
            return "resp"
        end
    }
    local mock1 = log_collector.get_log_dir
    local mock2 = log_collector.create_dir

    log_collector.get_log_dir = function()
        return "path"
    end
    log_collector.create_dir = function()
        return true
    end
    dpu_object.pciecard = {
        SlotID = 'slotid'
    }
    dpu_object:get_sdi_cpld_info()
    utils.remove_file('pathsmartNic_cpld_infoslotid')

    log_collector.get_log_dir = mock1
    log_collector.create_dir = mock2
end

function TestDpuObj:Test_register_cpld_firmware_info()
    local mock1 =  skynet.fork 
    skynet.fork = function(cb)
        cb()
    end
    dpu_object.get_slot_id = function()
        return 'slotid'
    end
    local mock2 = mdb.get_object
    mdb.get_object = function()
    end
    dpu_object.cpld_fw_id = 'id'
    dpu_object.pciecard = {
        BoardName = 'BoardName'
    }
    dpu_object.dpucard = {
        CPLDCount = 1,
        MultiLogicVersion = {
            'version'
        }
    }
    local mock = utils.retry_func
    utils.retry_func = function()
        return 'rsp'
    end

    dpu_object:register_cpld_firmware_info("bus")
    lu.assertEquals('id', dpu_object.registered_fw_id)

    utils.retry_func = mock
    skynet.fork = mock1
    mdb.get_object = mock2
end

function TestDpuObj:Test_monitor_serial_connection()
    local get_object = mdb.get_object
    mdb.get_object = function ()
        return true
    end
    local dpu_ser = dpu_service.get_instance()
    dpu_ser.monitor_serial_task = false

    local update_serial_disconnect_alarm = self.dpu_object.update_serial_disconnect_alarm
    local general_serial_disconnect_event = self.dpu_object.general_serial_disconnect_event
    self.dpu_object.general_serial_disconnect_event = function ()
    end

    local has_event = true
    self.dpu_object.update_serial_disconnect_alarm = function (dpu_obj)
        dpu_obj.serial_disconnect_alarm = false
    end
    dpu_ser:monitor_serial_connection(self.dpu_object)
    lu.assertEquals(has_event, true)

    self.dpu_object.update_serial_disconnect_alarm = function (dpu_obj)
        dpu_obj.serial_disconnect_alarm = true
    end
    event.generate_event = function (param)
        error()
    end
    dpu_ser:monitor_serial_connection(self.dpu_object)
    lu.assertEquals(has_event, true)


    local generate_event = event.generate_event
    event.generate_event = function (param)
        print(param)
        has_event = false
    end

    dpu_ser:monitor_serial_connection(self.dpu_object)
    lu.assertEquals(has_event, false)

    self.dpu_object.update_serial_disconnect_alarm = update_serial_disconnect_alarm
    self.dpu_object.general_serial_disconnect_event = general_serial_disconnect_event
    event.generate_event = generate_event
    mdb.get_object = get_object
end

function TestDpuObj:Test_clear_task()
    local fork = skynet.fork
    skynet.fork = function (cb)
        cb()
    end
    local dpu_ser = dpu_service.get_instance()
    dpu_ser.ordered_dpu_position_objects = {}
    dpu_ser.dpu_count = 0
    pcall(function()
        dpu_ser:on_add_default_dpu_object(dpu_object, mds_obj, '010102')
    end)
    local dpu_obj = dpu_ser.dpu_objects['010102']
    lu.assertEquals(dpu_obj.clear_task_flag, false)
    pcall(function()
        dpu_ser:on_delete_object('DPUCard', mds_obj, '010102')
    end)
    dpu_obj:update_property_task()
    lu.assertEquals(dpu_obj.clear_task_flag, true)
    skynet.fork = fork
end

function TestDpuObj:Test_general_serial_disconnect_event()
    local original_obj_pciecard = self.dpu_object.pciecard
    local original_check_component_dpucard = self.dpu_object.check_component_dpucard
    self.dpu_object.check_component_dpucard = function()
        return true
    end
    local original_update_serial_disconnect_alarm = self.dpu_object.update_serial_disconnect_alarm
    self.dpu_object.update_serial_disconnect_alarm = function() end
    local original_encode = cjson.encode
    cjson.encode = function() end
    local original_generate_event = event.generate_event
    event.generate_event = function() end
    local bus = {}

    local presence = 2
    self.dpu_object:general_serial_disconnect_event(bus, presence)

    presence = 1
    self.dpu_object.pciecard = nil
    self.dpu_object:general_serial_disconnect_event(bus, presence)

    presence = 1
    self.dpu_object.pciecard = {
        ['DeviceName'] = 'DPU_Card_1',
        ['SlotID'] = 1,
        ['FunctionClass'] = 1
    }
    self.dpu_object.serial_disconnect_alarm = true
    self.dpu_object:general_serial_disconnect_event(bus, presence)

    presence = 0
    self.dpu_object.pciecard = {
        ['DeviceName'] = 'DPU_Card_1',
        ['SlotID'] = 1,
        ['FunctionClass'] = 1
    }
    self.dpu_object.serial_disconnect_alarm = false
    self.dpu_object:general_serial_disconnect_event(bus, presence)

    -- 还原
    self.dpu_object.pciecard = original_obj_pciecard
    self.dpu_object.check_component_dpucard = original_check_component_dpucard
    self.dpu_object.update_serial_disconnect_alarm = original_update_serial_disconnect_alarm
    cjson.encode = original_encode
    event.generate_event = original_generate_event
end

function TestDpuObj:Test_set_hotplug()
    local hotplug_obj = {
        HotplugNotify = 123
    }
    local ok = pcall(function()
        self.dpu_object:set_hotplug(hotplug_obj)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(self.dpu_object.hotplug.HotplugNotify, 123)
    self.dpu_object.hotplug = nil
end

function TestDpuObj:Test_dpu_set_linkage_powerstate()
    local ok

    local tmp_set_dpu_power_state = self.dpu_object.set_dpu_power_state
    self.dpu_object.set_dpu_power_state = function() end

    self.dpu_object.inner.IndependentPowerControlEnabled = true
    ok = pcall(function()
        self.dpu_object:dpu_set_linkage_powerstate(1)
    end)
    lu.assertEquals(ok, true)

    self.dpu_object.inner.IndependentPowerControlEnabled = false
    ok = pcall(function()
        self.dpu_object:dpu_set_linkage_powerstate(0)
    end)
    lu.assertEquals(ok, true)

    local tmp_is_silk_config_ready = self.dpu_object.is_silk_config_ready
    self.dpu_object.is_silk_config_ready = function()
        return false
    end
    ok = pcall(function()
        self.dpu_object:dpu_set_linkage_powerstate(0)
    end)
    lu.assertEquals(ok, true)

    self.dpu_object.is_silk_config_ready = tmp_is_silk_config_ready
    self.dpu_object.set_dpu_power_state = tmp_set_dpu_power_state
end

function TestDpuObj:Test_is_dpu_powerstate_valid()
    local ok = pcall(function()
        self.dpu_object:is_dpu_powerstate_valid(0)
    end)
    lu.assertEquals(ok, true)
    ok = pcall(function()
        self.dpu_object:is_dpu_powerstate_valid(6)
    end)
    lu.assertEquals(ok, true)
end

function TestDpuObj:Test_update_silk_info()
    local tmp_client_ForeachPcieAddrInfoObjects = client.ForeachPcieAddrInfoObjects
    client.ForeachPcieAddrInfoObjects = function(obj, cb)
        cb(obj)
    end
    local ok = pcall(function()
        self.dpu_object:update_silk_info()
    end)
    lu.assertEquals(ok, true)
    client.ForeachPcieAddrInfoObjects = tmp_client_ForeachPcieAddrInfoObjects
end

function TestDpuObj:Test_try_send_silk_info()
    local tmp_port_id = self.dpu_object.port_id
    self.dpu_object.port_id = 255
    local tmp_is_silk_info_ready = self.dpu_object.is_silk_info_ready
    self.dpu_object.is_silk_info_ready = function()
        return false
    end
    local tmp_update_silk_info = self.dpu_object.update_silk_info
    self.dpu_object.update_silk_info = function()
        return
    end
    local ok, ret = pcall(function()
        return self.dpu_object:try_send_silk_info()
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(ret, false)
    self.dpu_object.port_id = tmp_port_id
    self.dpu_object.is_silk_info_ready = tmp_is_silk_info_ready
    self.dpu_object.update_silk_info = tmp_update_silk_info
end