-- Copyright (c) 2025 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 client = require 'bios.client'
local fw_signal = require 'interface.signal'
local bios_factory = require 'factory.bios_factory'
local krun_service = require 'service.krun_service'
local fructl_handler = require 'infrastructure.fructl'
local file_util = require 'infrastructure.file_util'
local bs = require 'mc.bitstring'
local mdb = require 'mc.mdb'
local krun_firmware = require 'interface.mdb.krun_firmware'
local bin_builder = require 'libmgmt_protocol.bios.bin_parser.bin_parser_builder'
local package_builder = require 'domain.bios_firmware.package.package_builder'
local hisport_channel = require 'domain.bios_firmware.package.channel.hisport_channel'
local open_local_db = require 'bios.local_db'
local open_db = require 'bios.db'
local utils_core = require 'utils.core'
local utils = require 'mc.utils'
local ipmi = require 'ipmi'
local comp_code = ipmi.types.Cc

TestKrunUpgrade = {}

local REPLY_ERR = -1
local REPLY_OK = 0

local client_mock = client

local function construct_local_db()
    local ok, datas = pcall(require, 'bios.datas')
    if not ok then
        -- 如果没有datas配置，证明当前组件不需要datas，仅打开数据库
        datas = nil
    end
    utils.remove_file(utils_core.getcwd() .. '/bios.db')
    local local_db = open_local_db(
                        utils_core.getcwd() ..
                            '/bios.db', datas, 'poweroff')
    return local_db
end

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

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

    return ctx
end

local function test_signal_prepare()
    local lu_obj = {}
    client.PUpdateServiceUpdateServicePrepareReply = function(_,
        ctx, system_id, firmware_type, version, status, parameters)
        lu_obj.firmware_type = firmware_type
        lu_obj.version = version
        lu_obj.status = status
    end
    client.UpdateServiceUpdateServicePrepareReply = function(_,
        ctx, system_id, firmware_type, version, status, parameters)
        lu_obj.firmware_type = firmware_type
        lu_obj.version = version
        lu_obj.status = status
    end

    fw_signal.upgrade_prepare_callback({}, 1, 'Krun', '', '', {})
    lu.assertEquals(lu_obj.firmware_type, 'Krun')
    lu.assertEquals(lu_obj.version, '')
    lu.assertEquals(lu_obj.status, REPLY_ERR)

    local cfg = {
        system_id = 1,
        firmware_type = 'Krun',
        cfg_path = '',
        hpm_path = '',
        context = {},
        para = {}
    }
    lu_obj = {}
    fw_signal.prepare_upgrade(cfg, {})
    lu.assertEquals(lu_obj.version, '1.23')
    lu.assertEquals(lu_obj.status, REPLY_ERR)

    local build = package_builder.build
    package_builder.build = function ()
        return {
            period = 1,
            package_type = 'Krun',
            uid_list = {'11111', '22222'},
            process = function()
            end
        }
    end
    lu_obj = {}
    fw_signal.prepare_upgrade(cfg, {})
    lu.assertEquals(lu_obj.status, REPLY_OK)

    client.PUpdateServiceUpdateServicePrepareReply = client_mock.PUpdateServiceUpdateServicePrepareReply
    client.UpdateServiceUpdateServicePrepareReply = client_mock.UpdateServiceUpdateServicePrepareReply
    client.PFileFileChown = client_mock.PFileFileChown
    package_builder.build = build
end

local function test_signal_process()
    local lu_obj = {}
    client.PUpdateServiceUpdateServiceProcessReply = function(_,
        ctx, system_id, firmware_type, status, parameters)
        lu_obj.firmware_type = firmware_type
        lu_obj.status = status
    end
    fw_signal.upgrade_process_callback({}, 1, 'Cca', '', {})
    lu.assertEquals(lu_obj.firmware_type, 'Cca')
    fw_signal.upgrade_process_callback({}, 1, 'Krun', '', {})
    lu.assertEquals(lu_obj.firmware_type, 'Krun')
    lu.assertEquals(lu_obj.status, REPLY_ERR)

    local cfg = {
        system_id = 1,
        firmware_type = 'Krun',
        file_path = '',
        context = {},
        action = ''
    }

    fw_signal.process_upgrade(cfg, {})
    lu.assertEquals(lu_obj.status, REPLY_OK)
    client.PUpdateServiceUpdateServiceProcessReply = client_mock.PUpdateServiceUpdateServiceProcessReply
end

local function test_signal_finish()
    local lu_obj = {}
    client.PUpdateServiceUpdateServiceFinishReply = function(_,
        ctx, system_id, firmware_type, status, parameters)
        lu_obj.firmware_type = firmware_type
        lu_obj.status = status
    end
    fw_signal.upgrade_finish_callback({}, 1, 'Cca', {})
    lu.assertEquals(lu_obj, {})
    fw_signal.upgrade_finish_callback({}, 1, 'Krun', {})
    lu.assertEquals(lu_obj, {})

    local cfg = {
        system_id = 1,
        firmware_type = 'Krun',
        context = {},
        action = ''
    }

    fw_signal.finish_upgrade(cfg, {})
    lu.assertEquals(lu_obj.status, REPLY_OK)
    client.PUpdateServiceUpdateServiceProcessReply = client_mock.PUpdateServiceUpdateServiceProcessReply
end

local function test_krun_service(krun_ser)
    krun_ser.krun_flash_collection[1].is_cache = function ()
        return false
    end
    local result = krun_ser:power_off_effective(1)
    lu.assertEquals(result, REPLY_OK)
    result = krun_ser:power_off_effective(255)
    lu.assertEquals(result, REPLY_OK)
    krun_ser.krun_flash_collection[1].is_cache = function ()
        return true
    end
    local get_power_status = fructl_handler.get_power_status
    local set_power_on_lock = fructl_handler.set_power_on_lock
    fructl_handler.get_power_status = function ()
        return 'ON'
    end
    result = krun_ser:power_off_effective(1)
    lu.assertEquals(result, REPLY_OK)
    fructl_handler.get_power_status = function ()
        return 'OFF'
    end

    fructl_handler.set_power_on_lock = function ()
        return false
    end
    result = krun_ser:power_off_effective(1)
    lu.assertEquals(result, REPLY_ERR)

    fructl_handler.set_power_on_lock = function ()
        return true
    end
    result = krun_ser:power_off_effective(1)
    lu.assertEquals(result, REPLY_ERR)

    local decompress_file = file_util.decompress_file
    file_util.decompress_file = function ()
    end
    result = krun_ser:power_off_effective(1)
    lu.assertEquals(result, REPLY_ERR)

    local build = bin_builder.build
    bin_builder.build = function ()
        local bin_parser = {}
        bin_parser.parse = function ()
            return {
                set_channels = function ()
                end,
                upgrade = function ()
                end
            }
        end
        return bin_parser
    end
    result = krun_ser:power_off_effective(1)
    -- TBD: -> REPLY_OK
    lu.assertEquals(result, REPLY_ERR)

    fructl_handler.get_power_status = get_power_status
    fructl_handler.set_power_on_lock = set_power_on_lock
    file_util.decompress_file = decompress_file
    bin_builder.build = build
end

local function test_register_signal(krun_ser)
    local signal = nil
    local signal_type = nil
    krun_ser.krun_flash_collection[1].is_cache = function ()
        return true
    end
    client.FirmwareActiveFirmwareActiveRegisterActiveAction = function (_, ctx, para)
        signal = true
        signal_type = para[3].Value
    end

    krun_ser:judge_active_register(1)
    lu.assertEquals(signal, true)
    lu.assertEquals(signal_type, 'PowerOff')

    signal = nil
    krun_ser:enable_multihost()
    krun_ser:judge_active_register(255)
    lu.assertEquals(signal, true)
    lu.assertEquals(signal_type, 'ChassisPowerOff')

    signal = nil
    signal_type = nil
    krun_ser:judge_active_register(1)
    lu.assertEquals(signal, true)
    lu.assertEquals(signal_type, 'ChassisPowerOff')

    client.FirmwareActiveFirmwareActiveRegisterActiveAction = client_mock.FirmwareActiveFirmwareActiveRegisterActiveAction
end

local function test_krun_version(krun_ser)
    local set_krun_version_bs = bs.new([[<<
        ManufactureId:24,
        SubCmd:8,
        ParameterSelector:8,
        LastFrame:8,
        Offset:16,
        Length:8,
        Version/string >>]])
    local ipmi_send = '\xdb\x07\x00\x6d\x00\x00\x00\x00\x05\x32\x30\x2e\x33\x37'
    local req = set_krun_version_bs:unpack(ipmi_send)
    local rsp = krun_ser:update_krun_version(req, construct_ctx())
    lu.assertEquals(rsp.CompletionCode, comp_code.Success)

    ipmi_send = '\xdb\x07\x00\x6d\x00\x00\x00\x00\x04\x32\x30\x2e\x33\x37'
    req = set_krun_version_bs:unpack(ipmi_send)
    rsp = krun_ser:update_krun_version(req, construct_ctx())
    lu.assertEquals(rsp.CompletionCode, comp_code.ReqDataLenInvalid)

    ipmi_send = '\xdb\x07\x01\x6d\x00\x00\x00\x00\x05\x32\x30\x2e\x33\x37'
    req = set_krun_version_bs:unpack(ipmi_send)
    rsp = krun_ser:update_krun_version(req, construct_ctx())
    lu.assertEquals(rsp.CompletionCode, comp_code.InvalidCommand)
end

function TestKrunUpgrade:test_signal()
    local obj = {
        Id = 1,
        UId = '11111',
        RelatedSystems = {1}
    }
    local krun_ser = krun_service.new({}, construct_db(), construct_local_db())
    krun_ser:add_obj(obj)
    krun_ser:set_dft_mode()
    lu.assertEquals(krun_ser.upgrade_mode, 2)
    lu.assertEquals(krun_ser.origin_mode, 2)

    local krun_obj = krun_ser:get_obj(1)
    krun_obj:set_version('1.23')
    lu.assertEquals(krun_ser.krun_flash_collection[1].Version, '1.23')

    bios_factory.register_bean('krun_service', krun_ser)
    test_signal_prepare()
    test_signal_process()
    test_signal_finish()
    test_krun_service(krun_ser)
    test_register_signal(krun_ser)
    test_krun_version(krun_ser)
end

function TestKrunUpgrade:test_firmware_krun_version()
    local mdb_obj = {}
    local get_cached_object = mdb.get_cached_object
    mdb.get_cached_object = function ()
        return mdb_obj
    end
    local krun_fw = krun_firmware.new({}, construct_db(), 2, '', {1, 2})
    krun_fw:update_krun_version('2.34')
    lu.assertEquals(mdb_obj.Version, '2.34')
    mdb.get_cached_object = get_cached_object
end

function TestKrunUpgrade:test_hisport_flash()
    local his_channel = hisport_channel.new()
    local ctx = {
        BinPath = '',
        BlockChip = {},
        FlashChip = {},
    }
    local properties = {}
    local ok = pcall(function ()
        his_channel:upgrade(ctx, properties)
    end)
    lu.assertEquals(ok, false)

    properties = {
        Offset = 0,
        Length = 32
    }
    ok = pcall(function ()
        his_channel:upgrade(ctx, properties)
    end)
    lu.assertEquals(ok, false)
end