-- 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 ncsi_core = require 'ncsi.ncsi_core'
local ncsi_comm = require 'ncsi.ncsi_comm'
local datas = require 'network_adapter.datas'
local mc_utils = require 'mc.utils'
local c_network_adapter_db = require 'network_adapter.db'
local ncsi_service = require 'ncsi.ncsi_service'
local signal = require 'mc.signal'
local c_ncsi_nc_info = require 'ncsi.ncsi_ncinfo'
local c_ncsi_info = require 'ncsi.ncsi_info'
TEST_ncsi_eth = {}
local cwd = mc_utils.realpath('.')

function TEST_ncsi_eth:test_ncsi_clib_call()
    ret = ncsi_core.ncsi_get_log(1,"eth:0","",1)
    lu.assertEquals(ret, false)
end

function TEST_ncsi_eth:setUp()
    local test_db_path = cwd .. '/network_adapter.test.db'
    mc_utils.remove_file(test_db_path)
    self.database = c_network_adapter_db(test_db_path, datas)
    ncsi_comm.ncsi_wait_eth_interface_up = function ()
        return false
    end
end

function TEST_ncsi_eth:tearDown()
    local test_db_path = cwd .. '/network_adapter.test.db'
    mc_utils.remove_file(test_db_path)
    self.database.db:close()
end

function TEST_ncsi_eth:test_ncsi_port_basic_init()
    ncsi_comm.get_instance():ncsi_port_basic_init(self.database, 0, 'eth0')
    ncsi_comm.get_instance().ncsi_channel_cnt = 1
    ncsi_comm.get_instance():ncsi_basic_init(self.database)
    local ncsi_channel_cnt = ncsi_core.get_ncsi_channel_cnt()
    lu.assertEquals(ncsi_channel_cnt, 4)
end

function TEST_ncsi_eth:test_set_ncsi_ethid_task()
    local NcsiNCInfo = {
        Id = 1,
        NcsiEthId = 1,
        __rowid = 1
    }
    c_ncsi_nc_info.db = self.database
    ncsi_service.db = self.database
    c_ncsi_nc_info.collection:add_object(NcsiNCInfo)
    local ret = pcall(ncsi_service.set_ncsi_ethid_task, ncsi_service)
    lu.assertEquals(ret, true)
    local nc_info = ncsi_service.db:select(ncsi_service.db.NcsiNCInfo):first()
    lu.assertEquals(nc_info.EthId, 1)
end

function TEST_ncsi_eth:test_register_signal_change_method()
    ncsi_service.ncsi_info = {
        sig_multi_vlan_added = signal.new(),
        sig_eth_intf_add = signal.new(),
        sig_vlan_info_changed = signal.new(),
        sig_mac_addr_changed = signal.new(),
        ncsi_info_init = function ()
            return nil
        end,
        ncsi_uptade_link_status = function ()
            return nil
        end
    }
    ncsi_service.ncsi_comm = {
        sig_relink_ncsi_port = signal.new(),
        sig_update_link_status = signal.new(),
        ncsi_sync_eth_mac = function ()
            return nil
        end,
        ncsi_init = function ()
            return nil
        end,
        ncsi_set_vlan_info = function ()
            return nil
        end,
        ncsi_basic_init = function ()
            return nil
        end,
        ncsi_update_multi_vlan = function ()
            return nil
        end
    }
    local ok = pcall(ncsi_service.register_signal_change_method, ncsi_service)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_info.sig_mac_addr_changed.emit,ncsi_service.ncsi_info.sig_mac_addr_changed)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_info.sig_eth_intf_add.emit,ncsi_service.ncsi_info.sig_eth_intf_add)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_info.sig_vlan_info_changed.emit,ncsi_service.ncsi_info.sig_vlan_info_changed)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_info.sig_multi_vlan_added.emit,ncsi_service.ncsi_info.sig_multi_vlan_added)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_comm.sig_relink_ncsi_port.emit,ncsi_service.ncsi_comm.sig_relink_ncsi_port)
    lu.assertEquals(ok, true)
    ok = pcall(ncsi_service.ncsi_comm.sig_update_link_status.emit,ncsi_service.ncsi_comm.sig_update_link_status)
    lu.assertEquals(ok, true)
end

function TEST_ncsi_eth:test_set_eth_mac_addr()
    local c_ncsi_info_b = c_ncsi_info
    c_ncsi_info.db = {
        select = function (ncsi_info_obj)
            return {
                first = function ()
                    return {
                        Mac = '20:22:01:59:28:3E',
                        save = function() end
                    }
                end
            }
        end,
        NcsiNCInfo = {}
    }

    c_ncsi_info.ncsi_get_eth_obj = function()
        return {
            Mac = '20:22:01:59:28:3E'
        }
    end

    c_ncsi_info.sig_mac_addr_changed = {
        emit = function() end
    }

    local obj = c_ncsi_info.db:select(c_ncsi_info.db.NcsiNCInfo):first()
    lu.assertEquals(obj.Mac, '20:22:01:59:28:3E')
    c_ncsi_info.set_eth_mac_addr(c_ncsi_info, '20:22:01:59:28:3B')
    lu.assertEquals(obj.Mac, '20:22:01:59:28:3E')

    c_ncsi_info = c_ncsi_info_b
end

function TEST_ncsi_eth:test_nc_info_init_1()
    local c_ncsi_info_b = c_ncsi_info
    c_ncsi_info.db = {
        select = function (self, db_table)
            if db_table == 'NcsiNCInfo' then
                return {
                    first = function ()
                        return {
                            Mac = '20:22:01:59:28:4D',
                            save = function() end
                        }
                    end
                }
            end
            if db_table == 'NcsiNCPortInfo' then
                return {
                    where = function()
                        return {
                            first = function ()
                                return {
                                    PortId = 7,
                                    save = function() end
                                }
                        end}
                    end
                }
            end
            return {
                first = function ()
                    return {
                        save = function() end
                    }
                end
            }
        end,
        NcsiNCInfo = 'NcsiNCInfo',
        NcsiNCPortInfo = 'NcsiNCPortInfo'
    }

    c_ncsi_info.ncsi_get_eth_obj = function()
        return {
            Mac = '20:22:01:59:28:3E',
            VlanState = true,
            VlanId = 4094
        }
    end

    local obj = c_ncsi_info.db:select(c_ncsi_info.db.NcsiNCInfo):first()
    lu.assertEquals(obj.Mac, '20:22:01:59:28:4D')
    c_ncsi_info:ncsi_nc_info_init()
    lu.assertEquals(obj.Mac, '20:22:01:59:28:4D')

    c_ncsi_info = c_ncsi_info_b
end

function TEST_ncsi_eth:test_nc_info_init_2()
    local c_ncsi_info_b = c_ncsi_info
    local NcsiNCInfo_t = {
        save = function() end
    }
    c_ncsi_info.db = {
        select = function (self, db_table)
            if db_table == 'NcsiNCInfo' then
                return {
                    first = function (self)
                        return NcsiNCInfo_t
                    end
                }
            end
            if db_table == 'NcsiNCPortInfo' then
                return {
                    where = function()
                        return {
                            first = function ()
                                return {
                                    PortId = 7,
                                    save = function() end
                                }
                        end}
                    end
                }
            end
            return {
                first = function ()
                    return {
                        save = function() end
                    }
                end
            }
        end,
        NcsiNCInfo = 'NcsiNCInfo',
        NcsiNCPortInfo = 'NcsiNCPortInfo'
    }

    c_ncsi_info.ncsi_get_eth_obj = function()
        return {
            Mac = '20:22:01:59:28:3E',
            VlanState = true,
            VlanId = 4094
        }
    end

    local obj = c_ncsi_info.db:select(c_ncsi_info.db.NcsiNCInfo):first()
    lu.assertEquals(obj.Mac, nil)
    c_ncsi_info:ncsi_nc_info_init()
    lu.assertEquals(obj.Mac, '20:22:01:59:28:3E')

    c_ncsi_info = c_ncsi_info_b
end

function TEST_ncsi_eth:test_nc_info_init_3()
    local c_ncsi_info_b = c_ncsi_info
    local NcsiNCInfo_t = {
        Mac = '00:00:00:00:00:00',
        save = function() end
    }
    c_ncsi_info.db = {
        select = function (self, db_table)
            if db_table == 'NcsiNCInfo' then
                return {
                    first = function (self)
                        return NcsiNCInfo_t
                    end
                }
            end
            if db_table == 'NcsiNCPortInfo' then
                return {
                    where = function()
                        return {
                            first = function ()
                                return {
                                    PortId = 7,
                                    save = function() end
                                }
                        end}
                    end
                }
            end
            return {
                first = function ()
                    return {
                        save = function() end
                    }
                end
            }
        end,
        NcsiNCInfo = 'NcsiNCInfo',
        NcsiNCPortInfo = 'NcsiNCPortInfo'
    }

    c_ncsi_info.ncsi_get_eth_obj = function()
        return {
            Mac = '20:22:01:59:28:3E',
            VlanState = true,
            VlanId = 4094
        }
    end

    local obj = c_ncsi_info.db:select(c_ncsi_info.db.NcsiNCInfo):first()
    lu.assertEquals(obj.Mac, '00:00:00:00:00:00')
    c_ncsi_info:ncsi_nc_info_init()
    lu.assertEquals(obj.Mac, '20:22:01:59:28:3E')

    c_ncsi_info = c_ncsi_info_b
end

-- 测试 ncsi_channel_init 的重试逻辑（第55-63行）
function TEST_ncsi_eth:test_ncsi_channel_init_retry_success()
    local skynet = require 'skynet'
    local log = require 'mc.logging'

    -- 确保 ncsi_service 对象存在并初始化
    if not ncsi_service.ncsi_comm then
        ncsi_service.ncsi_comm = { ncsi_basic_init = function() return false end, ncsi_init = function() end }
    end
    if not ncsi_service.ncsi_info then
        ncsi_service.ncsi_info = { ncsi_info_init = function() end }
    end
    if not ncsi_service.db then
        ncsi_service.db = self.database
    end

    -- 保存原始函数
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_ncsi_basic_init = ncsi_service.ncsi_comm.ncsi_basic_init
    local orig_ncsi_info_init = ncsi_service.ncsi_info.ncsi_info_init
    local orig_ncsi_init = ncsi_service.ncsi_comm.ncsi_init

    -- 打桩：记录调用次数和日志
    local sleep_count = 0
    local error_logs = {}
    local basic_init_called = 0
    local info_init_called = false
    local ncsi_init_called = false

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    -- 第3次调用成功
    ncsi_service.ncsi_comm.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return basic_init_called >= 3
    end

    ncsi_service.ncsi_info.ncsi_info_init = function(self)
        info_init_called = true
    end

    ncsi_service.ncsi_comm.ncsi_init = function(self, db, bus)
        ncsi_init_called = true
    end

    -- 执行测试
    ncsi_service:ncsi_channel_init()

    -- 验证：重试了2次失败，第3次成功
    lu.assertEquals(basic_init_called, 3)
    lu.assertEquals(sleep_count, 2)  -- 失败2次，sleep 2次
    lu.assertEquals(#error_logs, 2)  -- 失败2次，记录2次错误日志
    lu.assertTrue(info_init_called)
    lu.assertTrue(ncsi_init_called)

    -- 还原打桩
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    ncsi_service.ncsi_comm.ncsi_basic_init = orig_ncsi_basic_init
    ncsi_service.ncsi_info.ncsi_info_init = orig_ncsi_info_init
    ncsi_service.ncsi_comm.ncsi_init = orig_ncsi_init
end

-- 测试 ncsi_channel_init 的重试逻辑（达到最大重试次数）
function TEST_ncsi_eth:test_ncsi_channel_init_retry_max_retries()
    local skynet = require 'skynet'
    local log = require 'mc.logging'

    -- 确保 ncsi_service 对象存在并初始化
    if not ncsi_service.ncsi_comm then
        ncsi_service.ncsi_comm = { ncsi_basic_init = function() return false end, ncsi_init = function() end }
    end
    if not ncsi_service.ncsi_info then
        ncsi_service.ncsi_info = { ncsi_info_init = function() end }
    end
    if not ncsi_service.db then
        ncsi_service.db = self.database
    end

    -- 保存原始函数
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_ncsi_basic_init = ncsi_service.ncsi_comm.ncsi_basic_init
    local orig_ncsi_info_init = ncsi_service.ncsi_info.ncsi_info_init
    local orig_ncsi_init = ncsi_service.ncsi_comm.ncsi_init

    -- 打桩：记录调用次数
    local sleep_count = 0
    local error_logs = {}
    local basic_init_called = 0

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    -- 始终返回失败
    ncsi_service.ncsi_comm.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return false
    end

    ncsi_service.ncsi_info.ncsi_info_init = function(self) end
    ncsi_service.ncsi_comm.ncsi_init = function(self, db, bus) end

    -- 执行测试
    ncsi_service:ncsi_channel_init()

    -- 验证：重试了18次（NCSI_BASIC_INIT_MAX_RETRY = 18）
    lu.assertEquals(basic_init_called, 18)
    lu.assertEquals(sleep_count, 18)
    lu.assertEquals(#error_logs, 18)

    -- 还原打桩
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    ncsi_service.ncsi_comm.ncsi_basic_init = orig_ncsi_basic_init
    ncsi_service.ncsi_info.ncsi_info_init = orig_ncsi_info_init
    ncsi_service.ncsi_comm.ncsi_init = orig_ncsi_init
end

-- 测试 init 方法中 skynet.fork 的重试逻辑（第183-194行）
function TEST_ncsi_eth:test_init_fork_retry_logic()
    local skynet = require 'skynet'
    local log = require 'mc.logging'
    local ncsi_cmd = require 'ncsi.ncsi_protocol.ncsi_cmd'

    -- 确保 ncsi_service 对象存在并初始化
    if not ncsi_service.ncsi_comm then
        ncsi_service.ncsi_comm = { ncsi_basic_init = function() return false end, ncsi_init = function() end }
    end
    if not ncsi_service.ncsi_info then
        ncsi_service.ncsi_info = { ncsi_info_init = function() end, wait_bmc_network_service = function() end }
    end
    if not ncsi_service.work then
        ncsi_service.work = { start_ncsi_recv_task = function() end }
    end
    if not ncsi_service.db then
        ncsi_service.db = self.database
    end
    if not ncsi_service.bus then
        ncsi_service.bus = {}
    end

    -- 保存原始函数
    local orig_skynet_fork = skynet.fork
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_ncsi_basic_init = ncsi_service.ncsi_comm.ncsi_basic_init
    local orig_ncsi_info_init = ncsi_service.ncsi_info.ncsi_info_init
    local orig_ncsi_init = ncsi_service.ncsi_comm.ncsi_init
    local orig_ncsi_new_parameter = ncsi_cmd.ncsi_new_parameter
    local orig_set_ncsi_channel_task = ncsi_service.set_ncsi_channel_task
    local orig_set_ncsi_ethid_task = ncsi_service.set_ncsi_ethid_task
    local orig_wait_bmc_network_service = ncsi_service.ncsi_info.wait_bmc_network_service
    local orig_start_ncsi_recv_task = ncsi_service.work.start_ncsi_recv_task
    local orig_register_signal_change_method = ncsi_service.register_signal_change_method

    -- 打桩：捕获 fork 的回调并同步执行
    local fork_callbacks = {}
    local sleep_count = 0
    local error_logs = {}
    local basic_init_called = 0
    local info_init_called = false
    local ncsi_init_called = false
    local ncsi_new_parameter_called = false

    skynet.fork = function(callback)
        table.insert(fork_callbacks, callback)
    end

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    -- 第2次调用成功
    ncsi_service.ncsi_comm.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return basic_init_called >= 2
    end

    ncsi_cmd.ncsi_new_parameter = function()
        ncsi_new_parameter_called = true
    end

    ncsi_service.ncsi_info.ncsi_info_init = function(self)
        info_init_called = true
    end

    ncsi_service.ncsi_comm.ncsi_init = function(self, db, bus)
        ncsi_init_called = true
    end

    -- Mock 其他依赖
    ncsi_service.set_ncsi_channel_task = function() end
    ncsi_service.set_ncsi_ethid_task = function() end
    ncsi_service.ncsi_info.wait_bmc_network_service = function() end
    ncsi_service.work.start_ncsi_recv_task = function() end
    ncsi_service.register_signal_change_method = function() end

    -- 执行测试
    ncsi_service:init()

    -- 找到包含重试逻辑的 fork 回调（第3个fork，索引为3）
    local retry_callback = fork_callbacks[3]
    lu.assertNotNil(retry_callback, "Should have retry callback in fork")

    -- 执行重试回调
    if retry_callback then
        retry_callback()
    end

    -- 验证：重试了1次失败，第2次成功
    lu.assertTrue(ncsi_new_parameter_called)
    lu.assertEquals(basic_init_called, 2)
    lu.assertEquals(sleep_count, 1)  -- 失败1次，sleep 1次
    lu.assertEquals(#error_logs, 1)  -- 失败1次，记录1次错误日志
    lu.assertTrue(info_init_called)
    lu.assertTrue(ncsi_init_called)

    -- 还原打桩
    skynet.fork = orig_skynet_fork
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    ncsi_service.ncsi_comm.ncsi_basic_init = orig_ncsi_basic_init
    ncsi_service.ncsi_info.ncsi_info_init = orig_ncsi_info_init
    ncsi_service.ncsi_comm.ncsi_init = orig_ncsi_init
    ncsi_cmd.ncsi_new_parameter = orig_ncsi_new_parameter
    ncsi_service.set_ncsi_channel_task = orig_set_ncsi_channel_task
    ncsi_service.set_ncsi_ethid_task = orig_set_ncsi_ethid_task
    ncsi_service.ncsi_info.wait_bmc_network_service = orig_wait_bmc_network_service
    ncsi_service.work.start_ncsi_recv_task = orig_start_ncsi_recv_task
    ncsi_service.register_signal_change_method = orig_register_signal_change_method
end

-- 测试 register_signal_change_method 中信号回调的重试逻辑（第458-470行）
function TEST_ncsi_eth:test_sig_eth_intf_add_retry_logic()
    local skynet = require 'skynet'
    local log = require 'mc.logging'

    -- 确保 ncsi_service 对象存在并初始化
    if not ncsi_service.db then
        ncsi_service.db = self.database
    end
    if not ncsi_service.bus then
        ncsi_service.bus = {}
    end

    -- 设置信号对象（在打桩之前）
    local signal = require 'mc.signal'
    local orig_ncsi_info = ncsi_service.ncsi_info
    local orig_ncsi_comm = ncsi_service.ncsi_comm

    ncsi_service.ncsi_info = {
        sig_multi_vlan_added = signal.new(),
        sig_eth_intf_add = signal.new(),
        sig_vlan_info_changed = signal.new(),
        sig_mac_addr_changed = signal.new(),
        ncsi_info_init = function() end,
        ncsi_uptade_link_status = function() end
    }
    ncsi_service.ncsi_comm = {
        sig_relink_ncsi_port = signal.new(),
        sig_update_link_status = signal.new(),
        ncsi_sync_eth_mac = function() end,
        ncsi_init = function() end,
        ncsi_set_vlan_info = function() end,
        ncsi_basic_init = function() end,
        ncsi_update_multi_vlan = function() end
    }

    -- 保存原始函数
    local orig_skynet_fork = skynet.fork
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_log_notice = log.notice

    -- 打桩：捕获 fork 的回调和日志
    local fork_callbacks = {}
    local sleep_count = 0
    local error_logs = {}
    local notice_logs = {}
    local basic_init_called = 0
    local info_init_called = false
    local ncsi_init_called = false

    skynet.fork = function(callback)
        table.insert(fork_callbacks, callback)
    end

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    log.notice = function(self, fmt, ...)
        table.insert(notice_logs, string.format(fmt, ...))
    end

    -- 第4次调用成功
    ncsi_service.ncsi_comm.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return basic_init_called >= 4
    end

    ncsi_service.ncsi_info.ncsi_info_init = function(self)
        info_init_called = true
    end

    ncsi_service.ncsi_comm.ncsi_init = function(self, db, bus)
        ncsi_init_called = true
    end

    ncsi_service.ncsi_comm.ncsi_set_vlan_info = function() end

    -- 执行 register_signal_change_method 注册信号回调
    ncsi_service:register_signal_change_method()

    -- 触发 sig_eth_intf_add 信号
    ncsi_service.ncsi_info.sig_eth_intf_add:emit()

    -- 找到重试逻辑的 fork 回调
    local retry_callback = fork_callbacks[#fork_callbacks]
    lu.assertNotNil(retry_callback, "Should have retry callback in fork")

    -- 执行重试回调
    if retry_callback then
        retry_callback()
    end

    -- 验证：重试了3次失败，第4次成功
    lu.assertEquals(#notice_logs, 1)
    lu.assertNotNil(string.find(notice_logs[1], 'sig eth intf add'))
    lu.assertEquals(basic_init_called, 4)
    lu.assertEquals(sleep_count, 3)  -- 失败3次，sleep 3次
    lu.assertEquals(#error_logs, 3)  -- 失败3次，记录3次错误日志
    lu.assertTrue(info_init_called)
    lu.assertTrue(ncsi_init_called)

    -- 还原打桩
    skynet.fork = orig_skynet_fork
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    log.notice = orig_log_notice
    ncsi_service.ncsi_info = orig_ncsi_info
    ncsi_service.ncsi_comm = orig_ncsi_comm
end

-- 测试 ncsi_basic_init 中 ncsi_enable_available_port_rx 禁用广播/多播过滤（覆盖210-211行）
function TEST_ncsi_eth:test_ncsi_enable_available_port_rx_disable_filters()
    local ncsi_core = require 'ncsi.ncsi_core'

    -- 准备数据库数据：一个端口，PackageId=1, ChannelId=2, EthId=0 => eth0
    local orig_select = self.database.select
    local nc_info = {
        EthId = 0,
        PortNum = 1
    }
    local port_info = {
        PackageId = 1,
        ChannelId = 2,
        RxEnable = false,
        save = function() end
    }
    self.database.select = function(db_self, table)
        if table == db_self.NcsiNCInfo then
            return {
                first = function()
                    return nc_info
                end
            }
        elseif table == db_self.NcsiNCPortInfo then
            return {
                where = function(_, cond)
                    lu.assertEquals(cond.PortId, 1)
                    return {
                        first = function()
                            return port_info
                        end
                    }
                end
            }
        end
        return orig_select(db_self, table)
    end

    -- 创建 ncsi_comm 实例
    local comm = ncsi_comm.new(self.database, {})

    -- 保存原始 ncsi_core 函数
    local orig_enable_channel = ncsi_core.ncsi_enable_channel
    local orig_disable_tx = ncsi_core.ncsi_disable_channelTX
    local orig_disable_brdcast = ncsi_core.ncsi_disable_brdcast_filter
    local orig_disable_multi = ncsi_core.ncsi_disable_multi_filter

    local enable_called = false
    local disable_tx_called = false
    local disable_brdcast_called = false
    local disable_multi_called = false

    ncsi_core.ncsi_enable_channel = function(package_id, channel_id, eth_name)
        enable_called = true
        lu.assertEquals(package_id, 1)
        lu.assertEquals(channel_id, 2)
        lu.assertEquals(eth_name, 'eth0')
        return true
    end

    ncsi_core.ncsi_disable_channelTX = function(package_id, channel_id, eth_name)
        disable_tx_called = true
        lu.assertEquals(package_id, 1)
        lu.assertEquals(channel_id, 2)
        lu.assertEquals(eth_name, 'eth0')
    end

    ncsi_core.ncsi_disable_brdcast_filter = function(package_id, channel_id, eth_name)
        disable_brdcast_called = true
        lu.assertEquals(package_id, 1)
        lu.assertEquals(channel_id, 2)
        lu.assertEquals(eth_name, 'eth0')
    end

    ncsi_core.ncsi_disable_multi_filter = function(package_id, channel_id, eth_name)
        disable_multi_called = true
        lu.assertEquals(package_id, 1)
        lu.assertEquals(channel_id, 2)
        lu.assertEquals(eth_name, 'eth0')
    end

    -- 覆盖 ncsi_wait_eth_interface_up 保证链路为 up，避免提前返回
    local orig_wait_up = ncsi_comm.ncsi_wait_eth_interface_up
    ncsi_comm.ncsi_wait_eth_interface_up = function(self_comm, eth_name)
        lu.assertEquals(eth_name, 'eth0')
        return true
    end

    -- 执行：内部会调用 ncsi_enable_available_port_rx，从而覆盖 210-211 行
    -- 这里只关心内部是否调用到禁用广播/多播过滤的函数，不关心返回值
    comm:ncsi_basic_init(self.database)

    -- 断言：禁用广播/多播过滤函数均被调用
    lu.assertTrue(enable_called)
    lu.assertTrue(disable_tx_called)
    lu.assertTrue(disable_brdcast_called)
    lu.assertTrue(disable_multi_called)

    -- 还原打桩
    self.database.select = orig_select
    ncsi_core.ncsi_enable_channel = orig_enable_channel
    ncsi_core.ncsi_disable_channelTX = orig_disable_tx
    ncsi_core.ncsi_disable_brdcast_filter = orig_disable_brdcast
    ncsi_core.ncsi_disable_multi_filter = orig_disable_multi
    ncsi_comm.ncsi_wait_eth_interface_up = orig_wait_up
end

-- 测试 ncsi_comm:init_ncsi_port 的重试逻辑（第407-423行）- 成功场景
function TEST_ncsi_eth:test_init_ncsi_port_retry_success()
    local skynet = require 'skynet'
    local log = require 'mc.logging'
    local ncsi_core = require 'ncsi.ncsi_core'

    -- 准备数据库数据
    local nc_info = {
        Mac = '20:22:01:59:28:3E',
        ActivePort = 1
    }
    local orig_select = self.database.select
    self.database.select = function(self, table)
        return {
            first = function()
                return nc_info
            end
        }
    end

    -- 创建 ncsi_comm 实例
    local comm_instance = ncsi_comm.new(self.database, {})

    -- 保存原始函数
    local orig_skynet_fork = skynet.fork
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_log_notice = log.notice
    local orig_ncsi_basic_init = comm_instance.ncsi_basic_init
    local orig_ncsi_active_nc_port = comm_instance.ncsi_active_nc_port
    local orig_check_ncsi_link_task = comm_instance.check_ncsi_link_task
    local orig_ncsi_common_init = ncsi_core.ncsi_common_init
    local orig_ncsi_eth_mac_init = ncsi_core.ncsi_eth_mac_init

    -- 打桩：捕获 fork 回调和记录调用
    local fork_callbacks = {}
    local sleep_count = 0
    local error_logs = {}
    local notice_logs = {}
    local basic_init_called = 0
    local active_nc_port_called = false
    local check_ncsi_link_task_called = false
    local ncsi_common_init_called = false
    local ncsi_eth_mac_init_called = false

    skynet.fork = function(callback)
        table.insert(fork_callbacks, callback)
        return 'mock_co'  -- 返回一个模拟的协程对象，用于赋值给 check_ncsi_link_co
    end

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    log.notice = function(self, fmt, ...)
        table.insert(notice_logs, string.format(fmt, ...))
    end

    -- 第2次调用成功
    comm_instance.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return basic_init_called >= 2
    end

    comm_instance.ncsi_active_nc_port = function(self, db, active_port_id)
        active_nc_port_called = true
        lu.assertEquals(active_port_id, 1)
    end

    comm_instance.check_ncsi_link_task = function(self, db, bus)
        check_ncsi_link_task_called = true
    end

    ncsi_core.ncsi_common_init = function()
        ncsi_common_init_called = true
    end

    ncsi_core.ncsi_eth_mac_init = function(mac)
        ncsi_eth_mac_init_called = true
        lu.assertEquals(mac, '20:22:01:59:28:3E')
    end

    -- 执行测试
    comm_instance:init_ncsi_port(self.database, {})

    -- 找到并执行第一个 fork 回调（重试逻辑）
    lu.assertEquals(#fork_callbacks, 1)
    local retry_callback = fork_callbacks[1]
    lu.assertNotNil(retry_callback, "Should have retry callback in fork")

    -- 确保 check_ncsi_link_co 初始值为 nil
    comm_instance.check_ncsi_link_co = nil

    if retry_callback then
        retry_callback()
    end

    -- 验证：重试了1次失败，第2次成功
    lu.assertTrue(ncsi_common_init_called)
    lu.assertTrue(ncsi_eth_mac_init_called)
    lu.assertEquals(#notice_logs, 1)
    lu.assertNotNil(string.find(notice_logs[1], 'Init ncsi port'))
    lu.assertEquals(basic_init_called, 2)
    lu.assertEquals(sleep_count, 1)  -- 失败1次，sleep 1次
    lu.assertEquals(#error_logs, 1)  -- 失败1次，记录1次错误日志
    lu.assertTrue(active_nc_port_called)

    -- 执行第一个回调后，如果 check_ncsi_link_co == nil，会创建第二个 fork
    -- 执行第二个 fork 回调（check_ncsi_link_task）
    lu.assertEquals(#fork_callbacks, 2, "Should have two fork callbacks after retry callback execution")
    local check_link_callback = fork_callbacks[2]
    lu.assertNotNil(check_link_callback, "Should have check_ncsi_link_task callback in fork")

    if check_link_callback then
        check_link_callback()
    end

    lu.assertNotNil(comm_instance.check_ncsi_link_co)
    lu.assertTrue(check_ncsi_link_task_called)

    -- 还原打桩
    skynet.fork = orig_skynet_fork
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    log.notice = orig_log_notice
    comm_instance.ncsi_basic_init = orig_ncsi_basic_init
    comm_instance.ncsi_active_nc_port = orig_ncsi_active_nc_port
    comm_instance.check_ncsi_link_task = orig_check_ncsi_link_task
    ncsi_core.ncsi_common_init = orig_ncsi_common_init
    ncsi_core.ncsi_eth_mac_init = orig_ncsi_eth_mac_init
    self.database.select = orig_select
end

-- 测试 ncsi_comm:init_ncsi_port 的重试逻辑（第407-423行）- 达到最大重试次数
function TEST_ncsi_eth:test_init_ncsi_port_retry_max_retries()
    local skynet = require 'skynet'
    local log = require 'mc.logging'
    local ncsi_core = require 'ncsi.ncsi_core'

    -- 准备数据库数据
    local nc_info = {
        Mac = '20:22:01:59:28:3E',
        ActivePort = 1
    }
    local orig_select = self.database.select
    self.database.select = function(self, table)
        return {
            first = function()
                return nc_info
            end
        }
    end

    -- 创建 ncsi_comm 实例
    local comm_instance = ncsi_comm.new(self.database, {})

    -- 保存原始函数
    local orig_skynet_fork = skynet.fork
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_log_notice = log.notice
    local orig_ncsi_basic_init = comm_instance.ncsi_basic_init
    local orig_ncsi_active_nc_port = comm_instance.ncsi_active_nc_port
    local orig_check_ncsi_link_task = comm_instance.check_ncsi_link_task
    local orig_ncsi_common_init = ncsi_core.ncsi_common_init
    local orig_ncsi_eth_mac_init = ncsi_core.ncsi_eth_mac_init

    -- 打桩：捕获 fork 回调和记录调用
    local fork_callbacks = {}
    local sleep_count = 0
    local error_logs = {}
    local basic_init_called = 0
    local active_nc_port_called = false

    skynet.fork = function(callback)
        table.insert(fork_callbacks, callback)
    end

    skynet.sleep = function(ms)
        sleep_count = sleep_count + 1
    end

    log.error = function(self, fmt, ...)
        table.insert(error_logs, string.format(fmt, ...))
    end

    log.notice = function(self, fmt, ...)
    end

    -- 始终返回失败
    comm_instance.ncsi_basic_init = function(self, db)
        basic_init_called = basic_init_called + 1
        return false
    end

    comm_instance.ncsi_active_nc_port = function(self, db, active_port_id)
        active_nc_port_called = true
    end

    comm_instance.check_ncsi_link_task = function(self, db, bus)
    end

    ncsi_core.ncsi_common_init = function()
    end

    ncsi_core.ncsi_eth_mac_init = function(mac)
    end

    -- 执行测试
    comm_instance:init_ncsi_port(self.database, {})

    -- 找到并执行 fork 回调
    lu.assertEquals(#fork_callbacks, 1)
    local retry_callback = fork_callbacks[1]

    if retry_callback then
        retry_callback()
    end

    -- 验证：重试了18次（NCSI_BASIC_INIT_MAX_RETRY = 18）
    lu.assertEquals(basic_init_called, 18)
    lu.assertEquals(sleep_count, 18)
    lu.assertEquals(#error_logs, 18)
    lu.assertTrue(active_nc_port_called)  -- 即使失败，也会调用 active_nc_port

    -- 还原打桩
    skynet.fork = orig_skynet_fork
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    log.notice = orig_log_notice
    comm_instance.ncsi_basic_init = orig_ncsi_basic_init
    comm_instance.ncsi_active_nc_port = orig_ncsi_active_nc_port
    comm_instance.check_ncsi_link_task = orig_check_ncsi_link_task
    ncsi_core.ncsi_common_init = orig_ncsi_common_init
    ncsi_core.ncsi_eth_mac_init = orig_ncsi_eth_mac_init
    self.database.select = orig_select
end

-- 测试 ncsi_comm:init_ncsi_port 的 check_ncsi_link_co 已存在时不重复创建
function TEST_ncsi_eth:test_init_ncsi_port_check_ncsi_link_co_exists()
    local skynet = require 'skynet'
    local log = require 'mc.logging'
    local ncsi_core = require 'ncsi.ncsi_core'

    -- 准备数据库数据
    local nc_info = {
        Mac = '20:22:01:59:28:3E',
        ActivePort = 1
    }
    local orig_select = self.database.select
    self.database.select = function(db_self, table)
        return {
            first = function()
                return nc_info
            end
        }
    end

    -- 创建 ncsi_comm 实例，并设置 check_ncsi_link_co
    local comm_instance = ncsi_comm.new(self.database, {})
    comm_instance.check_ncsi_link_co = 'existing_co'

    -- 保存原始函数
    local orig_skynet_fork = skynet.fork
    local orig_skynet_sleep = skynet.sleep
    local orig_log_error = log.error
    local orig_log_notice = log.notice
    local orig_ncsi_basic_init = comm_instance.ncsi_basic_init
    local orig_ncsi_active_nc_port = comm_instance.ncsi_active_nc_port
    local orig_check_ncsi_link_task = comm_instance.check_ncsi_link_task
    local orig_ncsi_common_init = ncsi_core.ncsi_common_init
    local orig_ncsi_eth_mac_init = ncsi_core.ncsi_eth_mac_init

    -- 打桩：捕获 fork 回调和记录调用
    local fork_callbacks = {}
    local check_ncsi_link_task_called = false

    skynet.fork = function(callback)
        table.insert(fork_callbacks, callback)
    end

    skynet.sleep = function(ms)
    end

    log.error = function(self, fmt, ...)
    end

    log.notice = function(self, fmt, ...)
    end

    -- 第一次调用就成功
    comm_instance.ncsi_basic_init = function(self, db)
        return true
    end

    comm_instance.ncsi_active_nc_port = function(self, db, active_port_id)
    end

    comm_instance.check_ncsi_link_task = function(self, db, bus)
        check_ncsi_link_task_called = true
    end

    ncsi_core.ncsi_common_init = function()
    end

    ncsi_core.ncsi_eth_mac_init = function(mac)
    end

    -- 执行测试
    comm_instance:init_ncsi_port(self.database, {})

    -- 找到并执行 fork 回调
    lu.assertEquals(#fork_callbacks, 1)
    local retry_callback = fork_callbacks[1]

    if retry_callback then
        retry_callback()
    end

    -- 验证：check_ncsi_link_co 保持原值，不会创建新的
    lu.assertEquals(comm_instance.check_ncsi_link_co, 'existing_co')
    lu.assertFalse(check_ncsi_link_task_called)

    -- 还原打桩
    skynet.fork = orig_skynet_fork
    skynet.sleep = orig_skynet_sleep
    log.error = orig_log_error
    log.notice = orig_log_notice
    comm_instance.ncsi_basic_init = orig_ncsi_basic_init
    comm_instance.ncsi_active_nc_port = orig_ncsi_active_nc_port
    comm_instance.check_ncsi_link_task = orig_check_ncsi_link_task
    ncsi_core.ncsi_common_init = orig_ncsi_common_init
    ncsi_core.ncsi_eth_mac_init = orig_ncsi_eth_mac_init
    self.database.select = orig_select
end

-- 测试 set_ncsi_vlan_id 方法
function TEST_ncsi_eth:test_set_ncsi_vlan_id()
    local save_called = false
    local signal_emitted = false
    local db_obj = {
        VlanId = 0,
        save = function()
            save_called = true
        end
    }

    local c_ncsi_info_mock= {
        set_ncsi_vlan_id = c_ncsi_info.set_ncsi_vlan_id,
        db = {
            select = function (...)
                return {
                    first = function ()
                        return db_obj
                    end
                }
            end,
            NcsiNCInfo = {}
        },
        sig_vlan_info_changed = {
            emit = function()
                signal_emitted = true
            end
        }
    }

    -- 执行测试
    c_ncsi_info_mock:set_ncsi_vlan_id(4094)

    -- 验证结果
    lu.assertEquals(db_obj.VlanId, 4094)
    lu.assertTrue(save_called)
    lu.assertTrue(signal_emitted)
end

-- 测试 set_ncsi_vlan_state 方法
function TEST_ncsi_eth:test_set_ncsi_vlan_state()
    local save_called = false
    local signal_emitted = false
    local db_obj = {
        VlanState = 0,
        save = function()
            save_called = true
        end
    }

    local c_ncsi_info_mock= {
        set_ncsi_vlan_state = c_ncsi_info.set_ncsi_vlan_state,
        db = {
            select = function (...)
                return {
                    first = function ()
                        return db_obj
                    end
                }
            end,
            NcsiNCInfo = {}
        },
        sig_vlan_info_changed = {
            emit = function()
                signal_emitted = true
            end
        }
    }

    -- 执行测试
    c_ncsi_info_mock:set_ncsi_vlan_state(true)

    -- 验证结果
    lu.assertEquals(db_obj.VlanState, 1)
    lu.assertTrue(save_called)
    lu.assertTrue(signal_emitted)
end

-- 测试 OnEthernetInterfacesPropertiesChanged 回调
function TEST_ncsi_eth:test_on_ethernet_interfaces_properties_changed()
    local vlan_info = {
        vlan_state = 0,
        vlan_id = 0
    }
    local c_ncsi_info_mock = {
        listen = c_ncsi_info.listen,
        listen_register_count = 0,
        set_ncsi_vlan_state = function(self, vlan_state)
            vlan_info.vlan_state = vlan_state and 1 or 0
        end,
        set_ncsi_vlan_id = function(self, vlan_id)
            vlan_info.vlan_id = vlan_id
        end
    }
    local client = require 'network_adapter.client'

    local callback_registered = false
    local registered_callback = nil

    -- Mock client.OnEthernetInterfacesPropertiesChanged
    local orig_on_properties_changed = client.OnEthernetInterfacesPropertiesChanged
    client.OnEthernetInterfacesPropertiesChanged = function(self, cb)
        callback_registered = true
        registered_callback = cb
    end

    -- Mock 其他 client 方法
    local orig_subscribe = client.SubscribeEthernetInterfacesNCSIInfoChangedSignal
    local orig_on_inner_added = client.OnInnerNetworkInterfacesAdded
    local orig_on_inner_changed = client.OnInnerNetworkPropertiesChanged

    client.SubscribeEthernetInterfacesNCSIInfoChangedSignal = function(self, cb) end
    client.OnInnerNetworkInterfacesAdded = function(self, cb) end
    client.OnInnerNetworkPropertiesChanged = function(self, cb) end

    -- 执行 listen 方法注册回调
    c_ncsi_info_mock:listen()
    lu.assertTrue(callback_registered)
    lu.assertNotNil(registered_callback)

    -- 模拟属性变化：只有 VLANEnable
    local mock_values = {
        VLANEnable = {
            value = function()
                return true
            end
        },
        VLANId = {
            value = function()
                return 1
            end
        }
    }

    -- 执行回调
    if registered_callback then
        registered_callback(mock_values, nil, nil)
    end

    -- 验证结果
    lu.assertEquals(vlan_info.vlan_state, 1)
    lu.assertEquals(vlan_info.vlan_id, 1)

    -- 还原
    client.OnEthernetInterfacesPropertiesChanged = orig_on_properties_changed
    client.SubscribeEthernetInterfacesNCSIInfoChangedSignal = orig_subscribe
    client.OnInnerNetworkInterfacesAdded = orig_on_inner_added
    client.OnInnerNetworkPropertiesChanged = orig_on_inner_changed
end
