-- 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 bios_enum = require 'domain.bios_firmware.defs'
local package_snapshot = require 'domain.bios_firmware.package.package_snapshot'
local package_builder = require 'domain.bios_firmware.package.package_builder'
local open_db = require 'bios.db'
local ipmb_channel = require 'domain.bios_firmware.package.channel.ipmb_channel'
local executors = require 'domain.bios_firmware.package.executors.upgrade_executor'
local imu_communicate = require 'domain.transport.imu_communicate'
local package = require 'domain.bios_firmware.package.package'
local component_version = require 'domain.mapping.component_version'
local bios_factory = require 'factory.bios_factory'
local bios_service = require 'service.bios_service'
local json = require 'cjson'
local component_upgrade = require 'domain.transport.component_upgrade'
local firmware = require 'interface.mdb.firmware_multihost'
local pfr_service = require 'service.pfr_service'
local upgrade_service = require 'service.upgrade_service'
local context = require 'pojo.context'
local bios_object = require 'pojo.bios_object_mutihost'
local msg = require 'bios.ipmi.ipmi_message'
local obj_def = require 'macros.object_def'
local prop_def = require "macros.property_def"
local object_service = require 'service.object_service'
local mdb_service = require 'mc.mdb.mdb_service'
local fructl_handler = require 'infrastructure.fructl'
local privilege_check = require 'infrastructure.privilege_check'
local test_common = require 'test_common.utils'
local bus = test_common.dbus_launch('../.dbus', nil, true)
local component_collection = require 'domain.mapping.component_manager_collection'
local firmware_collection = require 'interface.mdb.firmware_multihost_collection'
local skynet_queue = require 'skynet.queue'
local client = require 'bios.client'

local client_mock = {}

local mdb = require 'mc.mdb'

TestMultihost = {}

local function construct_db()
    local ok, datas = pcall(require, 'bios.datas')
    if not ok then
        -- 如果没有datas配置，证明当前组件不需要datas，仅打开数据库
        datas = nil
    end
    local db = open_db(':memory:', datas)
    return db
end

function TestMultihost:test_bin_parser()
    local upgrade_component_executor = executors[bios_enum.UpgradeSteps.UpgradeComponent].new({})
    local ok = pcall(function()
        upgrade_component_executor:send_confirm({
            Global = {
                SystemId = 1,
                Period = bios_enum.PackagePeriod.Period3,
                UpgradeMode = bios_enum.UpgradeMode.Hot
            }
        })
    end)
    lu.assertEquals(ok, false)

    local imu_cmd = imu_communicate.get_instance()
    imu_cmd.query_firmware_process_status = function()
        return {Status = 0x00}
    end
    imu_cmd:query_firmware_process_status()
    local final_executor = executors[bios_enum.UpgradeSteps.WaitUpgradeFinish].new({})
    final_executor:execute({
        Global = {
            SystemId = 1
        }
    }, {
        execute = function()
            return
        end
    })
    lu.assertNotEquals(final_executor, nil)
end

function TestMultihost:test_update_bios_firmware_date()
    local mock2 = firmware.db
    firmware.db = {
        BiosFwInfoTable = function(...)
            return {
                ReleaseDate = 'ReleaseDate',
                Version = 'Version',
                save = function() end
            }
         end
    }
    firmware.sysid = 1

    local mock1 = mdb.get_cached_object
    mdb.get_cached_object = function(...)
        return {
            ReleaseDate = ''
        }
    end

    local ret, _ = pcall(firmware.update_bios_firmware_date, firmware, '10/05/2000')
    mdb.get_cached_object = mock1
    firmware.db = mock2
    lu.assertEquals(ret, true)
end

function TestMultihost:test_package()
    local package_instance = package.new({}, bios_enum.PackageType.Normal)
    local upgrade_info = {
        db = construct_db(),
        upgrade_path = '',
        upgrade_mode = bios_enum.UpgradeMode.Cold
    }
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local ok = pcall(function()
        package_instance:_start(upgrade_info, snapshot)
    end)
    lu.assertEquals(ok, false)
end

local function get_bios_obj()
    return {
        get_system_id = function()
            return 1
        end,
        register_mdb_objects = function ()
        end,
        unregister_mdb_objects = function ()
        end,
        PcieCardBDF = '',
        PcieDiskBDF = '',
        OCPCardBDF = '',
        Slot = 1,
        Version = '10.66'
    }
end

function TestMultihost:test_package_snapshot()
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local ok = pcall(function()
        snapshot:activate(0x01)
    end)
    lu.assertEquals(ok, false)

    local bios_ser = bios_service.get_instance(bus, construct_db())
    bios_ser:add_object(get_bios_obj())
    bios_factory.register_bean('bios_service', bios_ser)
    local component_version_instance = component_version.new(1)
    local version = {
        PatchVersion = 'HP001'
    }
    component_version_instance.data = json.encode(version)
    bios_ser.update_patch_version = function()
        return
    end
    ok = pcall(function()
        component_version_instance:update_cfg_version()
    end)
    lu.assertEquals(ok, true)
end

function TestMultihost:test_a_mapping()
    local imu_cmd = imu_communicate.get_instance()
    pcall(imu_cmd.activate, imu_cmd, 0x01, 1)
    pcall(imu_cmd.write_firmware, imu_cmd, '', 1)
    pcall(imu_cmd.query_firmware_process_status, imu_cmd, 1)
    local compt_upgrade = component_upgrade.new()
    compt_upgrade.system_id = 1
    compt_upgrade.component_data = {
        get_type = function()
            return 1
        end,
        get_size = function()
            return 1
        end,
        get_crc = function()
            return 1
        end,
        get_id = function()
            return 1
        end,
        get_offset = function()
            return 1
        end,
        fetch_data = function()
            return ''
        end
    }
    pcall(compt_upgrade.write, compt_upgrade, '', 1)
    pcall(compt_upgrade.write_prepare, compt_upgrade)
    pcall(compt_upgrade.write_data, compt_upgrade)
    pcall(compt_upgrade.write_finish, compt_upgrade)
end

function TestMultihost:test_firmware()
    local firmware_instance = firmware.new({}, construct_db(), 1)
    pcall(firmware_instance.get_activate_mode, firmware_instance)
    pcall(firmware_instance.clear_activate_mode, firmware_instance)
end

function TestMultihost:test_pfr_service()
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local pfr_service_instance = pfr_service.new(construct_db())
    pfr_service_instance:cache_hpm_after_activate(snapshot)
    pfr_service_instance:can_effective(1)
    pfr_service_instance.can_effective = function()
        return true
    end
    pfr_service_instance:try_get_version(snapshot)
    pfr_service_instance:wait_effective(snapshot)
    pfr_service_instance:backup_hpm(2)
    pfr_service_instance:remove_tmp_hpm(2)
end

local function construct_ctx()
    local ctx = {}
    ctx.ChanType = 1
    ctx.get_initiator = function()
        return {}
    end

    return ctx
end

local function test_activate_components(upgrade_service_instance, snapshot)
    local bios_ser = bios_service.get_instance(bus, construct_db())
    bios_ser.is_multihost = function()
        return true
    end
    upgrade_service_instance:activate_components(construct_ctx(), {'IMU'})
    bios_ser.is_multihost = function()
        return false
    end

    local ok = pcall(function()
        upgrade_service_instance:activate_components(construct_ctx(), {'IMU'})
    end)
    lu.assertEquals(ok, false)

    ok = pcall(function()
        upgrade_service_instance:activate_component(construct_ctx(), "IMU", snapshot)
    end)
    lu.assertEquals(ok, false)

    bios_ser:set_prop('SystemStartupState', 254, 1)
    ok = pcall(function()
        upgrade_service_instance:activate_component(construct_ctx(), {"IMU"}, snapshot)
    end)
    lu.assertEquals(ok, false)
    bios_ser.get_componet_bitmap_list = function()
        return {
            [1] = {
                BitMap = 0x01,
                Name = 'IMU'
            }
        }
    end
    ok, res = pcall(function()
        upgrade_service_instance:activate_component(construct_ctx(), {"IMU"}, snapshot)
    end)
    lu.assertEquals(ok, false)
    snapshot.activate = function()
        return true
    end
    local c_get_prop = bios_ser.get_prop
    bios_ser.get_prop = function()
        return 254
    end
    ok = pcall(function()
        upgrade_service_instance:activate_component(construct_ctx(), {"IMU"}, snapshot)
    end)
    lu.assertEquals(ok, true)
    ok = pcall(function()
        upgrade_service_instance:activate_multihost_components(construct_ctx(), {"IMU"})
    end)
    lu.assertEquals(ok, true)
    upgrade_service_instance.hpm_upgrade.upgrade_mode = bios_enum.UpgradeMode.Hot
    ok = pcall(function()
        upgrade_service_instance:checkout_before_activate(construct_ctx(), 1)
    end)
    lu.assertEquals(ok, false)
    bios_ser.get_prop = c_get_prop
end

local function test_upgrade(upgrade_service_instance, snapshot)
    upgrade_service_instance.hpm_upgrade:hot_upgrade_info({
        system_id = 1,
        firmware_type = 1,
        cfg_path = '',
        hpm_path = '',
        context = construct_ctx()
    }, 1)

    local ok = pcall(function()
        upgrade_service_instance.hpm_upgrade:online_force_upgrade_info({
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx()
        }, 1)
    end)
    lu.assertEquals(ok, false)

    upgrade_service_instance.hpm_upgrade:build_upgrade_mode({
        system_id = 1,
        firmware_type = 1,
        cfg_path = '',
        hpm_path = '',
        context = construct_ctx(),
        ActivateComponents = {"IMU"}
    }, 1)
end

local function test_patch(upgrade_service_instance, snapshot)
    upgrade_service_instance.hpm_upgrade.package =
        package_builder.new():build_with_snapshot(snapshot)
    upgrade_service_instance.hpm_upgrade:check_patch_version(1)
    upgrade_service_instance.hpm_upgrade:build_info({
        system_id = 1,
        firmware_type = 1,
        cfg_path = '',
        hpm_path = '',
        context = construct_ctx(),
        spi_rate = 10,
        Multihost = true
    }, 1)

    local ok = pcall(function()
        upgrade_service_instance.hpm_upgrade:_process_multihost(snapshot, {
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx(),
            spi_rate = 10,
            Multihost = true
        })
    end)
    lu.assertEquals(ok, false)

    ok, res = pcall(function()
        upgrade_service_instance.hpm_upgrade:_process_multihost(snapshot, {
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx(),
            spi_rate = 10,
            Multihost = true,
            ActivateComponents = {"IMU"}
        })
    end)
    lu.assertEquals(ok, false)
    
    local bios_ser = bios_service.get_instance(bus, construct_db())
    bios_ser:get_bios_id()
    local c_get_prop = bios_ser.get_prop
    local c_set_prop = bios_ser.set_prop
    bios_ser.get_prop = function()
        return 1
    end
    bios_ser.set_prop = function()
        return
    end
    ok, res = pcall(function()
        upgrade_service_instance.hpm_upgrade:_process_multihost(snapshot, {
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx(),
            spi_rate = 10,
            Multihost = false,
        })
    end)
    lu.assertEquals(ok, false)
    bios_ser.get_prop = c_get_prop
    bios_ser.set_prop = c_set_prop
end

local function test_process(upgrade_service_instance, snapshot)
    local ok = pcall(function()
        upgrade_service_instance.hpm_upgrade:_process({
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx(),
            spi_rate = 10,
            Multihost = false,
            Snapshots = {snapshot}
        })
    end)

    lu.assertEquals(ok, false)
    ok = pcall(function()
        upgrade_service_instance.hpm_upgrade:_process({
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            context = construct_ctx(),
            spi_rate = 10,
            Multihost = true,
            ActivateComponents = {"IMU"},
            Snapshots = {snapshot}
        })
    end)
    lu.assertEquals(ok, false)
end

function TestMultihost:test_multihost_reset_bios()
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    snapshot.SystemId = 1
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    upgrade_service_instance:add_snapshot(1, snapshot)
    upgrade_service_instance.fructl_handler = fructl_handler
    fructl_handler.get_power_status = function (system_id)
        return 'OFF'
    end
    local get_service = bios_factory.get_service
    upgrade_service_instance.bios_factory = bios_factory
    bios_factory.get_service = function(service_name)
        local obj_util = {reset_bios = function(ctx, system_id)
            return 0
        end}
        return obj_util
    end
    local ok = pcall(function()
        upgrade_service_instance.hpm_upgrade:multihost_reset_bios({
            system_id = 1,
            firmware_type = 1,
            cfg_path = '',
            hpm_path = '',
            Snapshots = upgrade_service_instance.package_snapshot_collection
        })
    end)
    bios_factory.get_service = get_service
    lu.assertEquals(ok, true)
end

function TestMultihost:test_upgrade_force_service()
    -- 全檫升级测试
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    local phrase = 'prepare'
    local cfg = {system_id = 1, para = {ForceUpgrade = "force"}}
    upgrade_service_instance.hpm_upgrade.upgrade = function(phrase, cfg, snapshots)
            
    end
    upgrade_service_instance.effective_upgrade = {}
    upgrade_service_instance:add_snapshot(1, snapshot)
    upgrade_service_instance:upgrade_hpm(phrase, cfg)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.PrepareStart)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.PrepareFinish)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.ProcessStart)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.ProcessFinish)
    upgrade_service_instance.hpm_upgrade:finish({
        system_id = 1,
        firmware_type = 1,
        cfg_path = '',
        hpm_path = '',
        context = construct_ctx()
    })
    test_activate_components(upgrade_service_instance, snapshot)
    test_patch(upgrade_service_instance, snapshot)
    test_process(upgrade_service_instance, snapshot)
end

function TestMultihost:test_upgrade_service()
    -- 原有测试逻辑。不走全檫升级
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    local phrase = 'prepare'
    local cfg = {system_id = 1, para = {ForceUpgrade = "noforce"}}
    upgrade_service_instance:add_snapshot(1, snapshot)
    upgrade_service_instance:upgrade_hpm(phrase, cfg)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.PrepareStart)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.PrepareFinish)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.ProcessStart)
    upgrade_service_instance.hpm_upgrade:_set_state_machine(bios_enum.UpgradeEvent.ProcessFinish)
    upgrade_service_instance.hpm_upgrade:finish({
        system_id = 1,
        firmware_type = 1,
        cfg_path = '',
        hpm_path = '',
        context = construct_ctx()
    })
    test_activate_components(upgrade_service_instance, snapshot)
    test_patch(upgrade_service_instance, snapshot)
    test_process(upgrade_service_instance, snapshot)
end

function TestMultihost:test_check_multiflash_and_effective_process()
    local snapshot = package_snapshot.new(construct_db(), {ActivatedStatus = 1}, 1)
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    upgrade_service_instance.effective_upgrade.package = package_builder.new():build_with_snapshot(snapshot)
    upgrade_service_instance.effective_upgrade.package.process = function ()
        return
    end
    local bios_ser = bios_service.get_instance(bus, construct_db())
    local c_get_prop = bios_ser.get_prop
    local c_set_prop = bios_ser.set_prop
    local bios_obj_flash = {
        FlashNum = 2,
        FlashChannelIds = {},
        FlashChannel = 0
    }
    bios_ser.get_prop = function(_, prop, sys_id)
        return bios_obj_flash[prop]
    end
    bios_ser.set_prop = function(_, prop, value, system_id)
        bios_obj_flash[prop] = value
    end
    local upgrade_info = {hpm_delete_flag = true}
    local ok = pcall(function ()
        upgrade_service_instance.effective_upgrade:check_multiflash_and_effective_process(upgrade_info, snapshot)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(bios_obj_flash['FlashChannel'], 1)

    bios_obj_flash['FlashNum'] = 1
    bios_obj_flash['FlashChannelIds'] = {3}

    ok = pcall(function ()
        upgrade_service_instance.effective_upgrade:check_multiflash_and_effective_process(upgrade_info, snapshot)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(bios_obj_flash['FlashChannel'], 3)
    bios_ser.get_prop = c_get_prop
    bios_ser.set_prop = c_set_prop
end

function TestMultihost:test_get_object_list()
    object_service.bus = {type = 'signal', path_namespace = '/bmc/kepler/Systems/1/Memory',
        interface = 'org.freedesktop.DBus.ObjectManager', member = 'InterfacesAdded'}
    mdb_service.get_sub_paths = function()
        rsp = {}
        rsp.SubPaths = {"test"}
        return rsp
    end
    object_service.get_object = function()
        return {}
    end
    object_service.bus.match = function(self, obj, cb)
        cb(msg)
        return true
    end
    local ok, _ = pcall(function()
        return object_service:get_object_list('object_service', "test", "test")
    end)
    lu.assertEquals(ok, true)
end

function TestMultihost:test_inner_upgrade()
    local snapshot = package_snapshot.new(construct_db(), {}, 1)
    snapshot.get_package_type = function ()
        return bios_enum.PackageType.Patch
    end
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    upgrade_service_instance:add_snapshot(1, snapshot)
    local ok, _ = pcall(function()
        return upgrade_service_instance:inner_upgrade('')
    end)
    lu.assertEquals(ok, true)
end

function TestMultihost:test_check_sp_state()
    local get_service = bios_factory.get_service
    local ctx = {
        Privilege = '1',
        UserName = '<host sms>'
    }
    bios_factory.get_service = function ()
        local bios_ser = {}
        bios_ser.is_multihost = function()
            return true
        end
        return bios_ser
    end
    local _, ret = pcall(function()
        return privilege_check.check(ctx)
    end)
    lu.assertEquals(ret, true)
    bios_factory.get_service = function ()
        return nil
    end
    _, ret = pcall(function()
        return privilege_check.check(ctx)
    end)
    lu.assertEquals(ret, false)
    bios_factory.get_service = get_service
end


function TestMultihost:test_get_disk_silk_config()
    local ok, _ = pcall(function()
        return bios_service:get_disk_silk_config()
    end)
    lu.assertEquals(ok, true)
    local props = {}
    props[obj_def.PCIE_ADDR_INFO_CONTROLLER_TYPE] = {value = function ()
        return 2
    end}
    props[obj_def.PCIE_ADDR_INFO_SOCKET_ID] = {value = function ()
        return 1
    end}
    props[obj_def.PCIE_ADDR_INFO_DEVICE_PORT_ID] = {value = function ()
        return 1
    end}
    props[obj_def.PCIE_ADDR_INFO_SLOT_ID] = {value = function ()
        return 1
    end}
    props[obj_def.PCIE_ADDR_INFO_BUS] = {value = function ()
        return 1
    end}
    props[obj_def.PCIE_ADDR_INFO_DEVICE] = {value = function ()
        return 1
    end}
    props[obj_def.PCIE_ADDR_INFO_FUNCTION] = {value = function ()
        return 1
    end}
    local get_service = bios_factory.get_service
    bios_factory.get_service = function()
        local obj_util = {get_object_list = function(name, path, int)
            return {
                props
            }
        end}
        return obj_util
    end
    local ok, _ = pcall(function()
        return bios_service:get_disk_silk_config()
    end)
    bios_factory.get_service = get_service
    lu.assertEquals(ok, false)

    local ok, _ = pcall(function()
        return bios_object:cache_pcie_info()
    end)
    lu.assertEquals(ok, true)
end

function TestMultihost:test_wait_until_power_off()
    local get_power_status = fructl_handler.get_power_status
    fructl_handler.get_power_status = function ()
        return 'OFF'
    end
    local pfr_service_instance = pfr_service.new(construct_db())
    local ok = pfr_service_instance:wait_until_power_off()
    lu.assertEquals(ok, true)
    fructl_handler.get_power_status = get_power_status
end

function TestMultihost:test_get_power_restart_cause()
    local get_fructl_obj = fructl_handler.get_fructl_obj
    fructl_handler.get_fructl_obj = function ()
        return nil
    end
    local ret = fructl_handler.get_power_restart_cause()
    lu.assertEquals(nil, ret)

    fructl_handler.get_fructl_obj = function ()
        return {CurrentRestartCause = 'CurrentRestartCause'}
    end
    ret = fructl_handler.get_power_restart_cause()
    lu.assertEquals('CurrentRestartCause', ret)

    fructl_handler.get_fructl_obj = get_fructl_obj
end

function TestMultihost:test_get_perserve_config_enabled()
    local snapshot_obj = {
        ActivatedStatus = 1,
        PreserveConfigEnabled = true
    }
    local snapshot = package_snapshot.new(construct_db(), snapshot_obj, 1)
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})

    local cfg = {
        Snapshots = {}
    }
    local preserve_config_enabled = upgrade_service_instance.hpm_upgrade:get_preserve_config_enabled(cfg)
    lu.assertEquals(preserve_config_enabled, true)

    cfg = {
        Snapshots = {snapshot}
    }
    preserve_config_enabled = upgrade_service_instance.hpm_upgrade:get_preserve_config_enabled(cfg)
    lu.assertEquals(preserve_config_enabled, true)

    snapshot_obj.PreserveConfigEnabled = false
    preserve_config_enabled = upgrade_service_instance.hpm_upgrade:get_preserve_config_enabled(cfg)
    lu.assertEquals(preserve_config_enabled, false)

    snapshot_obj.PreserveConfigEnabled = nil
    preserve_config_enabled = upgrade_service_instance.hpm_upgrade:get_preserve_config_enabled(cfg)
    lu.assertEquals(preserve_config_enabled, true)
end

function TestMultihost:test_is_need_reset_bios()
    local upgrade_service_instance = upgrade_service.new(construct_db(), {})
    upgrade_service_instance.hpm_upgrade.para = {
        RestoreFactorySettings = 'true'
    }
    local is_need_reset_bios = upgrade_service_instance.hpm_upgrade:is_need_reset_bios(true)
    lu.assertEquals(is_need_reset_bios, true)
    is_need_reset_bios = upgrade_service_instance.hpm_upgrade:is_need_reset_bios(false)
    lu.assertEquals(is_need_reset_bios, true)

    upgrade_service_instance.effective_upgrade.para = {}
    is_need_reset_bios = upgrade_service_instance.effective_upgrade:is_need_reset_bios(true)
    lu.assertEquals(is_need_reset_bios, false)
    is_need_reset_bios = upgrade_service_instance.effective_upgrade:is_need_reset_bios(false)
    lu.assertEquals(is_need_reset_bios, true)
end

local function cache_cpu_silk_info_mock()
    client_mock.OnCPUInterfacesAdded = client.OnCPUInterfacesAdded
    client_mock.OnCPUInterfacesRemoved = client.OnCPUInterfacesRemoved
    client_mock.ForeachCPUObjects = client.ForeachCPUObjects
    client.OnCPUInterfacesAdded = function (_, func)
        local path = '/bmc/kepler/Systems/1/CPU/CPU1'
        local props = {
            LogicalId = {
                value = function ()
                    return 0
                end
            },
            PhysicalId = {
                value = function ()
                    return 1
                end
            },
            DeviceLocator = {
                value = function ()
                    return 'aaa'
                end
            },
        }
        func(_, path, props)
        path = '/bmc/kepler/Systems/1/CPU/CPU2'
        props = {
            LogicalId = {
                value = function ()
                    return 1
                end
            },
            PhysicalId = {
                value = function ()
                    return 2
                end
            },
            DeviceLocator = {
                value = function ()
                    return 'bbb'
                end
            },
        }
        func(_, path, props)
    end
    client.OnCPUInterfacesRemoved = function (_, func)
        local path = '/bmc/kepler/Systems/1/CPU/CPU2'
        func(_, path)
    end
    client.ForeachCPUObjects = function (_, func)
        local objs = {
            {path = '/bmc/kepler/Systems/1/CPU/CPU3', LogicalId = 2, PhysicalId = 3, DeviceLocator = 'ccc'},
            {path = '/bmc/kepler/Systems/1/CPU/CPU4', LogicalId = 3, PhysicalId = 4, DeviceLocator = 'ddd'}
        }
        for _, obj in pairs(objs) do
            func(obj)
        end
    end
end

local function recover_cache_cpu_silk_info_mock()
    client.OnCPUInterfacesAdded = client_mock.OnCPUInterfacesAdded
    client.OnCPUInterfacesRemoved = client_mock.OnCPUInterfacesRemoved
    client.ForeachCPUObjects = client_mock.ForeachCPUObjects
end

local function cache_pcie_silk_info_mock()
    client_mock.OnPcieAddrInfoInterfacesAdded = client.OnPcieAddrInfoInterfacesAdded
    client_mock.OnPcieAddrInfoPropertiesChanged = client.OnPcieAddrInfoPropertiesChanged
    client_mock.OnPcieAddrInfoInterfacesRemoved = client.OnPcieAddrInfoInterfacesRemoved
    
    client.OnPcieAddrInfoInterfacesAdded = function (_, func)
        local path = '/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo1'
        local obj = {Name = 'PcieAddrInfo1'}
        func(_, path, obj)
        path = '/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo2'
        obj = {Name = 'PcieAddrInfo2'}
        func(_, path, obj)
    end
    client.OnPcieAddrInfoPropertiesChanged = function (_, func)
        local objs = {
            {
                values = {
                    ReadyToEnumerate = {value = function ()
                        return false
                    end}
                },
                path = '/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo3'
            },
            {
                values = {
                    ReadyToEnumerate = {value = function ()
                        return true
                    end}
                },
                path = '/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo4'
            }
        }
        local interfaces = 'bmc.kepler.Systems.PcieAddrInfo'
        for _, obj in pairs(objs) do
            func(obj.values, obj.path, interfaces)
        end
    end
    client.OnPcieAddrInfoInterfacesRemoved = function (_, func)
        local path = '/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo2'
        func(_, path)
    end
end

local function recover_cache_pcie_silk_info_mock()
    client.OnPcieAddrInfoInterfacesAdded = client_mock.OnPcieAddrInfoInterfacesAdded
    client.OnPcieAddrInfoPropertiesChanged = client_mock.OnPcieAddrInfoPropertiesChanged
    client.OnPcieAddrInfoInterfacesRemoved = client_mock.OnPcieAddrInfoInterfacesRemoved
end

local function test_cache_cpu_silk_info(bios_obj)
    local cpu_info = bios_obj:get_cpu_info()
    local cpu_info_result = {
        CPU1 = {LogicalSocketId = 0, PhysicalSocketId = 1, DeviceLocator = 'aaa', Silk = "CPU1"},
        CPU3 = {LogicalSocketId = 2, PhysicalSocketId = 3, DeviceLocator = 'ccc', Silk = "CPU3"},
        CPU4 = {LogicalSocketId = 3, PhysicalSocketId = 4, DeviceLocator = 'ddd', Silk = "CPU4"}
    }
    for _, cpu_info_signal_obj in pairs(cpu_info) do
        lu.assertEquals(cpu_info_signal_obj, cpu_info_result[cpu_info_signal_obj.Silk])
    end
end

local function test_cache_pcie_silk_info(bios_obj)
    local pcie_info = bios_obj:get_pcie_info()
    local pcie_info_result = {
        ['/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo1'] = {Name = 'PcieAddrInfo1'},
        ['/bmc/kepler/Systems/1/PCIeDevices/PcieAddrInfo4'] = {Name = 'PcieAddrInfo4'}
    }
    lu.assertEquals(pcie_info, pcie_info_result)

    local sas_info = bios_obj:get_sas_info()
    lu.assertEquals(sas_info, pcie_info_result)
end

function TestMultihost:test_cache_silk_info()
    local obj = {
        get_system_id = function ()
            return 1
        end
    }
    local get_component_mock = component_collection.get_component
    component_collection.get_component = function ()
        return {}
    end
    local get_firmware_mock  = firmware_collection.get_firmware
    firmware_collection.get_firmware = function ()
        return {}
    end
    local get_service_mock = bios_factory.get_service
    bios_factory.get_service = function ()
        local service = {}
        service.get_object = function (_, service_name, path, interface)
            return {
                Name = string.match(path, '([^/]+)$')
            }
        end
        service.get_object_list = function ()
            return {}
        end
        service.is_multihost = function ()
            return false
        end
        return service
    end

    local skynet_queue_mock = skynet_queue
    skynet_queue = function ()
        local f = function (func)
            func()
        end
        return f
    end

    cache_cpu_silk_info_mock()
    cache_pcie_silk_info_mock()

    local bios_obj = bios_object.new(obj, construct_db(), bus, '0101')
    bios_obj:cache_cpu_silk_info()
    bios_obj:cache_pcie_silk_info()
    test_cache_cpu_silk_info(bios_obj)
    test_cache_pcie_silk_info(bios_obj)

    skynet_queue = skynet_queue_mock
    recover_cache_cpu_silk_info_mock()
    recover_cache_pcie_silk_info_mock()
    component_collection.get_component = get_component_mock
    firmware_collection.get_firmware = get_firmware_mock
    bios_factory.get_service = get_service_mock
end