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

-- 定义 mock 对象（但不设置到 package.loaded，避免文件加载时污染全局环境）
local skynet_sleep_calls = {}
local mock_skynet = {
    sleep = function(ms)
        table.insert(skynet_sleep_calls, {ms = ms})
    end
}

local vos_calls = {}
local mock_vos = {
    get_file_accessible = function(path)
        table.insert(vos_calls, {method = 'get_file_accessible', path = path})
        if path == '/dev/mtd_already_exist' then
            return true
        end
        return false
    end
}

local mock_utils = {
    secure_tar_unzip = function(src, dst, max_size, max_files)
        return true
    end,
    safe_close_file = function(fd, callback)
        if callback then
            callback()
        end
    end
}

local mock_utils_core = {
    stat = function(path)
        if path == '/tmp/large_file.bin' then
            return {st_size = 999999999}
        end
        return {st_size = 1024}
    end
}

local mock_client = {
    PProxyDriverInsmod = function()
        return true
    end,
    PProxyDriverRmmod = function()
        return true
    end
}

local mock_parser_cfg = {
    extract_upgrade_file = function(file_path)
        if file_path == '/tmp/invalid.tar' then
            return nil, nil
        end
        return '/tmp/release/', 'Firmware1'
    end
}

local mock_file_handle = {
    seek = function(self, whence, offset)
        return 0
    end,
    read = function(self, length)
        return string.rep('A', length)
    end
}
local mock_file_sec = {
    open_s = function(path, mode)
        if path == '/tmp/open_failed.bin' then
            return nil, 'open failed'
        end
        return mock_file_handle, nil
    end
}

local mtd_calls = {}
local mock_mtd_api = {
    open = function(device, mode)
        table.insert(mtd_calls, {method = 'open', device = device, mode = mode})
        if device == '/dev/mtd_open_fail' then
            return nil
        end
        return {fd = 123}
    end,
    erase = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'erase', fd = fd, addr = addr, len = len})
        if len == 0 then
            error('erase length is 0')
        end
    end,
    write = function(fd, addr, data, len)
        table.insert(mtd_calls, {method = 'write', fd = fd, addr = addr, len = len})
        if len == 0 then
            error('write length is 0')
        end
    end,
    read = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'read', fd = fd, addr = addr, len = len})
        return string.rep('A', len)
    end,
    close = function(fd)
        table.insert(mtd_calls, {method = 'close', fd = fd})
    end
}

local spi_flash_calls = {}
local mock_spi_flash = {
    insmod_driver = function(driver, param, device, client, func)
        table.insert(spi_flash_calls, {
            method = 'insmod_driver',
            driver = driver,
            param = param,
            device = device
        })
        if driver == 'fail_driver' then
            error('insmod failed')
        end
    end,
    check_device_ready = function(device)
        table.insert(spi_flash_calls, {method = 'check_device_ready', device = device})
        if device == '/dev/mtd_not_ready' then
            error('device not ready')
        end
    end,
    rmmod_driver = function(driver)
        table.insert(spi_flash_calls, {method = 'rmmod_driver', driver = driver})
    end
}

local mock_network_adapter_firmware_objects = {}
local mock_network_adapter_firmware = {
    collection = {
        objects = mock_network_adapter_firmware_objects,
        fold = function(self, func, acc)
            for key, obj in pairs(self.objects) do
                local result = func(acc, obj, key)
                if result ~= nil then
                    return result
                end
            end
            return acc
        end
    }
}

-- 延迟加载被测试模块，在 setUp 中加载
local upgrade_defs
local upgrade_flash

TEST_Upgrade_Flash = {}

function TEST_Upgrade_Flash:setUp()
    -- 设置 package.loaded 中的 mock（只在测试用例执行时设置，避免文件加载时污染全局环境）
    package.loaded['skynet'] = mock_skynet
    package.loaded['mc.logging'] = {
        notice = function() end,
        error = function() end,
        info = function() end,
        maintenance = function() end,
        running = function() end,
        RLOG_INFO = 0,
        RLOG_ERROR = 1
    }
    -- 对于 mc.context，先加载真实模块，然后只覆盖需要 mock 的函数，保留其他函数（如 get_context）
    package.loaded['mc.context'] = nil
    local real_context = require 'mc.context'
    local context_mock = {}
    -- 复制真实模块的所有函数
    for k, v in pairs(real_context) do
        context_mock[k] = v
    end
    -- 覆盖需要 mock 的函数
    context_mock.new = function()
        return {}
    end
    package.loaded['mc.context'] = context_mock
    
    -- 对于 utils.vos，先加载真实模块，然后只覆盖需要 mock 的函数，保留其他函数（如 vos_tick_get）
    package.loaded['utils.vos'] = nil
    local real_vos = require 'utils.vos'
    local vos_mock = {}
    -- 复制真实模块的所有函数
    for k, v in pairs(real_vos) do
        vos_mock[k] = v
    end
    -- 覆盖需要 mock 的函数
    vos_mock.get_file_accessible = function(path)
        table.insert(vos_calls, {method = 'get_file_accessible', path = path})
        if path == '/dev/mtd_already_exist' then
            return true
        end
        return false
    end
    package.loaded['utils.vos'] = vos_mock
    
    package.loaded['mc.utils'] = mock_utils
    package.loaded['utils.core'] = mock_utils_core
    package.loaded['client'] = mock_client
    package.loaded['upgrade.parser_cfg'] = mock_parser_cfg
    package.loaded['utils.file'] = mock_file_sec
    package.loaded['mtd.drv'] = mock_mtd_api
    package.loaded['libmgmt_protocol.bios.infrastructure.spi_flash'] = mock_spi_flash
    package.loaded['upgrade.network_adapter_firmware_object'] = mock_network_adapter_firmware
    
    -- 清除被测试模块的缓存，确保使用最新的 mock
    package.loaded['upgrade.upgrade_flash'] = nil
    package.loaded['upgrade.upgrade_defs'] = nil
    package.loaded['upgrade.upgrade_process'] = nil
    package.loaded['upgrade.network_adapter_upgrade'] = nil
    package.loaded['upgrade.active_process'] = nil
    
    -- 加载被测试模块（使用最新的 mock）
    upgrade_defs = require 'upgrade.upgrade_defs'
    upgrade_flash = require 'upgrade.upgrade_flash'
    
    -- Reset call records
    vos_calls = {}
    mtd_calls = {}
    spi_flash_calls = {}
    skynet_sleep_calls = {}
    
    -- Reset network adapter firmware objects
    for k in pairs(mock_network_adapter_firmware_objects) do
        mock_network_adapter_firmware_objects[k] = nil
    end
    
    -- Reset mock functions
    -- 更新 package.loaded['utils.vos'] 中的 get_file_accessible 函数
    package.loaded['utils.vos'].get_file_accessible = function(path)
        table.insert(vos_calls, {method = 'get_file_accessible', path = path})
        if path == '/dev/mtd_already_exist' then
            return true
        end
        return false
    end
    
    mock_utils_core.stat = function(path)
        if path == '/tmp/large_file.bin' then
            return {st_size = 999999999}
        end
        return {st_size = 1024}
    end
    
    mock_mtd_api.read = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'read', fd = fd, addr = addr, len = len})
        return string.rep('A', len)
    end
    
    -- Create mock self object
    self.mock_self = {
        cfgs = {
            Firmware1 = {
                name = 'Firmware1',
                component_id = 1,
                component_idex = 2
            }
        },
        upgrade_list = {}
    }
    
    -- Create mock upgrade object
    self.mock_upgrade_obj = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 1,
        DefaultRoute = 0,
        Route = 0
    }
end

function TEST_Upgrade_Flash:tearDown()
    -- 重置调用记录
    vos_calls = {}
    mtd_calls = {}
    spi_flash_calls = {}
    skynet_sleep_calls = {}
    
    -- 只清除 test_upgrade 特有的模块，不清除基础模块（如 skynet、mc.logging 等）
    -- 基础模块可能被其他测试使用，清除它们会导致其他测试失败
    package.loaded['upgrade.parser_cfg'] = nil
    package.loaded['upgrade.network_adapter_firmware_object'] = nil
    package.loaded['upgrade.upgrade_flash'] = nil
    package.loaded['upgrade.upgrade_defs'] = nil
    package.loaded['upgrade.upgrade_process'] = nil
    package.loaded['upgrade.network_adapter_upgrade'] = nil
    package.loaded['upgrade.active_process'] = nil
end

-- 测试模块存在
function TEST_Upgrade_Flash:test_module_exists()
    lu.assertNotNil(upgrade_flash)
    lu.assertTrue(type(upgrade_flash) == 'table')
end

-- 测试set_network_adapter_firmware_route - nil对象
function TEST_Upgrade_Flash:test_set_route_nil_object()
    lu.assertError(function()
        upgrade_flash.set_network_adapter_firmware_route(nil, 1)
    end)
end

-- 测试set_network_adapter_firmware_route - 成功场景
function TEST_Upgrade_Flash:test_set_route_success()
    local obj = {Route = 0}
    
    upgrade_flash.set_network_adapter_firmware_route(obj, 1)
    
    -- 应该设置成功
    lu.assertEquals(obj.Route, 1)
    lu.assertEquals(#skynet_sleep_calls, 0) -- 成功不需要重试
end

-- 测试set_network_adapter_firmware_route - 需要重试
function TEST_Upgrade_Flash:test_set_route_retry()
    local attempt = 0
    local obj = {}
    setmetatable(obj, {
        __newindex = function(t, k, v)
            attempt = attempt + 1
            if attempt < 3 then
                error('route set failed')
            else
                rawset(t, k, v)
            end
        end
    })
    
    upgrade_flash.set_network_adapter_firmware_route(obj, 1)
    
    -- 应该重试了2次，第3次成功
    lu.assertEquals(attempt, 3)
    lu.assertTrue(#skynet_sleep_calls >= 2) -- 至少sleep了2次
end

-- 测试set_network_adapter_firmware_route - 重试失败
function TEST_Upgrade_Flash:test_set_route_retry_failed()
    local obj = {}
    setmetatable(obj, {
        __newindex = function(t, k, v)
            error('route set always failed')
        end
    })
    
    lu.assertError(function()
        upgrade_flash.set_network_adapter_firmware_route(obj, 1)
    end)
    
    -- 应该重试了SET_ROUTE_RETRY次
    lu.assertEquals(#skynet_sleep_calls, upgrade_defs.SET_ROUTE_RETRY)
end

-- 测试network_adapter_upgrade_flash - 成功场景
function TEST_Upgrade_Flash:test_upgrade_flash_success()
    -- 添加匹配的firmware对象
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该成功
    lu.assertEquals(ret, upgrade_defs.RET.OK)
    
    -- 验证调用顺序
    -- 1. 应该调用insmod_driver
    local has_insmod = false
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'insmod_driver' then
            has_insmod = true
        end
    end
    lu.assertTrue(has_insmod)
    
    -- 2. 应该调用mtd操作
    local has_open = false
    local has_erase = false
    local has_write = false
    local has_read = false
    local has_close = false
    for _, call in ipairs(mtd_calls) do
        if call.method == 'open' then
            has_open = true
        elseif call.method == 'erase' then
            has_erase = true
        elseif call.method == 'write' then
            has_write = true
        elseif call.method == 'read' then
            has_read = true
        elseif call.method == 'close' then
            has_close = true
        end
    end
    lu.assertTrue(has_open)
    lu.assertTrue(has_erase)
    lu.assertTrue(has_write)
    lu.assertTrue(has_read)
    lu.assertTrue(has_close)
    
    -- 3. 应该调用rmmod_driver
    local has_rmmod = false
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'rmmod_driver' then
            has_rmmod = true
        end
    end
    lu.assertTrue(has_rmmod)
end

-- 测试network_adapter_upgrade_flash - 没有升级列表
function TEST_Upgrade_Flash:test_upgrade_flash_no_upgrade_list()
    -- 不添加匹配的firmware对象
    mock_network_adapter_firmware_objects['test'] = {
        ComponentID = 999,
        ComponentIDEx = 999
    }
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
end

-- 测试network_adapter_upgrade_flash - 设备已经存在
function TEST_Upgrade_Flash:test_upgrade_flash_device_already_exist()
    -- 添加匹配的firmware对象
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    -- Mock设备已经存在
    local original_get_accessible = package.loaded['utils.vos'].get_file_accessible
    package.loaded['utils.vos'].get_file_accessible = function(path)
        table.insert(vos_calls, {method = 'get_file_accessible', path = path})
        return true -- 设备已经存在
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复原始函数
    package.loaded['utils.vos'].get_file_accessible = original_get_accessible
end

-- 测试network_adapter_upgrade_flash - 驱动加载失败
function TEST_Upgrade_Flash:test_upgrade_flash_insmod_failed()
    -- Mock insmod失败
    local original_insmod = mock_spi_flash.insmod_driver
    mock_spi_flash.insmod_driver = function(driver, param, device, client, func)
        table.insert(spi_flash_calls, {method = 'insmod_driver'})
        error('insmod failed')
    end
    
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 应该尝试清理（rmmod）
    local has_rmmod = false
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'rmmod_driver' then
            has_rmmod = true
        end
    end
    lu.assertTrue(has_rmmod)
    
    -- 恢复原始函数
    mock_spi_flash.insmod_driver = original_insmod
end

-- 测试network_adapter_upgrade_flash - 打开文件失败
function TEST_Upgrade_Flash:test_upgrade_flash_open_file_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    -- Mock文件打开失败
    local original_open = mock_file_sec.open_s
    mock_file_sec.open_s = function(path, mode)
        return nil, 'open failed'
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 应该清理资源
    local has_close = false
    local has_rmmod = false
    for _, call in ipairs(mtd_calls) do
        if call.method == 'close' then
            has_close = true
        end
    end
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'rmmod_driver' then
            has_rmmod = true
        end
    end
    lu.assertTrue(has_close)
    lu.assertTrue(has_rmmod)
    
    -- 恢复原始函数
    mock_file_sec.open_s = original_open
end

-- 测试network_adapter_upgrade_flash - 文件大小超过flash大小
function TEST_Upgrade_Flash:test_upgrade_flash_file_too_large()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    -- Mock文件大小超过flash大小
    local original_stat = mock_utils_core.stat
    mock_utils_core.stat = function(path)
        return {st_size = 999999999} -- 超大文件
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复原始函数
    mock_utils_core.stat = original_stat
end

-- 测试network_adapter_upgrade_flash - 读回校验失败
function TEST_Upgrade_Flash:test_upgrade_flash_verify_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    -- Mock读回数据不一致
    local original_read = mock_mtd_api.read
    mock_mtd_api.read = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'read', fd = fd, addr = addr, len = len})
        return string.rep('B', len) -- 返回不一样的数据
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 应该清理资源
    local has_close = false
    local has_rmmod = false
    for _, call in ipairs(mtd_calls) do
        if call.method == 'close' then
            has_close = true
        end
    end
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'rmmod_driver' then
            has_rmmod = true
        end
    end
    lu.assertTrue(has_close)
    lu.assertTrue(has_rmmod)
    
    -- 恢复原始函数
    mock_mtd_api.read = original_read
end

-- 测试network_adapter_upgrade_flash - 多个网卡升级
function TEST_Upgrade_Flash:test_upgrade_flash_multiple_adapters()
    -- 添加多个匹配的firmware对象
    mock_network_adapter_firmware_objects['test1'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 1,
        DefaultRoute = 0,
        Route = 0
    }
    mock_network_adapter_firmware_objects['test2'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 2,
        DefaultRoute = 0,
        Route = 0
    }
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该成功
    lu.assertEquals(ret, upgrade_defs.RET.OK)
    
    -- 应该有两次insmod/rmmod操作
    local insmod_count = 0
    local rmmod_count = 0
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'insmod_driver' then
            insmod_count = insmod_count + 1
        elseif call.method == 'rmmod_driver' then
            rmmod_count = rmmod_count + 1
        end
    end
    lu.assertEquals(insmod_count, 2)
    lu.assertEquals(rmmod_count, 2)
end

-- 测试network_adapter_upgrade_flash - 第二个网卡升级失败
function TEST_Upgrade_Flash:test_upgrade_flash_second_adapter_failed()
    -- 添加两个匹配的firmware对象
    mock_network_adapter_firmware_objects['test1'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 1,
        DefaultRoute = 0,
        Route = 0
    }
    mock_network_adapter_firmware_objects['test2'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 2,
        DefaultRoute = 0,
        Route = 0
    }
    
    -- Mock第二次insmod失败
    local insmod_count = 0
    local original_insmod = mock_spi_flash.insmod_driver
    mock_spi_flash.insmod_driver = function(driver, param, device, client, func)
        table.insert(spi_flash_calls, {method = 'insmod_driver'})
        insmod_count = insmod_count + 1
        if insmod_count > 1 then
            error('second insmod failed')
        end
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复原始函数
    mock_spi_flash.insmod_driver = original_insmod
end

-- 测试extract_upgrade_file失败
function TEST_Upgrade_Flash:test_upgrade_flash_extract_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/invalid.tar'
    )
    
    -- 应该返回错误
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
end

-- 测试set_network_adapter_firmware_route - upgrade_obj为nil
function TEST_Upgrade_Flash:test_set_route_nil_obj()
    local ok, err = pcall(function()
        upgrade_flash.set_network_adapter_firmware_route(nil, 0x01)
    end)
    
    lu.assertFalse(ok)
    lu.assertNotNil(err)
end

-- 测试upgrade_flash - erase失败
function TEST_Upgrade_Flash:test_upgrade_flash_erase_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_erase = mock_mtd_api.erase
    mock_mtd_api.erase = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'erase', fd = fd, addr = addr, len = len})
        error('erase failed')
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_mtd_api.erase = original_erase
end

-- 测试upgrade_flash - write失败
function TEST_Upgrade_Flash:test_upgrade_flash_write_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_write = mock_mtd_api.write
    mock_mtd_api.write = function(fd, addr, data, len)
        table.insert(mtd_calls, {method = 'write', fd = fd, addr = addr, len = len})
        error('write failed')
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_mtd_api.write = original_write
end

-- 测试upgrade_flash - read回读校验失败
function TEST_Upgrade_Flash:test_upgrade_flash_verify_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_read = mock_mtd_api.read
    mock_mtd_api.read = function(fd, addr, len)
        table.insert(mtd_calls, {method = 'read', fd = fd, addr = addr, len = len})
        return string.rep('B', len)  -- 返回不同的数据，导致校验失败
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_mtd_api.read = original_read
end

-- 测试upgrade_flash - mtd_api.close失败但不影响结果
function TEST_Upgrade_Flash:test_upgrade_flash_close_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_close = mock_mtd_api.close
    local close_called = false
    mock_mtd_api.close = function(fd)
        close_called = true
        table.insert(mtd_calls, {method = 'close', fd = fd})
        error('close failed')  -- close失败
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- close失败会导致整体失败
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    lu.assertTrue(close_called)
    
    -- 恢复
    mock_mtd_api.close = original_close
end

-- 测试upgrade_flash - rmmod失败
function TEST_Upgrade_Flash:test_upgrade_flash_rmmod_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_rmmod = mock_spi_flash.rmmod_driver
    mock_spi_flash.rmmod_driver = function(driver, device, client, func)
        table.insert(spi_flash_calls, {method = 'rmmod_driver'})
        error('rmmod failed')
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    -- rmmod失败会导致整体失败
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_spi_flash.rmmod_driver = original_rmmod
end

-- 测试upgrade_flash - check_device_ready失败
function TEST_Upgrade_Flash:test_upgrade_flash_check_device_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_check = mock_spi_flash.check_device_ready
    mock_spi_flash.check_device_ready = function(device)
        table.insert(spi_flash_calls, {method = 'check_device_ready'})
        error('device not ready')
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_spi_flash.check_device_ready = original_check
end

-- 测试upgrade_flash - 文件seek失败
function TEST_Upgrade_Flash:test_upgrade_flash_file_seek_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_open_s = mock_file_sec.open_s
    mock_file_sec.open_s = function(path, mode)
        local mock_file = {
            seek = function(self, whence, offset)
                error('seek failed')
            end,
            read = function(self, length)
                return string.rep('A', length)
            end
        }
        return mock_file, nil
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_file_sec.open_s = original_open_s
end

-- 测试upgrade_flash - 文件read失败
function TEST_Upgrade_Flash:test_upgrade_flash_file_read_failed()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local original_open_s = mock_file_sec.open_s
    mock_file_sec.open_s = function(path, mode)
        local mock_file = {
            seek = function(self, whence, offset)
                return 0
            end,
            read = function(self, length)
                error('read failed')
            end
        }
        return mock_file, nil
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.ERR)
    
    -- 恢复
    mock_file_sec.open_s = original_open_s
end

-- 测试upgrade_flash - secure_tar_unzip成功（覆盖该调用）
function TEST_Upgrade_Flash:test_upgrade_flash_unzip_success()
    mock_network_adapter_firmware_objects['test'] = self.mock_upgrade_obj
    
    local unzip_called = false
    local original_unzip = mock_utils.secure_tar_unzip
    mock_utils.secure_tar_unzip = function(src, dst, max_size, max_files)
        unzip_called = true
        return true
    end
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.OK)
    lu.assertTrue(unzip_called)
    
    -- 恢复
    mock_utils.secure_tar_unzip = original_unzip
end

-- 测试upgrade_flash - 多个网卡全部升级成功
function TEST_Upgrade_Flash:test_upgrade_flash_three_adapters_all_success()
    -- 添加三个匹配的firmware对象
    mock_network_adapter_firmware_objects['test1'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 1,
        DefaultRoute = 0,
        Route = 0
    }
    mock_network_adapter_firmware_objects['test2'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 2,
        DefaultRoute = 0,
        Route = 0
    }
    mock_network_adapter_firmware_objects['test3'] = {
        ComponentID = 1,
        ComponentIDEx = 2,
        FlashSizeKiB = 2048,
        FirmwareRoute = 3,
        DefaultRoute = 0,
        Route = 0
    }
    
    local ret = upgrade_flash.network_adapter_upgrade_flash(
        self.mock_self,
        '/tmp/firmware.tar'
    )
    
    lu.assertEquals(ret, upgrade_defs.RET.OK)
    
    -- 验证所有网卡都被处理
    local insmod_count = 0
    for _, call in ipairs(spi_flash_calls) do
        if call.method == 'insmod_driver' then
            insmod_count = insmod_count + 1
        end
    end
    lu.assertEquals(insmod_count, 3)
end

return TEST_Upgrade_Flash
