-- 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 defs = require 'unit_manager.class.logic_fw.comm_defs'
local cmn = require 'common'
local fw_mgmt_client = require 'general_hardware.client'
local chip = require 'unit_manager.class.logic_fw.cpld_chip'
local drivers = require 'unit_manager.class.logic_fw.upgrade.drivers_api'
local fw_upgrade = require 'unit_manager.class.logic_fw.upgrade.fw_upgrade'
local valid = require 'unit_manager.class.logic_fw.upgrade.valid'
local factory = require 'factory'
local utils = require 'mc.utils'
local file_sec = require 'utils.file'

local FW_VERSION <const> = '2.3.3'
local SYSTEM_ID <const> = 1
local project_dir = os.getenv('PROJECT_DIR')
local cfg_mutil_cold_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_mutil_cold.cfg'
local cfg_mutil_hot_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_mutil_hot.cfg'
local cfg_single_cold_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_single_cold.cfg'
local cfg_i2c_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_i2c.cfg'
local file_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/Firmware1'
local cpld_01 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld01.svf'
local valid_01 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid01.svf'
local valid_02 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid02.svf'
local cpld_vme <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld.vme'
local valid_vme <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid.vme'
local bin <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld01.bin'
local release_path_ret
local fw_version_ret
local prepare_ret

local signal = {
    fw_list = {
        [1] = {
            csr = {
                DefaultRoute   = 0,
                FirmwareRoute  = 4,
                I2CUpgradeChip = {},
                Manufacturer   = "Huawei",
                Name           = "BCU_CPLD1",
                SmcChip        = {},
                SoftwareId     = "CPLD-BC83AMDA",
                UId            = "00000001020302031825",
                UpgradeChip    = {}
            },
            uid = "00000001020302031825",
            name = "CpuBoard1 CPLD1",
            system_id = SYSTEM_ID,
            component_id = 24,
            component_id_ex = 1,
            get_fw_version = function()
                return FW_VERSION
            end,
            set_lock_status = function()
            end,
            i2c_update_chip_lock = function()
                return true, 0
            end,
            i2c_update_chip_unlock = function()
                return true, 0
            end,
            update_chip_lock = function()
                return true, 0
            end,
            update_chip_unlock = function()
                return true, 0
            end,
            switch_to_firmware_route = function()
            end,
            switch_to_default_route = function()
            end,
            chip_info = {
                SetBypassMode = function()
                    return
                end
            },
            update_chip = {
                -- 无法真正写入数据，打桩
                Write = function()
                end
            },
            smc_chip = {
                Read = function()
                    return '\x03\x00\x01' -- i2c模式+anlu
                end
            },
            i2c_upgrade_chip = {
                Write = function()
                end
            }
        }
    },
    db = {
        CpldValidating = function()
            return {
                ValidatingCpldFlag = 1,
                save = function()
                end
            }
        end
    }
}

TestLogicalFwUpgrade = {}

function TestLogicalFwUpgrade:setupClass()
    -- 公共函数打桩
    cmn.skynet.sleep = function()
    end
    -- 框架提供的接口执行异常，打桩为os.execute
    utils.secure_tar_unzip = function(file_path, release_path)
        local cmd = string.format('/usr/bin/unzip %s -d %s > /dev/null 2>&1', file_path, release_path)
        os.execute(cmd)
        release_path_ret = release_path
    end

    fw_mgmt_client.UpdateServiceUpdateServicePrepareReply = function(_, _, _, _, version, ret, parameters)
        fw_version_ret = version
        prepare_ret = ret
    end

    fw_mgmt_client.UpdateServiceUpdateServiceProcessReply = function()
    end

    fw_mgmt_client.UpdateServiceUpdateServiceFinishReply = function()
    end

    -- Mock process handlers
    local process_handle = require 'unit_manager.class.logic_fw.upgrade.process'
    self.old_upgrade_component_cpld = process_handle.upgrade_component_cpld
    process_handle.upgrade_component_cpld = function(self, db, system_id, fw_list, cfg_list, file_path, upgrade_list,
                                                     hot_upgrade)
        -- Always return success for these tests
        if fw_list == 1 then
            error('error')
        end
        return defs.RET.OK, true
    end

    local iic_upgrade = require 'unit_manager.class.logic_fw.upgrade.iic_process'
    self.old_iic_upgrade_cpld = iic_upgrade.iic_upgrade_cpld
    iic_upgrade.iic_upgrade_cpld = function(self,
        db, fw_list, cfg_list, file_path, system_id, firmware_type, parameters)
        -- Always return success for i2c upgrade tests
        return defs.RET.OK, true
    end

    local valid_handle = require 'unit_manager.class.logic_fw.upgrade.valid'
    self.old_cpld_valid_task = valid_handle.cpld_valid_task
    valid_handle.cpld_valid_task = function(signal, system_id)
        return true
    end

    self.old_cpld_hot_valid = valid_handle.cpld_hot_valid
    valid_handle.cpld_hot_valid = function(system_id, signal)
        return true
    end
end

function TestLogicalFwUpgrade:teardownClass()
    -- Restore original functions
    local process_handle = require 'unit_manager.class.logic_fw.upgrade.process'
    process_handle.upgrade_component_cpld = self.old_upgrade_component_cpld

    local iic_upgrade = require 'unit_manager.class.logic_fw.upgrade.iic_process'
    iic_upgrade.iic_upgrade_cpld = self.old_iic_upgrade_cpld

    local valid_handle = require 'unit_manager.class.logic_fw.upgrade.valid'
    valid_handle.cpld_valid_task = self.old_cpld_valid_task
    valid_handle.cpld_hot_valid = self.old_cpld_hot_valid
end

local function rm_hpm_package()
    local cmd
    if release_path_ret then
        cmd = string.format('/usr/bin/rm %s*.svf > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
        cmd = string.format('/usr/bin/rm %s*.vme > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
        cmd = string.format('/usr/bin/rm %s*.bin > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
    end
    cmd = string.format('/usr/bin/rm %s', file_path)
    os.execute(cmd)
end

local function build_hpm_package(cpld, cpld_len, vaild, valid_len)
    local file = file_sec.open_s(cpld, 'w+')
    file:write(string.rep('1', cpld_len))
    file:close()

    if vaild then
        file = file_sec.open_s(vaild, 'w+')
        file:write(string.rep('1', valid_len))
        file:close()
    end
    local cmd = string.format('/usr/bin/zip %s -j %s %s > /dev/null 2>&1', file_path, cpld, vaild)
    os.execute(cmd)
    -- 重命名
    cmd = string.format('/usr/bin/mv %s %s', file_path .. '.zip', file_path)
    os.execute(cmd)
    -- 删掉过程文件
    if vaild then
        cmd = string.format('/usr/bin/rm %s %s', cpld, vaild)
    else
        cmd = string.format('/usr/bin/rm %s', cpld)
    end
    os.execute(cmd)
end

local function mock_mutli_cold_valid()
    drivers.get_cpld_device_info = function()
    end

    chip.check_has_pango = function()
        -- 打桩不包含紫光
        return false
    end
    chip.get_cpld_upgrade_file_id = function()
        return 1
    end

    local old_init_valid_file = valid.init_valid_file

    valid.init_valid_file = function()
    end

    -- Mock the configuration loading to ensure check_fw_uid_exist method exists
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 2,        -- 多厂商模式
                cold_valid_list = { 1, 2 }, -- 冷升级列表
                hot_valid_list = {},
                chip_num = 2
            }
        }
    end

    return {
        restore = function()
            cfg.get_cfg_list = old_get_cfg_list
            valid.init_valid_file = old_init_valid_file
        end
    }
end

-- 多厂商、冷升级、正常的HPM包
function TestLogicalFwUpgrade:test_mutli_cold_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- 用例场景打桩
    local mock_handle = mock_mutli_cold_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    mock_handle.restore()

    -- 清理hpm包
    rm_hpm_package()
end

-- 多厂商、冷升级、异常的HPM包
function TestLogicalFwUpgrade:test_mutli_cold_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_02, 1)
    defs.UPGRADING_CPLD = 0

    -- Override the process mock to return failure for this test
    local process_handle = require 'unit_manager.class.logic_fw.upgrade.process'
    local old_upgrade_component_cpld = process_handle.upgrade_component_cpld
    process_handle.upgrade_component_cpld = function(...)
        return defs.RET.ERR, false
    end

    -- 用例场景打桩
    local mock_handle = mock_mutli_cold_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    mock_handle.restore()
    process_handle.upgrade_component_cpld = old_upgrade_component_cpld

    -- 清理hpm包
    rm_hpm_package()
end

local function mock_mutli_hot_valid()
    local old_get_obj = factory.get_obj
    factory.get_obj = function()
        return {
            cpu_boards = {},
            exp_boards = {}
        }
    end

    -- Mock the configuration loading for hot upgrade
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 2,    -- 多厂商模式
                cold_valid_list = {}, -- 空冷升级列表
                hot_valid_list = { 1 }, -- 热升级列表
                chip_num = 2
            }
        }
    end

    return {
        restore = function()
            factory.get_obj = old_get_obj
            cfg.get_cfg_list = old_get_cfg_list
        end
    }
end

-- 多厂商、无感升级、正常的HPM包
function TestLogicalFwUpgrade:test_mutli_hot_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- 用例场景打桩
    local mock_handle = mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path, {FirmwareIndex = 'Firmware1'})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    mock_handle.restore()

    -- 清理hpm包
    rm_hpm_package()
end

-- 多厂商、无感升级、异常的HPM包
function TestLogicalFwUpgrade:test_mutli_hot_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_02, 1)
    defs.UPGRADING_CPLD = 0

    -- Override the process mock to return failure for this test
    local process_handle = require 'unit_manager.class.logic_fw.upgrade.process'
    local old_upgrade_component_cpld = process_handle.upgrade_component_cpld
    process_handle.upgrade_component_cpld = function(...)
        return defs.RET.ERR, false
    end

    -- 用例场景打桩
    local mock_handle = mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    mock_handle.restore()
    process_handle.upgrade_component_cpld = old_upgrade_component_cpld

    -- 清理hpm包
    rm_hpm_package()
end

-- 单厂商、冷升级、正常的HPM包
function TestLogicalFwUpgrade:test_single_cold_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_vme, 1, valid_vme, 1)

    -- Mock configuration for single supplier mode
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 1, -- 单厂商模式
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1
            }
        }
    end

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_single_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    cfg.get_cfg_list = old_get_cfg_list

    -- 清理hpm包
    rm_hpm_package()
end

-- 单厂商、冷升级、异常的HPM包
function TestLogicalFwUpgrade:test_single_cold_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- Override the process mock to return failure for this test
    local process_handle = require 'unit_manager.class.logic_fw.upgrade.process'
    local old_upgrade_component_cpld = process_handle.upgrade_component_cpld
    process_handle.upgrade_component_cpld = function(...)
        return defs.RET.ERR, false
    end

    -- Mock configuration for single supplier mode
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 1, -- 单厂商模式
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1
            }
        }
    end

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_single_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复配置
    cfg.get_cfg_list = old_get_cfg_list
    process_handle.upgrade_component_cpld = old_upgrade_component_cpld

    -- 清理hpm包
    rm_hpm_package()
end

local function mock_i2c_mode(test_self)
    fw_mgmt_client.UpdateServiceUpdateServiceUpdateUpgradeStatus = function()
    end

    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    local old_is_support_multifirmware = cmn.is_support_multifirmware

    cfg.get_cfg_list = function(cfg_path)
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 1,
                update_link = "1",
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1
            }
        }
    end

    cfg.is_support_multifirmware = function()
        return false
    end

    -- 保存原函数以便恢复
    test_self.old_cfg_get_cfg_list = old_get_cfg_list
    test_self.old_is_support_multifirmware = old_is_support_multifirmware
end

-- i2c升级两小盘背板
function TestLogicalFwUpgrade:test_i2c_mode()
    -- 生成hpm包
    build_hpm_package(bin, 0X60000)

    -- 用例场景打桩
    mock_i2c_mode(self)

    -- Reset any previous test state
    fw_version_ret = ""
    prepare_ret = nil

    -- Reset upgrade state
    defs.UPGRADING_CPLD = 0
    fw_upgrade.hot_upgrade = false
    fw_upgrade.cpld_valid = false

    -- Ensure signal has correct structure for i2c test
    local test_signal = {
        fw_list = signal.fw_list,
        db = signal.db
    }

    -- 执行测试
    fw_upgrade.prepare_upgrade(test_signal, SYSTEM_ID, 'CPLD', cfg_i2c_path, {})
    fw_upgrade.process_upgrade(test_signal, SYSTEM_ID, 'CPLD', file_path, {})
    fw_upgrade.finish_upgrade(test_signal, SYSTEM_ID, 'CPLD')

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 恢复原函数
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    cfg.get_cfg_list = self.old_cfg_get_cfg_list
    cfg.is_support_multifirmware = self.old_is_support_multifirmware

    -- 清理hpm包
    rm_hpm_package()
end

local function start_ac(version)
    defs.UPGRADING_CPLD = 0
    fw_version_ret = version
end

-- 冷升级后AC，升级成功
function TestLogicalFwUpgrade:test_cold_valid_hpm_and_ac()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    local mock_handle = mock_mutli_cold_valid()
    -- 执行测试
    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    start_ac(FW_VERSION)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
    mock_handle.restore()
end

-- 冷升级后热升级，升级成功
function TestLogicalFwUpgrade:test_cold_and_hot_valid_hpm()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    local mock_handle = mock_mutli_cold_valid()
    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)
    defs.UPGRADING_CPLD = 0

    -- 生成热升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path, {})
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path, {})
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
    mock_handle.restore()
end

-- 冷升级生效过程中触发升级，升级失败
function TestLogicalFwUpgrade:test_active_cpld_and_upgrade()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    local mock_handle = mock_mutli_cold_valid()
    -- 模拟生效过程中
    fw_upgrade.cpld_valid = true
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    fw_upgrade.cpld_valid = false

    -- 校验结果
    lu.assertEquals(prepare_ret, defs.RET.ERR)

    -- 清理hpm包
    rm_hpm_package()
    mock_handle.restore()
end

-- 测试单Host机型准备阶段版本号返回实际值
function TestLogicalFwUpgrade:test_singlehost_prepare_version()
    -- 执行测试
    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
end

-- 测试HPC机型准备阶段版本号返回""
function TestLogicalFwUpgrade:test_mutlihost_prepare_version()
    -- 执行测试
    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, 255, _, cfg_mutil_cold_path, {})

    -- 校验结果
    lu.assertEquals(fw_version_ret, "")
    lu.assertEquals(prepare_ret, defs.RET.OK)
end

local function deepCopy(original)
    local copy
    if type(original) == "table" then
        copy = {}
        for key, value in pairs(original) do
            copy[key] = deepCopy(value)
        end
    else
        copy = original
    end
    return copy
end

-- 测试process阶段的正常处理流程
function TestLogicalFwUpgrade:test_outlier()
    -- 保存原始状态
    local original_upgrading = defs.UPGRADING_CPLD
    
    -- 测试正常的process流程 - fw_list存在但为空时
    local test_signal = deepCopy(signal)
    test_signal.fw_list = {}
    test_signal.upg_cfg_list = signal.upg_cfg_list
    defs.UPGRADING_CPLD = 0
    local ret = fw_upgrade.process_upgrade(test_signal, 1, _, cfg_mutil_cold_path, {})
    -- 验证有空fw_list时返回成功（mock会处理）
    lu.assertEquals(ret, defs.RET.OK)
    
    -- 测试正常的process流程 - 使用非I2C升级模式
    test_signal = deepCopy(signal)
    -- 确保不使用I2C模式
    if test_signal.upg_cfg_list and test_signal.upg_cfg_list[1] then
        test_signal.upg_cfg_list[1].update_link = "0"
    end
    defs.UPGRADING_CPLD = 0
    ret = fw_upgrade.process_upgrade(test_signal, 1, _, cfg_mutil_cold_path, {})
    -- 验证非I2C模式升级成功
    lu.assertEquals(ret, defs.RET.OK)
    
    -- 测试使用空文件路径的情况
    test_signal = deepCopy(signal)
    defs.UPGRADING_CPLD = 0
    ret = fw_upgrade.process_upgrade(test_signal, 1, _, "", {})
    -- 验证空文件路径时也能返回成功（由mock处理）
    lu.assertEquals(ret, defs.RET.OK)
    
    -- 恢复原始状态
    defs.UPGRADING_CPLD = original_upgrading
end

-- 测试已有升级任务触发并行升级
function TestLogicalFwUpgrade:test_parallel_in_upgrade()
    -- 执行测试
    defs.UPGRADING_CPLD = 1
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path, {})
    -- 校验结果
    lu.assertEquals(prepare_ret, defs.RET.IN_UPGRADING)
    defs.UPGRADING_CPLD = 0
end

-- 测试FPGA正在升级时阻止CPLD升级
function TestLogicalFwUpgrade:test_fpga_upgrading_blocks_cpld()
    -- 模拟FPGA正在升级
    defs.UPGRADING_FPGA = 1
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    -- 校验结果
    lu.assertEquals(prepare_ret, defs.RET.OTHERS_UPGRADING)
    -- 恢复状态
    defs.UPGRADING_FPGA = 0
end

-- 测试获取固件版本失败时返回空字符串
function TestLogicalFwUpgrade:test_get_fw_version_fail()
    local test_signal = deepCopy(signal)
    test_signal.fw_list[1].get_fw_version = function()
        error("get version failed")
    end

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(test_signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    lu.assertEquals(fw_version_ret, "")
end

-- 测试热升级版本为"0.00"时回退到常规版本获取的逻辑
function TestLogicalFwUpgrade:test_hot_upgrade_version_zero()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list

    -- Mock配置为单厂商模式，这样会跳过热升级检查，直接获取版本
    cfg.get_cfg_list = function()
        return {
            {
                supplier_mode = 1, -- 单厂商模式，跳过热升级检查
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1,
                check_fw_uid_exist = function(fw)
                    return true
                end
            }
        }
    end

    local test_signal = deepCopy(signal)
    -- 设置固件版本返回"0.00"
    test_signal.fw_list[1].get_fw_version = function()
        return "0.00"
    end
    test_signal.fw_list[1].system_id = SYSTEM_ID

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(test_signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    -- 测试单厂商模式下版本为"0.00"的情况：
    -- 单厂商模式会跳过热升级检查，直接调用get_fw_version
    -- 应该返回"0.00"
    lu.assertEquals(fw_version_ret, "0.00")

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
end

-- 测试多固件支持时的固件索引获取逻辑
function TestLogicalFwUpgrade:test_multifirmware_support()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    local old_is_support_multifirmware = cmn.is_support_multifirmware

    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function()
                    return true
                end,
                supplier_mode = 1,
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1
            }
        }
    end

    cfg.is_support_multifirmware = function()
        return true
    end

    local parameters = { FirmwareIndex = "Firmware1" }

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, parameters)
    -- 验证多固件支持标志被正确设置
    lu.assertEquals(parameters.MultipleSupported, 'Support')

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
    cfg.is_support_multifirmware = old_is_support_multifirmware
end

-- 测试匹配固件配置失败的情况
function TestLogicalFwUpgrade:test_find_firmware_cfg_fail()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    local old_is_support_multifirmware = cmn.is_support_multifirmware

    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function()
                    return false -- 模拟固件UID不匹配
                end,
                supplier_mode = 1,
                cold_valid_list = {},
                hot_valid_list = {},
                chip_num = 1
            }
        }
    end

    cfg.is_support_multifirmware = function()
        return true
    end

    local parameters = { FirmwareIndex = "Firmware1" } -- 使用存在但UID不匹配的固件索引

    defs.UPGRADING_CPLD = 0
    -- 执行prepare_upgrade，由于UID不匹配，应该返回空版本
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, parameters)

    -- 验证固件配置匹配失败时版本为空
    lu.assertEquals(fw_version_ret, "")

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
    cfg.is_support_multifirmware = old_is_support_multifirmware
end

-- 测试冷升级列表非空时的处理逻辑
function TestLogicalFwUpgrade:test_cold_valid_list_not_empty()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list

    cfg.get_cfg_list = function()
        return {
            {
                supplier_mode = 2,        -- 多厂商模式
                cold_valid_list = { 1, 2 }, -- 非空冷升级列表
                hot_valid_list = {},
                chip_num = 3,
                check_fw_uid_exist = function()
                    return true
                end
            }
        }
    end

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    -- 验证冷升级模式被正确设置
    lu.assertEquals(fw_upgrade.hot_upgrade, false)

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
end

-- 测试热升级列表非空时的处理逻辑
function TestLogicalFwUpgrade:test_hot_valid_list_not_empty()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list

    cfg.get_cfg_list = function()
        return {
            {
                supplier_mode = 2,    -- 多厂商模式
                cold_valid_list = {}, -- 空冷升级列表
                hot_valid_list = { 1 }, -- 非空热升级列表
                chip_num = 3
            }
        }
    end

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    -- 验证热升级模式被正确设置
    lu.assertEquals(fw_upgrade.hot_upgrade, true)

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
end

-- 测试配置列表为空或nil的情况
function TestLogicalFwUpgrade:test_empty_cfg_list()
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {} -- 返回空配置列表
    end

    defs.UPGRADING_CPLD = 0
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, 'CPLD', cfg_mutil_cold_path, {})
    -- 配置列表为空时，准备阶段应返回错误
    lu.assertEquals(prepare_ret, defs.RET.ERR)
    -- CPLD升级标志应被清除
    lu.assertEquals(defs.UPGRADING_CPLD, 0)

    -- 恢复原函数
    cfg.get_cfg_list = old_get_cfg_list
end

-- process失败
function TestLogicalFwUpgrade:test_process_fail()
    local tab = {
        upg_cfg_list = { [1] = { update_link = '2' } },
        fw_list = 1
    }
    local ret = fw_upgrade.process_upgrade(tab, SYSTEM_ID, _, file_path, {})
    lu.assertEquals(ret, -1)
end

local function mock_mutli_full_hot_valid()
    local old_get_obj = factory.get_obj
    factory.get_obj = function()
        return {
            cpu_boards = {},
            exp_boards = {}
        }
    end

    -- Mock the configuration loading for hot upgrade
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 2,    -- 多厂商模式
                cold_valid_list = {}, -- 空冷升级列表
                hot_valid_list = { 1 }, -- 热升级列表
                chip_num = 2,
                full_hot_valid = true
            }
        }
    end

    return {
        restore = function()
            factory.get_obj = old_get_obj
            cfg.get_cfg_list = old_get_cfg_list
        end
    }
end

local function mock_mutli_valid()
    local old_get_obj = factory.get_obj
    factory.get_obj = function()
        return {
            cpu_boards = {},
            exp_boards = {}
        }
    end

    -- Mock the configuration loading for hot upgrade
    local cfg = require 'unit_manager.class.logic_fw.upgrade.fw_cfgs'
    local old_get_cfg_list = cfg.get_cfg_list
    cfg.get_cfg_list = function()
        return {
            {
                name = "Firmware1",
                check_fw_uid_exist = function(fw)
                    return true
                end,
                supplier_mode = 2,    -- 多厂商模式
                cold_valid_list = {}, -- 空冷升级列表
                hot_valid_list = {}, -- 热升级列表
                chip_num = 2
            }
        }
    end

    return {
        restore = function()
            factory.get_obj = old_get_obj
            cfg.get_cfg_list = old_get_cfg_list
        end
    }
end

function TestLogicalFwUpgrade:test_hot_prepare_upgrade()
    defs.UPGRADING_CPLD = 0
    local mock_handle = mock_mutli_full_hot_valid()
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path, {FirmwareIndex = 'Firmware1'})
    mock_handle.restore()
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
end

function TestLogicalFwUpgrade:test_mutli_prepare_upgrade()
    defs.UPGRADING_CPLD = 0
    local mock_handle = mock_mutli_valid()
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path, {FirmwareIndex = 'Firmware1'})
    mock_handle.restore()
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
end