-- 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 utils = require 'power_mgmt_utils'
local power_upgrade = (require 'power_upgrade').get_instance()
local file_sec = require 'utils.file'
local utils_core = require 'utils.core'
local c_psu_object = require 'device.psu'
local c_power_configuration = require 'device.power_configuration'
local signal = require 'signal'
local client = require 'power_mgmt.client'
local c_power_supplies_object = require 'device.power_supplies'
local plugin = require 'hwproxy.plugins.power_mgmt.init'
local canbus_tpsu = require 'device_tree.adapters.power_mgmt.protocol.canbus_tpsu'
local lock_chip = (require 'lock_chip').get_instance()
local fw_def = require 'macros.fw_def'
local E_OK <const> = nil -- 函数执行成功返回nil
local E_FAILED <const> = '' -- 空错误信息

TestPowerUpgrade = {}
utils.cfgs.cfgs = {{component_idex = 2}}
utils.secure_tar_unzip = function () return true end

c_power_configuration.collection.find = function ()end
c_power_supplies_object.collection.find = function ()end

-- 对依赖环境较重的接口做简单打桩，避免 UT 依赖真实硬件/DB
power_upgrade.process_upgrade = function(...) end
power_upgrade.finish_upgrade = function(...) end
power_upgrade.upgrade_power_process = function(...) end

local f_log = file_sec.open_s('/root/test_upgrade', 'w+')
utils.close(f_log, pcall(f_log.write, f_log, 'xxx'))

client.PFileFileChown = function() return true end
client.UpdateServiceUpdateServicePrepareReply = function () end
client.UpdateServiceUpdateServiceProcessReply = function () end
client.UpdateServiceUpdateServiceFinishReply = function () end

function TestPowerUpgrade:test_prepare()
    local ok = pcall(function ()
        power_upgrade.prepare_upgrade(1, 'Power', '/root/test_upgrade', '', {})
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_procesee()
    local ok = pcall(function ()
        power_upgrade.process_upgrade(nil, 1, 'Power', '/root/test_upgrade', {})
    end)
    lu.assertTrue(ok)

    local ok = pcall(function ()
        power_upgrade.process_upgrade(nil, 1, 'Non', '/root/test_upgrade', {})
    end)
    lu.assertTrue(ok)

    utils.cfgs.cfgs = {{component_idex = 0x7D}}
    local ok = pcall(function ()
        power_upgrade.process_upgrade(nil, 1, 'Power', '/root/test_upgrade', {})
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_finish()
    utils.get_instance().get_update_cfg_firmware_version = function ()
        return 1
    end
    utils.get_instance().get_upgrade_psu_objs = function ()
        local psu_objs = {
            {ps_id = 1, FirmwareVersion = 1},
            {ps_id = 2, FirmwareVersion = 2}
        }
        return psu_objs
    end
    local ok = pcall(function ()
        power_upgrade.finish_upgrade(1, 'Power', {})
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_upgrade_set_psu_channle_1()
    local psj = {}
    psj.get_upgrade_channel = function ()
        return E_FAILED
    end
    local res = power_upgrade.upgrade_set_psu_channle(psj)
    lu.assertEquals(res, E_FAILED)
end

function TestPowerUpgrade:test_upgrade_set_psu_channle_2()
    local psj = {}
    psj.get_upgrade_channel = function ()
        return 1
    end
    c_psu_object.collection.fold = function (...)
        return E_OK
    end
    local res = power_upgrade.upgrade_set_psu_channle(psj)
    lu.assertEquals(res, 1)
end

function TestPowerUpgrade:test_upgrade_recover_psu_channle_1()
    c_psu_object.collection.fold = function (...)
        return E_OK
    end
    local psj = {}
    local res1 = E_OK
    local res = power_upgrade.upgrade_recover_psu_channle(psj, res1)
    lu.assertEquals(res, E_OK)
end

function TestPowerUpgrade:test_upgrade_recover_psu_channle_2()
    c_psu_object.collection.fold = function (...)
        return E_OK
    end
    local psj = {}
    local res1 = E_FAILED
    local res = power_upgrade.upgrade_recover_psu_channle(psj, res1)
    lu.assertEquals(res, E_OK)
end

function TestPowerUpgrade:test_upgrade_plugin_canbus_tpsu_1()
   local chip, block_id, block_data, len, slot_addr
   local frame_num = 0
   local res = plugin:run_cmd(chip, 'canbus_load_firmware_by_frame_tpsu', block_id, frame_num,
                                                                    block_data, len, slot_addr)
   lu.assertEquals(res, E_OK)
end

function TestPowerUpgrade:test_upgrade_plugin_canbus_tpsu_2()
    local chip, block_data, slot_addr
    local frame_num = 1
    local len = 1
    local block_id = 1
    local res = plugin:run_cmd(chip, 'canbus_load_firmware_by_frame_tpsu', block_id, frame_num,
                                                                        block_data, len, slot_addr)
    lu.assertEquals(res, E_FAILED)
 end

 function TestPowerUpgrade:test_canbus_tpsu_upgrade_file_check()
    local canbus_upgrade_tpsu_bin_table_test = {
        "PS_HD.bin",
        "PS_ZY.bin",
        "PS_DCDC.bin",
        "PS_PFC.bin",
        "PS_TURBO.bin",
        "PS_TURBOSAMPLE.bin",
    }
    local i = 1
    local canbus_upgrade_bin_name = canbus_tpsu:canbus_get_upgrade_bin_table()
    for _, file_name in ipairs(canbus_upgrade_bin_name) do
        lu.assertEquals(file_name, canbus_upgrade_tpsu_bin_table_test[i])
        i = i + 1
    end
end

function TestPowerUpgrade:test_upgrade_psu_check()
    local psu_obj = {}
    psu_obj.psu_slot = {
        PsuChip = {},
        PsuBackupChip = {}
    }
    psu_obj.power_upgrade = function (...)
        return E_OK
    end
    lock_chip.retry_unlock_chip = function (...)
        return E_OK
    end
    lock_chip.retry_lock_chip = function (...)
        return E_OK
    end
    psu_obj.LossOfInput = 0
    local res = power_upgrade.upgrade_psu(psu_obj, nil, nil, nil, 0)
    lu.assertEquals(res, E_OK)
    res = power_upgrade.upgrade_psu(psu_obj, nil, nil, nil, 1)
    lu.assertEquals(res, E_OK)
end

function TestPowerUpgrade:test_upgrade_psu_loss_of_input()
    local psu_obj = {
        ps_id = 1,
        LossOfInput = 1
    }
    local ret = power_upgrade.upgrade_psu(psu_obj, nil, nil, nil, 0)
    lu.assertEquals(ret, fw_def.SUPPLY_3)
end

function TestPowerUpgrade:test_upgrade_check()
    local ok = pcall(function ()
        power_upgrade.upgrade_power_process(1, 'Power', '/root/test_upgrade', 0)
    end)
    lu.assertTrue(ok)
    ok = pcall(function ()
        c_psu_object:power_upgrade()
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_upgrade_result_check()
    local result = {}
    result[1] = 0
    result[2] = 0
    local psu_num = 2
    local ok = pcall(function ()
        power_upgrade.check_upgrade_result(result, psu_num)
    end)
    lu.assertTrue(ok)

    local result1 = {}
    local psu_num1 = 2
    local ok1 = pcall(function ()
        power_upgrade.check_upgrade_result(result1, psu_num1)
    end)
    lu.assertTrue(ok1)

    local result2 = {}
    result2[1] = 206
    result2[2] = 0
    local psu_num2 = 2
    local ok2 = pcall(function ()
        power_upgrade.check_upgrade_result(result2, psu_num2)
    end)
    lu.assertTrue(ok2)
end

function TestPowerUpgrade:test_upgrade_each_psu_check()
    local tab = {}
    local tab2 = {}
    power_upgrade.upgrade_process.init(tab, 1)
    local ok = pcall(function ()
        power_upgrade.upgrade_each_psu('/root/test_upgrade', tab, 1, tab2)
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_signal()
    local ok = pcall(function ()
        signal.upgrade_prepare_callback()
    end)
    lu.assertTrue(ok)

    ok = pcall(function ()
        signal.upgrade_process_callback()
    end)
    lu.assertTrue(ok)

    local tmp = {
        db = {
            AclostRecordTable = function()
                return{
                    Id = "AclostRecord",
                    ResetFlag = 1,
                    Flag = 0,
                    CurTime = 0
                }
            end
        },
        gpio_work = {
            start_module = function(module)
            end
        },
        get_ac_lost_flag = function()end,
        aclost_int_task = function(aclost_record)end
    }
    local system_id= 1
    COMP_CPLD_TYPE = 'Cpld'
    signal.db = tmp.db
    ok = pcall(function ()
        signal.upgrade_finish_callback(tmp, system_id, COMP_CPLD_TYPE)
    end)
    lu.assertTrue(ok)
end

function TestPowerUpgrade:test_upgrade_status_check()
    local tab = {}
    local res = c_power_supplies_object.get_upgrade_status(tab)
    lu.assertEquals(res, nil)
end

function TestPowerUpgrade:test_set_upgrade_status_check()
    local tab = {}
    local upgrade_status = 1
    local res = c_power_supplies_object.set_upgrade_status(tab, upgrade_status)
    lu.assertEquals(res, nil)
end