-- 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'

TestVrdUpgrade = {}

function TestVrdUpgrade:setUp()
    -- 在加载任何模块前先模拟依赖
    self.mock_bus = { test = "bus" }
    self.mock_db = { test = "db" }
    self.mock_vrd_collection = {}

    package.loaded['independent_vrd.chip.vrd_upgrade'] = nil

    self.mock_upgrade_service = {
        on_upgrade_prepare = function(self_param, system_id, firmware_type, cfg_path, hpm_path, parameters)
            return "prepare_result"
        end,
        on_upgrade_process = function(self_param, system_id, firmware_type, file_path, parameters)
            return "process_result"
        end,
        on_upgrade_finish = function(self_param, system_id, firmware_type, parameters)
            return "finish_result"
        end,
        is_firmware_upgrading = function(self_param)
            return false
        end,
        get_vrd_load = function(self_param)
            return true
        end,
        register_vrd_active_action = function(self_param, firmware_type)
            return "register_result"
        end
    }

    package.loaded['mcu.upgrade.upgrade_service.upgrade_service_vrd'] = {
        new = function(bus, db, vrd_collection)
            return self.mock_upgrade_service
        end
    }
    package.loaded['utils.vos'] = {
        get_file_accessible = function(path)
            return true
        end
    }

    package.loaded['utils.core'] = {
        dir = function(path)
            return { "HOST_1_Independent_Vrd.hpm", "HOST_2_Independent_Vrd.hpm" }
        end
    }

    package.loaded['mc.utils'] = {
        string_split = function(str, delimiter, plain)
            if str == "HOST_1_Independent_Vrd.hpm" then
                return { "HOST", "1", "Independent", "Vrd.hpm" }
            elseif str == "HOST_2_Independent_Vrd.hpm" then
                return { "HOST", "2", "Independent", "Vrd.hpm" }
            end
            return {}
        end
    }

    package.loaded['mc.context'] = {
        new = function()
            return { test = "context" }
        end
    }

    self.firmware_mgmt_calls = {}
    package.loaded['general_hardware.client'] = {
        FirmwareActiveFirmwareActiveUpdateActiveStatus = function(context, param)
            table.insert(self.firmware_mgmt_calls, { func = "UpdateActiveStatus", context = context, param = param })
        end,
        FirmwareActiveFirmwareActiveActiveProcessReply = function(context, system_id, firmware_type, ret)
            table.insert(self.firmware_mgmt_calls,
                {
                    func = "ActiveProcessReply",
                    context = context,
                    system_id = system_id,
                    firmware_type = firmware_type,
                    ret =
                        ret
                })
        end
    }

    package.loaded['mcu.upgrade.upgrade_service.upgrade_service_comm'] = {
        update_active_status = function(firmware_id, firmware_type, status)
            local fw_mgmt = package.loaded['general_hardware.client']
            local ctx = package.loaded['mc.context'].new()
            local param = {}
            param[#param+1] = { Key = 'FirmwareId', Value = firmware_id }
            param[#param+1] = { Key = 'FirmwareType', Value = firmware_type }
            param[#param+1] = { Key = 'ActiveStatus', Value = status }
            fw_mgmt:FirmwareActiveFirmwareActiveUpdateActiveStatus(ctx, param)
        end
    }

    package.loaded['mcu.upgrade.defs'] = {
        FW_VALID_FILE_DIR = {
            Independent_Vrd = "/test/path/",
            INDEPENDENT_VRD = "/test/path/"
        },
        RET = {
            OK = 0
        }
    }

    self.validation_calls = {}
    package.loaded['mcu.upgrade.valid'] = {
        vrd_valid_task = function(upgrade_service, sys_id, file_path)
            table.insert(self.validation_calls,
                { upgrade_service = upgrade_service, sys_id = sys_id, file_path = file_path })
        end
    }

    self.sleep_calls = {}
    package.loaded['common'] = {
        skynet = {
            sleep = function(ms)
                table.insert(self.sleep_calls, ms)
            end
        }
    }

    package.loaded['mc.logging'] = {
        notice = function(format, ...)
        end,
        error = function(format, ...)
        end
    }

    package.loaded['upgrade_subject.upgrade_subject'] = {
        get_instance = function()
            return {
                register_upgrade_observer = function() end,
                register_active_observer = function() end
            }
        end
    }

    package.loaded['general_hardware.fructl'] = {
        is_multihost_type = function(bus)
            return false
        end
    }

    -- 加载模块并创建测试对象
    local vrd_upgrade = require 'independent_vrd.chip.vrd_upgrade'
    self.obj = vrd_upgrade.new(self.mock_bus, self.mock_db, self.mock_vrd_collection)
end

function TestVrdUpgrade:test_new_creates_upgrade_service()
    lu.assertNotNil(self.obj._upgrade_service)
    lu.assertEquals(self.obj.bus, self.mock_bus)
    lu.assertEquals(self.obj.db, self.mock_db)
    lu.assertEquals(self.obj.vrd_collection, self.mock_vrd_collection)
end

function TestVrdUpgrade:test_on_upgrade_prepare_delegates()
    local result = self.obj:on_upgrade_prepare(nil, "system1", "Independent_Vrd", "cfg_path", "hpm_path", {})
    lu.assertEquals(result, "prepare_result")
end

function TestVrdUpgrade:test_on_upgrade_process_delegates()
    local result = self.obj:on_upgrade_process(nil, "system1", "Independent_Vrd", "file_path", {})
    lu.assertEquals(result, "process_result")
end

function TestVrdUpgrade:test_on_upgrade_finish_delegates()
    local result = self.obj:on_upgrade_finish(nil, "system1", "Independent_Vrd", {})
    lu.assertEquals(result, "finish_result")
end

function TestVrdUpgrade:test_update_active_status_function()
    self.obj:on_active_process(nil, "system1", "INDEPENDENT_VRD")

    local update_calls = {}
    for _, call in ipairs(self.firmware_mgmt_calls) do
        if call.func == "UpdateActiveStatus" then
            table.insert(update_calls, call)
        end
    end

    lu.assertTrue(#update_calls > 0)
    lu.assertNotNil(update_calls[1].context)
    lu.assertNotNil(update_calls[1].param)
end

function TestVrdUpgrade:test_on_active_process_independent_vrd()
    self.obj:on_active_process(nil, "system1", "INDEPENDENT_VRD")

    lu.assertTrue(#self.validation_calls > 0)
    for _, call in ipairs(self.validation_calls) do
        lu.assertEquals(call.upgrade_service, self.obj._upgrade_service)
        lu.assertTrue(call.sys_id == 1 or call.sys_id == 2)
    end

    local reply_calls = {}
    for _, call in ipairs(self.firmware_mgmt_calls) do
        if call.func == "ActiveProcessReply" then
            table.insert(reply_calls, call)
        end
    end

    lu.assertTrue(#reply_calls > 0)

    local call = reply_calls[1]

    -- 验证期望的值在某个字段中存在
    local has_context = (type(call.context) == "table" and call.context.test == "context") or
        (type(call.system_id) == "table" and call.system_id.test == "context")
    local has_system1 = call.system_id == "system1" or call.firmware_type == "system1" or call.ret == "system1"
    local has_independent_vrd = call.firmware_type == "INDEPENDENT_VRD" or call.ret == "INDEPENDENT_VRD"

    lu.assertTrue(has_context)
    lu.assertTrue(has_system1)
    lu.assertTrue(has_independent_vrd)
end

function TestVrdUpgrade:test_on_active_process_file_processing()
    self.obj:on_active_process(nil, "system1", "INDEPENDENT_VRD")

    lu.assertEquals(#self.validation_calls, 2)

    local found_sys_1 = false
    local found_sys_2 = false
    for _, file_data in ipairs(self.validation_calls) do
        if file_data.sys_id == 1 then
            found_sys_1 = true
            lu.assertEquals(file_data.file_path, "/test/path/HOST_1_Independent_Vrd.hpm")
        elseif file_data.sys_id == 2 then
            found_sys_2 = true
            lu.assertEquals(file_data.file_path, "/test/path/HOST_2_Independent_Vrd.hpm")
        end
    end
    lu.assertTrue(found_sys_1)
    lu.assertTrue(found_sys_2)
end

function TestVrdUpgrade:test_on_active_process_file_not_accessible()
    package.loaded['utils.vos'].get_file_accessible = function(path)
        return false
    end

    self.obj:on_active_process(nil, "system1", "INDEPENDENT_VRD")

    lu.assertEquals(#self.validation_calls, 0)
end

function TestVrdUpgrade:test_on_active_process_with_sleep_and_reply()
    self.obj:on_active_process(nil, "system1", "INDEPENDENT_VRD")

    lu.assertTrue(#self.sleep_calls > 0)
    lu.assertEquals(self.sleep_calls[1], 1000)

    local reply_calls = {}
    for _, call in ipairs(self.firmware_mgmt_calls) do
        if call.func == "ActiveProcessReply" then
            table.insert(reply_calls, call)
        end
    end
    lu.assertTrue(#reply_calls > 0)
end

function TestVrdUpgrade:test_get_vrd_load_delegates()
    local result = self.obj:get_vrd_load()
    lu.assertEquals(result, true)
end

function TestVrdUpgrade:test_register_vrd_active_action_delegates()
    local result = self.obj:register_vrd_active_action("INDEPENDENT_VRD")
    lu.assertEquals(result, "register_result")
end

function TestVrdUpgrade:tearDown()
    package.loaded['independent_vrd.chip.vrd_upgrade'] = nil
    package.loaded['mcu.upgrade.upgrade_service.upgrade_service_vrd'] = nil
    package.loaded['utils.vos'] = nil
    package.loaded['utils.core'] = nil
    package.loaded['mc.utils'] = nil
    package.loaded['mc.context'] = nil
    package.loaded['general_hardware.client'] = nil
    package.loaded['mcu.upgrade.defs'] = nil
    package.loaded['mcu.upgrade.valid'] = nil
    package.loaded['common'] = nil
    package.loaded['mc.logging'] = nil
    package.loaded['upgrade_subject.upgrade_subject'] = nil
    package.loaded['general_hardware.fructl'] = nil
    package.loaded['mcu.upgrade.upgrade_service.upgrade_service_comm'] = nil
end
