-- 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 c_device_manager = require 'device.device_manager'
local task_mgmt = require 'mc.mdb.task_mgmt'
local utils = require 'mc.utils'
local client = require 'network_adapter.client'
local skynet = require 'skynet'
local ipmi = require 'ipmi'
local cc = ipmi.types.Cc
local file_sec = require 'utils.file'
local vos = require 'utils.vos'
local utils_core = require 'utils.core'
local c_network_adapter = require 'device.class.network_adapter'
local log_collector = require 'device.class.log_collector'
TEST_device_manager = {}

mock_func = {}

local function mock_backup()
    mock_func.chmod_s = utils_core.chmod_s
    mock_func.fork_once = skynet.fork_once
    mock_func.system_s = vos.system_s
    mock_func.remove_file = utils.remove_file
    mock_func.check_realpath_before_open_s = file_sec.check_realpath_before_open_s
    mock_func.check_real_path = file_sec.check_real_path
    mock_func.get_file_name_s = file_sec.get_file_name_s
    mock_func.check_real_path_s = file_sec.check_real_path_s
    mock_func.get_file_accessible = vos.get_file_accessible
    mock_func.GetDPUCardObjects = client.GetDPUCardObjects
    mock_func.GetPackagePackageObject = client.GetPackagePackageObject
    mock_func.get_power_status = c_device_manager.get_power_status
    mock_func.chown_s = utils_core.chown_s 
    mock_func.update_task = task_mgmt.update_task
    mock_func.update_code = task_mgmt.update_code
    mock_func.is_dir = utils_core.is_dir
    mock_func.start_blackbox_coroutine = c_device_manager.start_blackbox_coroutine
    mock_func.collection = c_network_adapter.collection
    mock_func.create_code = task_mgmt.create_code
end

local function mock_restore()
    utils_core.chmod_s = mock_func.chmod_s
    skynet.fork_once = mock_func.fork_once
    vos.system_s = mock_func.system_s
    vos.remove_file = mock_func.remove_file
    file_sec.check_realpath_before_open_s = mock_func.check_realpath_before_open_s
    file_sec.check_real_path = mock_func.check_real_path
    file_sec.get_file_name_s = mock_func.get_file_name_s
    file_sec.check_real_path_s = mock_func.check_real_path_s
    vos.get_file_accessible = mock_func.get_file_accessible
    client.GetDPUCardObjects = mock_func.GetDPUCardObjects
    client.GetPackagePackageObject = mock_func.GetPackagePackageObject
    c_device_manager.get_power_status = mock_func.get_power_status
    utils_core.chown_s = mock_func.chown_s 
    task_mgmt.update_task = mock_func.update_task
    task_mgmt.update_code = mock_func.update_code
    utils_core.is_dir = mock_func.is_dir
    c_device_manager.start_blackbox_coroutine = mock_func.start_blackbox_coroutine
    c_network_adapter.collection = mock_func.collection
    task_mgmt.create_code = mock_func.create_code
end

local function mock_prepare_1()
    client.GetDPUCardObjects = function()
        return {}
    end
    task_mgmt.create_code = { TASK_CREATE_SUCCESSFUL = 1 }
    task_mgmt.create_task = function()
        return 2, 0, 1
    end
end
local function mock_prepare_2()
    skynet.fork_once = function(cb, id , card, path, ctx)
        cb(id, card, path, ctx)
    end

    file_sec.check_realpath_before_open_s = function()
        return 1
    end
    vos.system_s = function(cmd, flag1, path, flag2, src_dir, dest_dir)
        return true
    end
    utils.remove_file = function(dir)
        return dir
    end
    utils.realpath = function(obj)
        return false
    end
    utils_core.chmod_s = function()
        
    end
    utils_core.chown_s = function()
        
    end
    task_mgmt.update_task = function()
        return 1
    end
    task_mgmt.update_code = {TASK_UPDATE_SUCCESSFUL = 1}
end
function TEST_device_manager:test_start_blackbox_coroutine()
    mock_backup()
    mock_prepare_1()
    local obj = { path = "test" }
    
    c_network_adapter.collection = {}
    local cards = {{SlotNumber = 1, Name = "test", pldm_config_obj = {BlackBox = function()
        return {value = function()  return {data = 1, length = 1} end} 
    end}}}

    local ctx = {
        get_initiator = function()
            return "test"
        end
    }
    c_network_adapter.collection.fetch = function()
        return {}
    end
    -- 测试 create_blackbox_task
    local ok, _ = pcall(c_device_manager.start_blackbox_coroutine, c_device_manager, obj, nil, ctx)
    lu.assertEquals(ok, false)

    mock_prepare_2()
    task_mgmt.create_code.TASK_CREATE_SUCCESSFUL = 2
    --测试 dump_func, compress_file_to_path
    ok, _ = pcall(c_device_manager.start_blackbox_coroutine, c_device_manager, obj, nil, ctx)
    lu.assertEquals(ok, true)
    c_network_adapter.collection.fetch = function()
        return cards
    end
    local open_s = file_sec.open_s 
    file_sec.open_s = function()
        return false
    end
    -- local c_device_manager_tmp = c_device_manager
    -- --测试 dump_one_card_blackbox_by_pldm， dump_one_card_blackbox_by_rmii， write_blackbox_to_file
    pcall(c_device_manager.start_blackbox_coroutine, c_device_manager, obj, nil, ctx)
    cards[1].pldm_config_obj.BlackBox = function()
        return {value = function()  return nil end} 
    end
    pcall(c_device_manager.start_blackbox_coroutine, c_device_manager, obj, nil, ctx)
 
    file_sec.open_s = open_s
    mock_restore()
end

function TEST_device_manager:test_method_dump_black_box()
    mock_backup()
    c_device_manager.get_power_status = function()
        return "OFF"
    end
    local ctx = {get_initiator = function()
        return 'test'
    end}
    
    -- 测试 method_dump_black_box 749行
    local ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, path)
    lu.assertEquals(ok, false)
    c_device_manager.get_power_status = function()
        return "ON"
    end
    -- 测试 method_dump_black_box 757行
    local path = string.rep('1', 256)
    ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, path)
    lu.assertEquals(ok, false)
    mock_restore()
end

function TEST_device_manager:test_method_dump_black_box_2()
    mock_restore()
    local true_func = function()
        return true
    end
    c_device_manager.start_blackbox_coroutine = function ()
        return false, 1
    end

    file_sec.check_realpath_before_open_s = true_func
    file_sec.get_file_name_s = function() return "test" end
    file_sec.check_real_path_s = true_func
    vos.get_file_accessible = function()
        return false
    end
    local path = string.rep('1', 10)
    c_device_manager.get_power_status = function()
        return "ON"
    end
    -- 测试 method_dump_black_box 757
    utils_core.is_dir = true_func
    local ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, path)
    lu.assertEquals(ok, false)

    -- 测试 method_dump_black_box 763
    utils_core.is_dir = true_func
    local tmp_func = file_sec.check_realpath_before_open_s
    file_sec.check_realpath_before_open_s = function()
        return 0
    end
    file_path = path .. '.txt'
    local ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, file_path)
    lu.assertEquals(ok, false)
    file_sec.check_realpath_before_open_s = tmp_func

    -- 测试 method_dump_black_box 770行
    utils_core.is_dir = function ()
        return false 
    end
    c_device_manager.start_blackbox_coroutine = function ()
        return false, 1
    end
    ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, path)
    lu.assertEquals(ok, false)

    utils_core.is_dir = function ()
        return false 
    end
    -- 测试 method_dump_black_box   行
    ok, _ = pcall(c_device_manager.method_dump_black_box, c_device_manager,  {"test"}, ctx, path)
    lu.assertEquals(ok, false)
    mock_restore()
end

local function mock_dpu_ipmi_req_and_ctx()
    local req = {
        ManufactureId = 0x0007db,
    }
    local ctx = {
        chan_num = 13
    }
    return req, ctx
end

local function mock_dpu_ipmi_invalid_req_and_ctx()
    local req = {
        ManufactureId = 0,
    }
    local ctx = {
        chan_num = 13
    }
    return req, ctx
end

local function mock_dpu_ipmi_req_and_invalid_ctx()
    local req = {
        ManufactureId = 0x0007db,
    }
    local ctx = {
        chan_num = 255
    }
    return req, ctx
end

function TEST_device_manager:test_get_customer()
    mock_backup()
    local package_objs = {Customer = 'CTCC'}
    client.GetPackagePackageObject = function()
        return package_objs
    end
    local ok, _ = pcall(c_device_manager.get_customer, c_device_manager)
    lu.assertNotEquals(ok, 'CTCC')

    mock_restore()
end

function TEST_device_manager:test_get_customer_failed()
    mock_backup()
    client.GetPackagePackageObject = function()
        return 
    end
    local ok, _ = pcall(c_device_manager.get_customer, c_device_manager)
    lu.assertNotEquals(ok, '')

    local package_objs = {}
    client.GetPackagePackageObject = function()
        return package_objs
    end
    local ok, _ = pcall(c_device_manager.get_customer, c_device_manager)
    lu.assertNotEquals(ok, '')

    mock_restore()
end

function TEST_device_manager:test_get_customer_failed_package_objs_is_null()
    mock_backup()
    local package_objs = {}
    client.GetPackagePackageObject = function()
        return package_objs
    end
    local ok, _ = pcall(c_device_manager.get_customer, c_device_manager)
    lu.assertNotEquals(ok, '')

    mock_restore()
end

function TEST_device_manager:test_get_dpu_os_startup_status()
    mock_backup()
    c_network_adapter.collection = {}
    -- ManufactureId和DevNo，chan_num均合法
    local req, ctx = mock_dpu_ipmi_req_and_ctx()

    local bf2_cards = {nic_os_startup_status = 5}
    c_network_adapter.collection.find = function(model)
        return bf2_cards
    end
    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    c_network_adapter.nic_os_startup_status = 5
    pcall(c_device_manager.get_dpu_os_startup_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_os_startup_status_invalid_chan_num()
    mock_backup()
    -- ManufactureId合法，chan_num不合法
    local req, ctx = mock_dpu_ipmi_req_and_invalid_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    pcall(c_device_manager.get_dpu_os_startup_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_os_startup_status_invalid_manufacture()
    mock_backup()
    -- ManufactureId不合法，chan_num合法
    local req, ctx = mock_dpu_ipmi_invalid_req_and_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    pcall(c_device_manager.get_dpu_os_startup_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_os_startup_status_invalid_network_adapter()
    mock_backup()
    -- ManufactureId和DevNo，chan_num均合法
    local req, ctx = mock_dpu_ipmi_req_and_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    c_device_manager.get_network_adapter_by_model = function()
        return
    end
    pcall(c_device_manager.get_dpu_os_startup_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_force_power_on_status()
    mock_backup()
    c_network_adapter.collection = {}
    local bf2_cards = {bios_boot_wait_flag = 1}
    c_network_adapter.collection.find = function(model)
        return bf2_cards
    end

    -- ManufactureId和DevNo，chan_num均合法
    local req, ctx = mock_dpu_ipmi_req_and_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    pcall(c_device_manager.get_dpu_force_power_on_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_force_power_on_status_invalid_chan_num()
    mock_backup()
    -- ManufactureId合法，chan_num不合法
    local req, ctx = mock_dpu_ipmi_req_and_invalid_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    pcall(c_device_manager.get_dpu_force_power_on_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_force_power_on_status_invalid_manufactureid()
    mock_backup()
    -- ManufactureId不合法，chan_num合法
    local req, ctx = mock_dpu_ipmi_invalid_req_and_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    pcall(c_device_manager.get_dpu_force_power_on_status, c_device_manager, req, ctx)

    mock_restore()
end

function TEST_device_manager:test_get_dpu_force_power_on_status_invalid_network_adapter()
    mock_backup()
    -- ManufactureId和chan_num均合法
    local req, ctx = mock_dpu_ipmi_req_and_ctx()

    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    c_device_manager.get_network_adapter_by_model = function()
        return
    end

    pcall(c_device_manager.get_dpu_force_power_on_status, c_device_manager, req, ctx)

    mock_restore()
end


function TEST_device_manager:test_set_dpu_force_power_on_status()
    mock_backup()
    c_network_adapter.collection = {}
    local bf2_cards = {bios_boot_wait_flag = 0}
    c_network_adapter.collection.find = function(model)
        return bf2_cards
    end
    
    c_device_manager.get_customer = function()
        return 'CTCC'
    end

    pcall(c_device_manager.set_dpu_force_power_on_status, c_device_manager)

    mock_restore()
end

function TEST_device_manager:test_set_dpu_force_power_on_status_invalid_customer()
    mock_backup()
    
    c_device_manager.get_customer = function()
        return 'CMCC'
    end

    pcall(c_device_manager.set_dpu_force_power_on_status, c_device_manager)

    mock_restore()
end

function TEST_device_manager:test_set_dpu_force_power_on_status_invalid_network_adapter()
    mock_backup()
    
    c_device_manager.get_customer = function()
        return 'CTCC'
    end
    c_device_manager.get_network_adapter_by_model = function()
        return
    end

    pcall(c_device_manager.set_dpu_force_power_on_status, c_device_manager)

    mock_restore()
end

function TEST_device_manager:test_log_device_npu_optical_module()
    local col_tab = c_network_adapter.collection 
    c_network_adapter.collection = {
        fetch = function(...)
            return {
                test = {
                    collect_optical_static_info = function(...)
                        return "test"
                    end
                }
            }
        end
    }
    local dump_func = log_collector.dump_string_to_log
    local dump_log = 1
    log_collector.dump_string_to_log = function (...)
        dump_log = 2
        return nil
    end

    c_device_manager.log_device_npu_optical_module('test')
    lu.assertEquals(dump_log, 2)
    c_network_adapter.collection = col_tab
    log_collector.dump_string_to_log = dump_func
end

function TEST_device_manager:test_get_dpu_host_access()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {}
    local ctx = {}
    local rsp = c_device_manager:get_dpu_host_access(req, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    lu.assertEquals(rsp.Status, 0)
    
    -- 测试获取network_adapter成功的情况
    local mock_compcode = cc.Success
    local mock_host_access = 1
    local mock_adapter = {
        get_dpu_host_access = function()
            return mock_compcode, mock_host_access
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:get_dpu_host_access(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    lu.assertEquals(rsp.Status, mock_host_access)
    
    mock_restore()
end

function TEST_device_manager:test_set_dpu_host_access()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {Status = 1}
    local ctx = {ChanType = 1}

    local rsp = c_device_manager:set_dpu_host_access(req, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    
    -- 测试获取network_adapter成功且操作成功的情况
    local mock_compcode = cc.Success
    local mock_adapter = {
        set_dpu_host_access = function(status, ctx)
            return mock_compcode
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:set_dpu_host_access(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    
    -- 测试获取network_adapter成功但操作失败的情况
    mock_compcode = cc.UnspecifiedError
    mock_adapter.set_dpu_host_access = function(status, ctx)
        return mock_compcode
    end
    rsp = c_device_manager:set_dpu_host_access(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
 
    mock_restore()
end

function TEST_device_manager:test_get_dpu_extended_host_privilege()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {}
    local ctx = {}
    local rsp = c_device_manager:get_dpu_extended_host_privilege(req, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    lu.assertEquals(rsp.Status, 0)
    
    -- 测试获取network_adapter成功的情况
    local mock_compcode = cc.Success
    local mock_privilege = 2
    local mock_adapter = {
        get_dpu_extended_host_privilege = function()
            return mock_compcode, mock_privilege
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:get_dpu_extended_host_privilege(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    lu.assertEquals(rsp.Status, mock_privilege)
    
    mock_restore()
end

function TEST_device_manager:test_set_dpu_extended_host_privilege()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {Status = 1}
    local ctx = {ChanType = 1}

    local rsp = c_device_manager:set_dpu_extended_host_privilege(req.Status, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    
    -- 测试获取network_adapter成功且操作成功的情况
    local mock_compcode = cc.Success
    local mock_adapter = {
        set_dpu_extended_host_privilege = function(status, ctx)
            return mock_compcode
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:set_dpu_extended_host_privilege(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    
    -- 测试获取network_adapter成功但操作失败的情况
    mock_compcode = cc.UnspecifiedError
    mock_adapter.set_dpu_extended_host_privilege = function()
        return mock_compcode
    end
    rsp = c_device_manager:set_dpu_extended_host_privilege(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    
    mock_restore()
end

function TEST_device_manager:test_get_dpu_mode()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {}
    local ctx = {}
    local rsp = c_device_manager:get_dpu_mode(req, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    lu.assertEquals(rsp.Mode, 0)
    
    -- 测试获取network_adapter成功的情况
    local mock_compcode = cc.Success
    local mock_mode = 3
    local mock_adapter = {
        get_dpu_mode = function()
            return mock_compcode, mock_mode
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:get_dpu_mode(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    lu.assertEquals(rsp.Mode, mock_mode)
    
    mock_restore()
end

function TEST_device_manager:test_set_dpu_mode()
    mock_backup()
    
    -- 测试获取network_adapter失败的情况
    c_device_manager.get_bf3_by_model = function()
        return nil
    end
    local req = {Mode = 1}
    local ctx = {ChanType = 1}
    
    local rsp = c_device_manager:set_dpu_mode(req, ctx)
    lu.assertEquals(rsp.CompletionCode, cc.CommandDisabled)
    
    -- 测试获取network_adapter成功且操作成功的情况
    local mock_compcode = cc.Success
    local mock_adapter = {
        set_dpu_mode = function(mode)
            return mock_compcode
        end
    }
    c_device_manager.get_bf3_by_model = function()
        return mock_adapter
    end
    rsp = c_device_manager:set_dpu_mode(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)
    
    -- 测试获取network_adapter成功但操作失败的情况
    mock_compcode = cc.UnspecifiedError
    mock_adapter.set_dpu_mode = function()
        return mock_compcode
    end
    rsp = c_device_manager:set_dpu_mode(req, ctx)
    lu.assertEquals(rsp.CompletionCode, mock_compcode)

    mock_restore()
end
