-- Copyright (c) 2026 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 c_network_adapter = require 'device.class.network_adapter'
local skynet = require 'skynet'
local mdb = require 'mc.mdb'
local c_object_manage = require 'mc.orm.object_manage'
local comm_defs = require 'device.class.nic_mgmt.comm_defs'
local client = require 'network_adapter.client'
local ctx = require 'mc.context'
local network_adapter_app = require 'network_adapter_app'
local mdb_obj_mgmt = require 'mc.mdb.object_manage'

TestFirmwareInfo = {}

function TestFirmwareInfo:setUp()
    -- 每个测试用例自己管理 mock，这里不需要全局设置
end

function TestFirmwareInfo:tearDown()
end

-- ========== firmware_objs_mgmt 测试 ==========
TestFirmwareObjsMgmt = {}

function TestFirmwareObjsMgmt:test_init()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    lu.assertNotNil(firmware_objs_mgmt.upgradable_cards)
    lu.assertEquals(firmware_objs_mgmt.upgradable_cards, {})
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

function TestFirmwareObjsMgmt:test_on_add_firmware_object_class_mismatch()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_collection = c_network_adapter.collection
    local fold_called = false
    c_network_adapter.collection = { fold = function() fold_called = true end }

    firmware_objs_mgmt:on_add_firmware_object('OtherClass', nil, 1)
    lu.assertFalse(fold_called)

    c_network_adapter.collection = old_collection
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

function TestFirmwareObjsMgmt:test_on_add_firmware_object_found_first_try()
    local old_client_loaded = package.loaded['network_adapter.client']
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_collection = c_network_adapter.collection
    local old_client_add = test_client.PFirmwareInventoryFirmwareInventoryAdd
    local old_ctx_new = ctx.new

    ctx.new = function() return {} end
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function() return true end

    local register_callback_called = false
    local captured_prop_cb
    local mock_network_adapter = {
        NodeId = 'PCIeCard1(SP566)',
        ObjectIdentifier = { nil, nil, nil, 5 },
        register_firmware_info_flag = false,
        on_property_changed = {
            on = function(self, cb)
                register_callback_called = true
                captured_prop_cb = cb
                return { disconnect = function() end }
            end
        }
    }
    c_network_adapter.collection = {
        fold = function(self, fn)
            fn(nil, mock_network_adapter)
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)

    lu.assertTrue(register_callback_called)
    lu.assertNotNil(firmware_objs_mgmt.upgradable_cards[5])
    lu.assertEquals(firmware_objs_mgmt.upgradable_cards[5].firmware_id, 'PCIeCard_1_SP566')

    -- 覆盖 register_firmware_info: prop_name ~= 'FirmwareVersion' 分支
    lu.assertNotNil(captured_prop_cb)
    captured_prop_cb('OtherProp', 'value')
    lu.assertFalse(mock_network_adapter.register_firmware_info_flag)

    -- 覆盖 register_firmware_info: register_firmware_info_flag false -> register_firmware_info_by_firmware_version
    captured_prop_cb('FirmwareVersion', '2.0.0')
    lu.assertTrue(mock_network_adapter.register_firmware_info_flag)

    c_network_adapter.collection = old_collection
    test_client.PFirmwareInventoryFirmwareInventoryAdd = old_client_add
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

function TestFirmwareObjsMgmt:test_on_add_firmware_object_found_after_retry()
    local old_client_loaded = package.loaded['network_adapter.client']
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_collection = c_network_adapter.collection
    local old_client_add = test_client.PFirmwareInventoryFirmwareInventoryAdd
    local old_ctx_new = ctx.new

    local fold_count = 0
    local mock_network_adapter = {
        NodeId = 'PCIeCard1(SP566)',
        ObjectIdentifier = { nil, nil, nil, 7 },
        register_firmware_info_flag = false,
        on_property_changed = { on = function() return { disconnect = function() end } end }
    }
    ctx.new = function() return {} end
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function() return true end
    c_network_adapter.collection = {
        fold = function(self, fn)
            fold_count = fold_count + 1
            if fold_count >= 2 then
                fn(nil, mock_network_adapter)
            end
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 7)

    lu.assertEquals(fold_count, 2)
    lu.assertNotNil(firmware_objs_mgmt.upgradable_cards[7])

    c_network_adapter.collection = old_collection
    test_client.PFirmwareInventoryFirmwareInventoryAdd = old_client_add
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

function TestFirmwareObjsMgmt:test_on_add_firmware_object_not_found_after_retries()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_collection = c_network_adapter.collection

    local fold_count = 0
    c_network_adapter.collection = {
        fold = function(self, fn)
            fold_count = fold_count + 1
            fn(nil, { ObjectIdentifier = { nil, nil, nil, 999 } })
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 888)

    lu.assertEquals(fold_count, 30)
    lu.assertNil(firmware_objs_mgmt.upgradable_cards[888])

    c_network_adapter.collection = old_collection
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

function TestFirmwareObjsMgmt:test_on_delete_firmware_object_class_mismatch()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = { firmware_id = 'test_id' }

    local old_client_delete = client.PFirmwareInventoryFirmwareInventoryDelete
    local old_ctx_new = ctx.new
    local delete_called = false
    ctx.new = function() return {} end
    client.PFirmwareInventoryFirmwareInventoryDelete = function()
        delete_called = true;
        return true
    end

    firmware_objs_mgmt:on_delete_firmware_object('OtherClass', nil, 5)
    lu.assertFalse(delete_called)
    lu.assertNotNil(firmware_objs_mgmt.upgradable_cards[5])

    client.PFirmwareInventoryFirmwareInventoryDelete = old_client_delete
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

function TestFirmwareObjsMgmt:test_on_delete_firmware_object_with_firmware_id()
    local old_client_loaded = package.loaded['network_adapter.client']
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = { firmware_id = 'PCIeCard_1_SP566' }

    local old_client_delete = test_client.PFirmwareInventoryFirmwareInventoryDelete
    local old_ctx_new = ctx.new
    local delete_called = false
    ctx.new = function() return {} end
    test_client.PFirmwareInventoryFirmwareInventoryDelete = function(_, _, firmware_id)
        delete_called = true
        lu.assertEquals(firmware_id, 'PCIeCard_1_SP566')
        return true
    end

    firmware_objs_mgmt:on_delete_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)

    lu.assertTrue(delete_called)
    lu.assertNil(firmware_objs_mgmt.upgradable_cards[5])

    test_client.PFirmwareInventoryFirmwareInventoryDelete = old_client_delete
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

function TestFirmwareObjsMgmt:test_on_delete_firmware_object_no_firmware_id()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = {}

    local old_fork = skynet.fork
    local old_client_delete = client.PFirmwareInventoryFirmwareInventoryDelete
    local old_ctx_new = ctx.new
    local delete_called = false
    skynet.fork = function(cb) cb() end
    ctx.new = function() return {} end
    client.PFirmwareInventoryFirmwareInventoryDelete = function()
        delete_called = true;
        return true
    end

    firmware_objs_mgmt:on_delete_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)

    lu.assertFalse(delete_called)
    lu.assertNil(firmware_objs_mgmt.upgradable_cards[5])

    skynet.fork = old_fork
    client.PFirmwareInventoryFirmwareInventoryDelete = old_client_delete
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

function TestFirmwareObjsMgmt:test_on_delete_firmware_object_position_not_registered()
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_client_delete = client.PFirmwareInventoryFirmwareInventoryDelete
    local delete_called = false
    client.PFirmwareInventoryFirmwareInventoryDelete = function()
        delete_called = true;
        return true
    end

    firmware_objs_mgmt:on_delete_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 99)

    lu.assertFalse(delete_called)

    client.PFirmwareInventoryFirmwareInventoryDelete = old_client_delete
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
end

-- 覆盖 retry_update_firmware_info、get_firmware_id 回退、upgradable_cards 已存在、register_firmware_info_flag 为 true 时走 retry_update
function TestFirmwareObjsMgmt:test_on_add_retry_update_and_get_firmware_id_fallback()
    local old_client_loaded = package.loaded['network_adapter.client']
    local old_mdb_get = mdb.get_cached_object
    local old_get_instance = c_object_manage.get_instance
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = {}

    local old_collection = c_network_adapter.collection
    local old_client_add = test_client.PFirmwareInventoryFirmwareInventoryAdd
    local old_ctx_new = ctx.new

    ctx.new = function() return {} end
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function() return true end
    c_object_manage.get_instance = function() return { bus = {} } end

    local mock_obj = { Version = '' }
    mdb.get_cached_object = function() return mock_obj end

    local captured_prop_cb
    local mock_network_adapter = {
        NodeId = 'EthDevice1',
        ObjectIdentifier = { nil, nil, nil, 5 },
        register_firmware_info_flag = true,
        on_property_changed = {
            on = function(self, cb)
                captured_prop_cb = cb
                return { disconnect = function() end }
            end
        }
    }
    c_network_adapter.collection = {
        fold = function(self, fn)
            fn(nil, mock_network_adapter)
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)

    lu.assertEquals(firmware_objs_mgmt.upgradable_cards[5].firmware_id, 'EthDevice1')
    captured_prop_cb('FirmwareVersion', '3.0.0')
    lu.assertEquals(mock_obj.Version, '3.0.0')

    mdb.get_cached_object = old_mdb_get
    c_object_manage.get_instance = old_get_instance
    c_network_adapter.collection = old_collection
    test_client.PFirmwareInventoryFirmwareInventoryAdd = old_client_add
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

-- 覆盖 retry_update_firmware_info 重试后成功、重试耗尽失败
function TestFirmwareObjsMgmt:test_retry_update_firmware_info_retry_and_fail()
    local old_client_loaded = package.loaded['network_adapter.client']
    local old_mdb_get = mdb.get_cached_object
    local old_get_instance = c_object_manage.get_instance
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = { firmware_id = 'test_id' }

    local old_collection = c_network_adapter.collection
    local old_client_add = test_client.PFirmwareInventoryFirmwareInventoryAdd
    local old_ctx_new = ctx.new

    ctx.new = function() return {} end
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function() return true end
    c_object_manage.get_instance = function() return { bus = {} } end

    local get_cached_count = 0
    local mock_obj = { Version = '' }
    mdb.get_cached_object = function()
        get_cached_count = get_cached_count + 1
        if get_cached_count < 2 then
            error('not found')
        end
        return mock_obj
    end

    local captured_prop_cb
    local mock_network_adapter = {
        NodeId = 'PCIeCard1(SP566)',
        ObjectIdentifier = { nil, nil, nil, 5 },
        register_firmware_info_flag = true,
        on_property_changed = {
            on = function(self, cb)
                captured_prop_cb = cb
                return { disconnect = function() end }
            end
        }
    }
    c_network_adapter.collection = {
        fold = function(self, fn)
            fn(nil, mock_network_adapter)
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)
    captured_prop_cb('FirmwareVersion', '4.0.0')
    lu.assertEquals(get_cached_count, 2)
    lu.assertEquals(mock_obj.Version, '4.0.0')

    get_cached_count = 0
    mdb.get_cached_object = function()
        get_cached_count = get_cached_count + 1
        error('not found')
    end
    mock_network_adapter.register_firmware_info_flag = true
    captured_prop_cb('FirmwareVersion', '5.0.0')
    lu.assertEquals(get_cached_count, 3)

    mdb.get_cached_object = old_mdb_get
    c_object_manage.get_instance = old_get_instance
    c_network_adapter.collection = old_collection
    test_client.PFirmwareInventoryFirmwareInventoryAdd = old_client_add
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

-- 覆盖 retry_register_firmware_info 重试后成功、重试耗尽失败、param 默认值
function TestFirmwareObjsMgmt:test_retry_register_firmware_info_retry_and_fail()
    local old_client_loaded = package.loaded['network_adapter.client']
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()

    local old_collection = c_network_adapter.collection
    local old_client_add = test_client.PFirmwareInventoryFirmwareInventoryAdd
    local old_ctx_new = ctx.new

    ctx.new = function() return {} end

    local add_count = 0
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function(_, _, param)
        add_count = add_count + 1
        if add_count < 3 then
            return false, 'err'
        end
        return true
    end

    local captured_prop_cb
    local mock_network_adapter = {
        NodeId = 'PCIeCard1(SP566)',
        ObjectIdentifier = { nil, nil, nil, 5 },
        register_firmware_info_flag = false,
        Manufacturer = nil,
        on_property_changed = {
            on = function(self, cb)
                captured_prop_cb = cb
                return { disconnect = function() end }
            end
        }
    }
    c_network_adapter.collection = {
        fold = function(self, fn)
            fn(nil, mock_network_adapter)
        end
    }

    firmware_objs_mgmt:on_add_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)
    captured_prop_cb('FirmwareVersion', '1.0.0')
    lu.assertEquals(add_count, 3)
    lu.assertTrue(mock_network_adapter.register_firmware_info_flag)

    add_count = 0
    test_client.PFirmwareInventoryFirmwareInventoryAdd = function()
        add_count = add_count + 1
        return false, 'err'
    end
    mock_network_adapter.register_firmware_info_flag = false
    captured_prop_cb('FirmwareVersion', '2.0.0')
    lu.assertEquals(add_count, 3)
    lu.assertFalse(mock_network_adapter.register_firmware_info_flag)

    c_network_adapter.collection = old_collection
    test_client.PFirmwareInventoryFirmwareInventoryAdd = old_client_add
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

-- 覆盖 retry_unregister_firmware_info 重试后成功、重试耗尽失败
function TestFirmwareObjsMgmt:test_retry_unregister_firmware_info_retry_and_fail()
    local old_client_loaded = package.loaded['network_adapter.client']
    package.loaded['network_adapter.client'] = nil
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['mc.singleton'] = function(cls) return cls end
    package.loaded['skynet'] = {
        fork = function(cb) cb() end,
        fork_once = function(cb) cb() end,
        sleep = function() end
    }
    package.loaded['mc.logging'] = {
        warn = function() end,
        notice = function() end,
        error = function() end,
        info = function() end
    }
    local test_client = require 'network_adapter.client'
    local firmware_objs_mgmt = require 'upgrade.firmware_objs_mgmt'
    firmware_objs_mgmt:init()
    firmware_objs_mgmt.upgradable_cards[5] = { firmware_id = 'unreg_id' }

    local old_client_delete = test_client.PFirmwareInventoryFirmwareInventoryDelete
    local old_ctx_new = ctx.new

    ctx.new = function() return {} end

    local delete_count = 0
    test_client.PFirmwareInventoryFirmwareInventoryDelete = function(_, _, id)
        delete_count = delete_count + 1
        if delete_count < 2 then
            return false, 'err'
        end
        return true
    end

    firmware_objs_mgmt:on_delete_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 5)
    lu.assertEquals(delete_count, 2)
    lu.assertNil(firmware_objs_mgmt.upgradable_cards[5])

    firmware_objs_mgmt.upgradable_cards[6] = { firmware_id = 'fail_id' }
    delete_count = 0
    test_client.PFirmwareInventoryFirmwareInventoryDelete = function()
        delete_count = delete_count + 1
        return false, 'err'
    end

    firmware_objs_mgmt:on_delete_firmware_object(comm_defs.UPGRADABLE_FIRMWARE_OBJ, nil, 6)
    lu.assertEquals(delete_count, 30)

    test_client.PFirmwareInventoryFirmwareInventoryDelete = old_client_delete
    ctx.new = old_ctx_new
    package.loaded['upgrade.firmware_objs_mgmt'] = nil
    package.loaded['network_adapter.client'] = old_client_loaded
end

-- ========== network_adapter_app.register_object_mgmt 测试 ==========
TestRegisterObjectMgmt = {}

function TestRegisterObjectMgmt:test_register_object_mgmt()
    local old_on_add_object = mdb_obj_mgmt.on_add_object
    local old_on_delete_object = mdb_obj_mgmt.on_delete_object

    local captured_add_bus, captured_add_cb
    local captured_delete_bus, captured_delete_cb

    mdb_obj_mgmt.on_add_object = function(bus, cb)
        captured_add_bus = bus
        captured_add_cb = cb
    end
    mdb_obj_mgmt.on_delete_object = function(bus, cb)
        captured_delete_bus = bus
        captured_delete_cb = cb
    end

    local mock_bus = {}
    local add_called, add_args = false, nil
    local delete_called, delete_args = false, nil
    local app_instance = {
        bus = mock_bus,
        upgrade_mgmt = {
            on_add_firmware_object = function(self, class_name, object, position)
                add_called = true
                add_args = { class_name, object, position }
            end,
            on_delete_firmware_object = function(self, class_name, object, position)
                delete_called = true
                delete_args = { class_name, object, position }
            end
        }
    }
    setmetatable(app_instance, { __index = network_adapter_app })

    app_instance:register_object_mgmt()

    lu.assertEquals(captured_add_bus, mock_bus)
    lu.assertNotNil(captured_add_cb)
    lu.assertEquals(captured_delete_bus, mock_bus)
    lu.assertNotNil(captured_delete_cb)

    local test_class, test_obj, test_pos = 'TestClass', { id = 1 }, 5
    captured_add_cb(test_class, test_obj, test_pos)
    lu.assertTrue(add_called)
    lu.assertNotNil(add_args)
    lu.assertEquals(add_args[1], test_class)
    lu.assertEquals(add_args[2], test_obj)
    lu.assertEquals(add_args[3], test_pos)

    captured_delete_cb(test_class, test_obj, test_pos)
    lu.assertTrue(delete_called)
    lu.assertNotNil(delete_args)
    lu.assertEquals(delete_args[1], test_class)
    lu.assertEquals(delete_args[2], test_obj)
    lu.assertEquals(delete_args[3], test_pos)

    mdb_obj_mgmt.on_add_object = old_on_add_object
    mdb_obj_mgmt.on_delete_object = old_on_delete_object
end