-- 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 sml = require('sml')
local drives_object = require('drives.drives_object')
local c_pd_info = require('sml.pd_info')
local c_pd_sas_smart_info = require('sml.pd_sas_smart_info')
local pd_object = require('pd.pd_object')
local STOP_LOCATE <const> = 0
local LOCATE <const> = 1
local NUMBER_DRIVE_NAME = 1 -- 非法数字格式的drive_name
local INVALID_LOCATION_INDICATOR_STATE = 2 -- 非法的location_indicator_state
local file_sec = require('utils.file')
local utils = require('mc.utils')
local project_dir = os.getenv('PROJECT_DIR')
local add_event = require('add_event')
local client = require('storage.client')
local storageApp = require('storage_app')
local c_nvme = require('nvme.nvme_object')
local method_misc = require('method_misc')
local mdb = require('mc.mdb')
local rpc_service_subhealth = require('rpc_services.rpc_service_subhealth')
local rpc_service_drive = require('rpc_services.rpc_service_drive')
local common_def = require('common_def')
local c_storageconfig = require('storageconfig.storageconfig_object')
local sml_cmd = require('hwproxy.plugins.sml.cmds')
local c_ipmi_service = require('ipmi_service')
local c_customize_config = require('others.customize_config')
local c_export_import_engine = require('handler.export_import_engine')
local skynet = require('skynet')

local function make_pd_info(test_data)
    return c_pd_info.new({
        model = test_data.pd_model,
        temperature = test_data.pd_temperature,
        coerced_size = test_data.pd_capacity,
        interface_type = test_data.pd_interface_type,
        device_speed = test_data.pd_device_speed,
        link_speed = test_data.pd_link_speed,
        media_type = test_data.pd_media_type,
        fw_state = test_data.pd_fw_status,
        serial_num = test_data.pd_sn,
        manufacturer = test_data.pd_manufacturer,
        power_on_hours = test_data.pd_ssd_power_on_hours,
        firmware_version = test_data.pd_fw_version,
        rotation_speed = test_data.pd_rotation_speed,
        block_size = test_data.pd_block_size,
        sas_addr1 = test_data.pd_sas_addr_1,
        sas_addr2 = test_data.pd_sas_addr_2,
        remnant_media_wearout = test_data.pd_ssd_lifeleft,
        prefail_count = test_data.pd_predictive_fail_count,
        media_err_count = test_data.pd_media_err_count,
        other_err_count = test_data.pd_other_err_count,
        proginfo = {
            patrol_state = test_data.pd_patrol_status,
            rebuild_state = test_data.pd_rebuild_state,
            rebuild_progress = test_data.pd_rebuild_progress,
        },
        power_state = test_data.pd_power_state,
        boot_priority = test_data.pd_boot_priority,
        hot_spare = test_data.pd_hot_spare,
        last_prefail_event_seq_num = test_data.pd_last_prefail_event_seq_num,
    })
end

local function construct_ctx()
    local ctx = {}
    ctx.ChanType = 1
    ctx.get_initiator = function()
        return {}
    end

    return ctx
end

local ctx = construct_ctx()

local function make_pd_sas_smart_info(test_data)
    return c_pd_sas_smart_info.new({
        strip_temperature = test_data.strip_temperature,
        glist_len = test_data.glist_len,
        plist_len = test_data.plist_len,
        manufacture_data = test_data.manufacture_data,
        blocks_sent = test_data.blocks_sent,
        blocks_received = test_data.blocks_received,
        minutes_left = test_data.minutes_left,
    })
end

local function update_test_data(test_data)
    test_data.pd_model = 'new_model'
    test_data.pd_temperature = test_data.pd_temperature - 1
    test_data.pd_capacity = test_data.pd_capacity + 1
    test_data.pd_interface_type = 'new_SAS'
    test_data.pd_device_speed = 3
    test_data.pd_link_speed = 3
    test_data.pd_media_type = 2
    test_data.pd_fw_status = 2
    test_data.pd_sn = 'NEWZC1ALLK4'
    test_data.pd_manufacturer = 'NEWATA'
    test_data.pd_ssd_power_on_hours = 255
    test_data.pd_fw_version = 'NEWTN05'
    test_data.pd_rotation_speed = 7000
    test_data.pd_block_size = 612
    test_data.pd_sas_addr_1 = 'NEW5942533A44D5E000'
    test_data.pd_sas_addr_2 = 'NEWN/A'
    test_data.pd_ssd_lifeleft = 15
    test_data.pd_predictive_fail_count = 52
    test_data.pd_media_err_count = 15
    test_data.pd_other_err_count = 66
    test_data.pd_patrol_status = 2
    test_data.pd_power_state = 1
    test_data.pd_rebuild_state = 1
    test_data.pd_rebuild_progress = 30
    test_data.pd_boot_priority = 2
    test_data.pd_hot_spare = 0
    test_data.pd_last_prefail_event_seq_num = 123
end

local function check_drive(drive, test_data)
    lu.assertEquals(drive.Model, test_data.pd_model)
    lu.assertEquals(drive.TemperatureCelsius, test_data.pd_temperature)
    lu.assertEquals(drive.CapacityMiB, test_data.pd_capacity)
    lu.assertEquals(drive.Protocol, test_data.pd_interface_type)
    lu.assertEquals(drive.CapableSpeedGbs, test_data.pd_device_speed)
    lu.assertEquals(drive.NegotiatedSpeedGbs, test_data.pd_link_speed)
    lu.assertEquals(drive.MediaType, test_data.pd_media_type)
    lu.assertEquals(drive.FirmwareStatus, test_data.pd_fw_status)
    lu.assertEquals(drive.SerialNumber, test_data.pd_sn)
    lu.assertEquals(drive.Manufacturer, test_data.pd_manufacturer)
    lu.assertEquals(drive.PowerOnHours, test_data.pd_ssd_power_on_hours)
    lu.assertEquals(drive.Revision, test_data.pd_fw_version)
    lu.assertEquals(drive.RotationSpeedRPM, test_data.pd_rotation_speed)
    lu.assertEquals(drive.BlockSizeBytes, test_data.pd_block_size)
    lu.assertEquals(drive.SASAddress1, test_data.pd_sas_addr_1)
    lu.assertEquals(drive.SASAddress2, test_data.pd_sas_addr_2)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, test_data.pd_ssd_lifeleft)
    lu.assertEquals(drive.PredictedFailCount, test_data.pd_predictive_fail_count)
    lu.assertEquals(drive.MediaErrorCount, test_data.pd_media_err_count)
    lu.assertEquals(drive.OtherErrorCount, test_data.pd_other_err_count)
    lu.assertEquals(drive.PatrolState, test_data.pd_patrol_status)
    lu.assertEquals(drive.PowerState, test_data.pd_power_state)
    lu.assertEquals(drive.RebuildState, test_data.pd_rebuild_state)
    lu.assertEquals(drive.RebuildProgress, test_data.pd_rebuild_progress)
    lu.assertEquals(drive.BootPriority, test_data.pd_boot_priority)
    lu.assertEquals(drive.HotspareType, test_data.pd_hot_spare)
    lu.assertEquals(drive.LastPrefailEventSeqNum, test_data.pd_last_prefail_event_seq_num)
end

local function create_drive(self)
    local drive_info = { HddBackplaneStartSlot = 0, Presence = 1, RelativeSlot = 1, ObjectName = 'Drive1', Id = 1 }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    local drive = self.drive_collection:get_drive(drive_info.Name)
    local pd_info = { controller_id = 1, controller_type = 0 }
    local pd = { slot_num = 2, enclosure_id = 3, device_id = 1 }
    local obj = pd_object.new(pd_info.controller_id, pd, pd_info.controller_type)
    drive:identified(obj)

    return drive_info, drive
end

function TestStorage:test_add_drive_object()
    local _, drive = create_drive(self)
    lu.assertNotIsNil(drive)

    -- 模拟定位 pd
    local pd = { controller_id = 1, slot_num = 2, enclosure_id = 3, device_id = 1 }
    lu.assertIsTrue(drive:is_identified())
    lu.assertEquals(drive.RefControllerId, pd.controller_id)
    lu.assertEquals(drive.SlotNumber, pd.slot_num)
    lu.assertEquals(drive.EnclosureId, pd.enclosure_id)
    lu.assertEquals(drive.device_id, pd.device_id)

    -- hook drive 用到的 sml 函数
    local params = {}

    ---@diagnostic disable-next-line: duplicate-set-field
    sml.get_pd_info = function(...)
        params['get_pd_info'] = { ... } -- 启动任务更新 drive 信息
        return make_pd_info(self.test_pd_info)
    end
    ---@diagnostic disable-next-line: duplicate-set-field
    sml.get_pd_sas_smart_info = function(...)
        params['get_pd_sas_smart_info'] = { ... } -- 启动任务更新 drive 信息
        return make_pd_sas_smart_info(self.test_pd_sas_smart_info)
    end

    drives_object.get_instance().assert_max_temp = function()
        return false, false
    end

    self.tasks:run_all_task()
    lu.assertEquals(drive:get_task_count(), 5)

    -- 校验任务执行结果
    lu.assertEquals(params, { get_pd_info = { { Priority = 'Secondary' }, pd.controller_id, drive.device_id } })
    lu.assertEquals(drive.NodeId, 'HDDPlaneDisk1')
    check_drive(drive, self.test_pd_info)

    -- 模拟解除 pd 定位，所有任务都会退出
    drive:identified(nil)
    lu.assertEquals(drive:get_task_count(), 2)

    -- 重新定位，模拟换一个 controller_id 但是 slot_num 和 enclosure_id 不变，
    -- 定位后会重启所有任务，这里故意更换一下静态信息，确保静态信息也能得到更新
    local new_pd = { controller_id = 2, slot_num = 2, enclosure_id = 3, device_id = 2 }
    update_test_data(self.test_pd_info)

    drive:identified(new_pd)
    self.tasks:run_all_task()

    lu.assertEquals(params, { get_pd_info = { { Priority = 'Secondary' }, new_pd.controller_id, drive.device_id } })
    check_drive(drive, self.test_pd_info)
end

-- pd_operation 打桩
local function sml_pd_operation(self)
    return function(controller_index, device_id, operation, data)
        local sml_drive = self.drive_collection:get_drive_by_id(controller_index, device_id)
        if operation == STOP_LOCATE then
            sml_drive.LocateLed = STOP_LOCATE
        elseif operation == LOCATE then
            sml_drive.LocateLed = LOCATE
        elseif operation == 5 then -- 5: 设置固件状态
            if data == string.pack('B', 0) .. '\x01' then -- 0: 设置 UG 状态
                sml_drive.FirmwareStatus = 0
            elseif data == string.pack('B', 6) .. '\x01' then -- 6: 设置 online 状态
                sml_drive.FirmwareStatus = 6 -- 6: 设置 online 状态
            end
        elseif operation == 8 then -- 8: 设置启动盘
            if data == string.pack('B', 1) then -- 1: 设置启动盘
                sml_drive.BootPriority = 1
            elseif data == string.pack('B', 0) then -- 0 取消启动盘
                sml_drive.BootPriority = 0
            end
        elseif operation == 9 then -- 9: 设置巡检状态
            if data == string.pack('B', 1) then -- 1: 开启巡检状态
                sml_drive.PatrolState = 1
            elseif data == string.pack('B', 0) then
                sml_drive.PatrolState = 0
            end
        elseif operation == 2 then -- 2: 设置全局热备盘
            sml_drive.HotspareType = 1
        elseif operation == 3 then -- 3: 设置局部热备盘
            sml_drive.HotspareType = 2
        elseif operation == 4 then -- 4 取消热备盘
            sml_drive.HotspareType = 0
        end
        return 0
    end
end

function TestStorage:test_set_drive_location_indicator_state()
    local drive_info, drive = create_drive(self)

    sml.pd_operation = sml_pd_operation(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    -- 设置定位灯
    drive.LocateLed = STOP_LOCATE
    local ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        drive_info.Name,
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.LocateLed, LOCATE)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        drive_info.Name,
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.LocateLed, STOP_LOCATE)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        NUMBER_DRIVE_NAME,
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        NUMBER_DRIVE_NAME,
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        'Invalid_Param',
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        'Invalid_Param',
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        drive_info.Name,
        ctx,
        INVALID_LOCATION_INDICATOR_STATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        drive_info.Name,
        ctx,
        INVALID_LOCATION_INDICATOR_STATE
    )
    lu.assertEquals(ret, false)
    drive.RefControllerId = 0xff
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetLocationIndicatorState',
        drive_info.Name,
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, true)
end

function TestStorage:test_set_drive_fault_indicator_state()
    local drive_info = create_drive(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    -- 设置故障灯
    local ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        drive_info.Name,
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, true)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        drive_info.Name,
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, true)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        NUMBER_DRIVE_NAME,
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        NUMBER_DRIVE_NAME,
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        'Invalid_Param',
        ctx,
        STOP_LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        'Invalid_Param',
        ctx,
        LOCATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        drive_info.Name,
        ctx,
        INVALID_LOCATION_INDICATOR_STATE
    )
    lu.assertEquals(ret, false)
    ret = pcall(
        rpc_service_drive.drive_operate,
        rpc_service_drive,
        'SetFaultIndicatorState',
        drive_info.Name,
        ctx,
        INVALID_LOCATION_INDICATOR_STATE
    )
    lu.assertEquals(ret, false)
end

function TestStorage:test_set_drive_hotspare_type()
    local drive_info, drive = create_drive(self)

    sml.pd_operation = sml_pd_operation(self)

    local rpc_service_drive = self.rpc_service_drive
    self.tasks:run_all_task()

    drive.HotspareType = 0
    lu.assertEquals(drive.HotspareType, 0)
    sml.get_pd_info = function()
        return { hot_spare = 1 }
    end
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 1, 0)
    lu.assertEquals(ret, false)

    method_misc.test_controller_vendor = function()
        return true
    end

    drive.RefControllerTypeId = 1
    sml.get_pd_info = function()
        return { hot_spare = 1 }
    end
    local ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 1, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 1)
    sml.get_pd_info = function()
        return { hot_spare = 0 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 0, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 0)
    sml.get_pd_info = function()
        return { hot_spare = 2 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 2, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 2)
    sml.get_pd_info = function()
        return { hot_spare = 3 }
    end
    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 3, 0)
    lu.assertEquals(ret, true)
    lu.assertEquals(drive.HotspareType, 3)

    ret = pcall(rpc_service_drive.drive_operate, rpc_service_drive, 'SetHotspareType', drive_info.Name, ctx, 4, 0)
    lu.assertEquals(ret, false)
end

-- 测试定位状态更新
function TestStorage:test_update_location_indicator_state()
    local drive_info = {
        HddBackplaneStartSlot = 0,
        Presence = 0,
        RelativeSlot = 1,
        ObjectName = 'Drive1',
        LocateLed = 1,
        FaultLed = 0,
    }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.tasks:run_task(self.mctp_service.mctp_start_task)
    local drive = self.drive_collection:get_drive(drive_info.Name)
    lu.assertEquals(drive.LocationIndicatorState, 1)

    drive.LocateLed = 0
    drive.on_property_changed:emit('LocateLed', 0)
    lu.assertEquals(drive.LocationIndicatorState, 0)

    drive.LocateLed = 1
    drive.on_property_changed:emit('LocateLed', 1)
    lu.assertEquals(drive.LocationIndicatorState, 1)
end

local function loop_exec_func(func, times, ...)
    for i = 1, times do
        func(...)
    end
end

function TestStorage:test_firmware_status_err_associated_cpld()
    local controller_1 = {
        Id = 0,
        path = 'Controller_1_0001010102',
        CtrlOption1 = 2275649443,
        CtrlOption2 = 32647,
        CtrlOption3 = 0,
        TypeId = 99,
        OOBSupport = 1,
    }
    local controller_2 = {
        Id = 1,
        path = 'Controller_2_0001010102',
        CtrlOption1 = 2275649443,
        CtrlOption2 = 32647,
        CtrlOption3 = 0,
        TypeId = 99,
        OOBSupport = 1,
    }

    local storageconfig = self.c_storageconfig
    storageconfig.task = nil

    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.object_manager.mc:add_object('Controller', controller_1, 1)

    loop_exec_func(self.tasks.run_task, 15, self.tasks, storageconfig.task)

    -- 检测15次未满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, false)

    -- 加一个控制器重新开始计数
    self.object_manager.mc:add_object('Controller', controller_2, 2)

    loop_exec_func(self.tasks.run_task, 15, self.tasks, storageconfig.task)

    -- 检测15次未满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, false)

    self.tasks:run_task(storageconfig.task)

    -- 第16次满足条件
    lu.assertEquals(storageconfig.all_ctrl_loaded, true)

    local drive_info = {
        HddBackplaneStartSlot = 0,
        Presence = 1,
        RelativeSlot = 1,
        ObjectName = 'Drive1',
        LocateLed = 0,
        FaultLed = 1,
    }
    self.object_manager.mc:add_object('Drive', drive_info)

    -- 由于FaultLed加入了防抖，需要获取5次
    loop_exec_func(self.tasks.run_all_task, 5, self.tasks)
    local drive = self.drive_collection:get_drive('Disk1')
    lu.assertEquals(drive.FirmwareStatusError, true)
    self.object_manager.mc:del_object('Controller', controller_1, 1)
    self.object_manager.mc:del_object('Controller', controller_2, 2)
    self.tasks:run_all_task()
end

function TestStorage:test_update_drive_slot_info()
    local drive_info = { Id = 1, Name = 'Disk1', Presence = 1, ObjectName = 'Drive1' }
    self.object_manager.mc:add_object('Drive', drive_info)
    self.bus_monitor_service.on_os_state_changed:emit(true)
    self.tasks:run_task(self.mctp_service.mctp_start_task)

    local drive = self.drive_collection:get_drive('Disk1')
    lu.assertEquals(drive.Id, 1)
end

-- 测试物理盘dump日志
function TestStorage:test_drive_dump_info()
    -- 添加物理盘对象
    local drive_info, drive = create_drive(self)
    lu.assertNotIsNil(drive)
    lu.assertNotIsNil(drive_info)
    local obj_info_path = project_dir .. '/test/unit/drive_info.txt'
    local fp_w = file_sec.open_s(obj_info_path, 'w+')
    lu.assertNotIsNil(fp_w)

    -- 调用物理盘收集日志到桩文件中
    drive:dump_physical_drive(fp_w)
    fp_w:close()

    local file = file_sec.open_s(obj_info_path, 'r')
    lu.assertNotIsNil(file)
    local content = utils.close(file, pcall(file.read, file, '*a'))
    lu.assertNotIsNil(content)
    -- 检查物理盘收集的信息
    lu.assertNotEquals(string.find(content, 'On Hours                           : N/A'), nil)

    utils.remove_file(obj_info_path)
end

-- 测试增加磁盘事件
function TestStorage:test_drive_add_event()
    local event = {
        path = '/bmc/kepler/Systems/1/Events',
        AddEvent_PACKED = function()
            return {
                unpack = function()
                    return 1
                end,
            }
        end,
    }
    client.ForeachEventsObjects = function(self, cb)
        cb(event)
    end
    local ret = pcall(add_event.generate_drive_max_temp, add_event, 'true', 60)
    lu.assertIsTrue(ret)

    ret = pcall(add_event.generate_percentage_threshold, add_event, 'true', 'Disk0')
    lu.assertEquals(ret, true)
    ret = pcall(add_event.generate_nvme_link_fault, add_event, 'true', 'Disk0')
    lu.assertEquals(ret, true)
end

--测试更新nvme盘数据
function TestStorage:test_update_nvme_mi_info()
    c_nvme.SerialNumber = 'N/A'
    c_nvme.vpd_nvme_mi_get_product_info = function()
        return string.char(140)
    end

    c_nvme:update_static_prop()
    lu.assertEquals(c_nvme.SerialNumber, 'N/A')
    c_nvme.vpd_nvme_mi_get_product_info = function()
        return string.char(120)
    end
    c_nvme:update_static_prop()
    lu.assertEquals(c_nvme.SerialNumber, 'x')

    local ret = pcall(c_nvme.vpd_nvme_mi_get_vender_id, c_nvme, 'INTEL')
    lu.assertIsTrue(ret)

    ret = pcall(c_nvme.vpd_nvme_mi_get_vender_id, c_nvme, 'Samsung')
    lu.assertIsTrue(ret)

    local nvme = {
        SSDChip = 1,
    }
    ret = pcall(c_nvme.get_firmeware_version, nvme)
    lu.assertIsTrue(ret)
end

function TestStorage:test_update_dynamic_drive_info()
    c_nvme.TemperatureCelsius = 100
    c_nvme.PredictedMediaLifeLeftPercent = 50
    c_nvme.PowerOnHours = 0
    c_nvme.MediaErrorCount = 0
    c_nvme.Revision = c_nvme:get_firmeware_version() or 'N/A'
    c_nvme.link_fault = false
    c_nvme.Status = 0
    c_nvme.LifeUsedPercentage = 0
    local dynamic_info = c_nvme:update_dynamic_drive_info()
    lu.assertEquals(dynamic_info.Status, 0)
    lu.assertEquals(dynamic_info.LifeUsedPercentage, 0)
end

function TestStorage:test_vpd_nvme_mi_get_info()
    c_nvme.update_static_prop = function(c_nvme)
        c_nvme.nvme_mi_info = {}
        c_nvme.Model = 'Huawei'
        c_nvme.SerialNumber = ''
        c_nvme.Manufacturer = ''
        c_nvme.ManufacturerId = 0
        c_nvme.CapableSpeedGbs = 0
        c_nvme.NegotiatedSpeedGbs = 0
        c_nvme.CapacityMiB = 0
        c_nvme.MediaType = 0
        c_nvme.Protocol = 0
        c_nvme.ResourceId = 0
        c_nvme.Status = 0
        c_nvme.LifeUsedPercentage = 0
    end
    local nvme_mi_info = c_nvme:vpd_nvme_mi_get_info()
    lu.assertEquals(nvme_mi_info.Status, 0)
    lu.assertEquals(nvme_mi_info.LifeUsedPercentage, 0)
end

function TestStorage:test_c_nvme_ctor()
    c_nvme:ctor()
    lu.assertEquals(c_nvme.Status, 255)
    lu.assertEquals(c_nvme.LifeUsedPercentage, 255)
end

-- 测试更新nvme盘资源归属
function TestStorage:test_update_nvme_resource_id()
    c_nvme.nvme_info = {}
    c_nvme.RefComponent = {}
    local socket_id = 1
    mdb.get_sub_objects = function(...)
        return {
            {
                SlotID = 0,
                ComponentType = 0,
                SocketID = socket_id - 1,
            },
        }
    end

    mdb.get_object = function(...)
        return {
            Instance = 0,
            Type = 0,
        }
    end
    c_nvme:update_res_id()
    lu.assertEquals(c_nvme.nvme_info.ResouceId, socket_id)

    socket_id = 0XFF
    c_nvme:update_res_id()
    lu.assertEquals(c_nvme.nvme_info.ResouceId, socket_id - 1)
end

-- 测试返回硬盘慢盘参数
function TestStorage:test_get_drives_io_diagnose_info()
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'
    self.drive_collection.get_all_drives = function()
        return { drive }
    end

    local ret =
        pcall(rpc_service_subhealth.get_drives_subhealth_diagnose_info, rpc_service_subhealth, 'IODeterioration')
    lu.assertEquals(ret, true)
end

-- 测试获取硬盘寿命信息
function TestStorage:test_get_drives_lifespan_info()
    create_drive(self)
    self.tasks:run_all_task()

    local ok, ret =
        pcall(rpc_service_subhealth.get_drives_subhealth_diagnose_info, rpc_service_subhealth, 'EstimatedLifespan')
    lu.assertEquals(ok, true)
    lu.assertNotIsNil(ret)
end

-- 测试手动获取硬盘日志信息
function TestStorage:test_start_collect_log()
    local ctx = construct_ctx()
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'
    self.drive_collection.get_all_drives = function()
        return { drive }
    end
    skynet.fork_once = function(cb, ...)
        return cb(...)
    end
    rpc_service_subhealth.log_collect_status = {
        global = false,
        drives = {},
    }

    local ok, _ = pcall(rpc_service_subhealth.start_collect_log, rpc_service_subhealth, ctx, 'Disk1')
    lu.assertEquals(ok, true)
    ok = pcall(rpc_service_subhealth.start_collect_log, rpc_service_subhealth, ctx, '*')
    lu.assertEquals(ok, true)

    -- 模拟已经开始收集所有日志
    rpc_service_subhealth.log_collect_status.global = true
    ok = pcall(rpc_service_subhealth.start_collect_log, rpc_service_subhealth, ctx, '*')
    lu.assertEquals(ok, true)

    -- 模拟已经开始收集单个日志
    rpc_service_subhealth.log_collect_status.global = false
    rpc_service_subhealth.log_collect_status.drives = { ['Disk1'] = true }
    ok = pcall(rpc_service_subhealth.start_collect_log, rpc_service_subhealth, ctx, 'Disk1')
    lu.assertEquals(ok, true)
end

function TestStorage:test_update_drive_static_info_by_sepecific_obj()
    local _, drive = create_drive(self)
    local obj = {
        update_drive_info = function()
            local info = {}
            info.Model = 'Huawei'
            info.SerialNumber = ''
            info.Manufacturer = ''
            info.ManufacturerId = 0
            info.CapableSpeedGbs = 0
            info.NegotiatedSpeedGbs = 0
            info.CapacityMiB = 0
            info.MediaType = 0
            info.Protocol = 0
            info.ResourceId = 0
            info.Status = 0
            info.LifeUsedPercentage = 0
            return info
        end,
    }
    drive:update_drive_static_info_by_sepecific_obj(obj)
    lu.assertEquals(drive.Status, 0)
    lu.assertEquals(drive.LifeUsedPercentage, 0)

    drive.LinkFault = false
    obj = {
        update_dynamic_drive_info = function()
            local info = {}
            info.TemperatureCelsius = 35
            info.PredictiveFailure = 0
            info.PredictedMediaLifeLeftPercent = 100
            info.PowerOnHours = 0
            info.MediaErrorCount = 0
            info.Revision = ''
            info.Failure = 0
            info.SpareBlockPercentage = 255
            info.FaultLed = 0
            info.link_fault = false
            info.Status = 0
            info.LifeUsedPercentage = 0
            return info
        end,
    }
    drive:update_dynamic_info(obj)
    lu.assertEquals(drive.TemperatureCelsius, 35)
    local storageconfig = self.c_storageconfig
    storageconfig.power_on = 0
    drive:update_dynamic_info(obj)
    lu.assertEquals(drive.TemperatureCelsius, 255)
end

--测试更新nvme盘smartlog
function TestStorage:test_update_nvme_smart_info()
    local _, drive = create_drive(self)
    c_nvme.get_nvme_smart_info = function()
        return {
            avail_spare = 8,
            critical_warning = 1,
            percent_used = 64,
            power_on_hours = 952,
        }
    end
    c_nvme.get_nvme_hw_defined_smart_info = function() end
    c_nvme.nvme_mi_mctp_obj = {
        nvme_mi_obj = {},
    }
    drive:update_nvme_smart_log_info({})
    lu.assertEquals(drive.CriticalWarning, 0xff)
    drive:update_nvme_smart_log_info(c_nvme)
    lu.assertEquals(drive.CriticalWarning, 1)
    lu.assertEquals(drive.AvailableSpare, 8)
    lu.assertEquals(drive.TLCSpareBlockPercentage, 8)
    lu.assertEquals(drive.UsedPercentage, 64)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, 36)
    lu.assertEquals(drive.PowerOnHours, 952)
end

-- 测试更新nvme盘hwdefined smartlog
function TestStorage:test_update_nvme_hw_defined_smart_info()
    local _, drive = create_drive(self)
    local nvme = {
        get_nvme_hw_defined_smart_info = function()
            return {
                remnant_wearout = 8,
                slc_avg_ec = 1,
                tlc_avg_ec = 64,
                slc_spare_block = 98,
                tlc_spare_block = 99,
            }
        end,
        nvme_mi_mctp_obj = {
            nvme_mi_obj = {},
        },
        get_nvme_smart_info = function() end,
    }

    drive:update_nvme_smart_log_info(nvme)
    lu.assertEquals(drive.PredictedMediaLifeLeftPercent, 8)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.is_support_hw_defined, 1)
    lu.assertEquals(drive.TLCSpareBlockPercentage, 99)
    lu.assertEquals(drive.SLCSpareBlockPercentage, 98)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.slc_avg_ec, 1)
    lu.assertEquals(drive.estimated_remaining_lifespan_info.tlc_avg_ec, 64)
end

-- 测试硬盘冗余信息监控功能
function TestStorage:test_spare_black_to_maintaince_log()
    local _, drive = create_drive(self)
    drive.SLCSpareBlockPercentage = 100
    drive.TLCSpareBlockPercentage = 100

    local spareblock = {
        slc_value = 70,
        tlc_value = 70,
    }

    drive.Name = 'Disk1'
    drive.last_record_spare_block = 0
    -- 测试传入无效硬盘id
    local ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 0, 100, 75)
    lu.assertEquals(ret, true)
    -- 测试传入用户区与非用户区冗余信息均越界
    ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 1, 101, 101)
    lu.assertEquals(ret, true)
    -- 测试debug模式及冗余块达到70%
    ret = pcall(rpc_service_drive.mock_record_spare_block, rpc_service_drive, 1, 100, 75)
    lu.assertEquals(ret, true)

    -- 测试正常更新流程及冗余块达到50%
    spareblock.tlc_value = 50
    spareblock.slc_value = 50
    drive:record_maintaince_log_for_spare_black(spareblock, true)

    -- 测试冗余块达到30%
    spareblock.tlc_value = 30
    spareblock.slc_value = 30
    drive:record_maintaince_log_for_spare_black(spareblock, false)

    -- 测试用户区冗余信息更新
    spareblock.tlc_value = 30
    spareblock.slc_value = common_def.INVALID_U8
    drive:record_maintaince_log_for_spare_black(spareblock, false)

    -- 测试24小时内再次更新
    drive.spare_block_last_check_time = math.floor(os.time())
    drive:record_maintaince_log_for_spare_black(spareblock, true)
end

function TestStorage:test_update_smbios_status()
    c_nvme.new = function()
        c_nvme:ctor()
        c_nvme:init()
    end
    c_nvme.super.init = function() end
    c_nvme.new()
    c_nvme.Slot = 1
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'
    self.drive_collection.get_drive = function()
        return drive
    end
    self.drive_collection.on_smbios_status_changed:emit(true)
    self.drive_collection.on_add_nvme:emit(c_nvme)
    self.drive_collection.on_smbios_status_changed:emit(true)
    c_nvme.protocol = common_def.NVME_VPD_PROTOCOL_NVME_MI
    self.drive_collection.on_smbios_status_changed:emit(true)
    c_nvme.check_support_mctp = function()
        return true
    end
    client.GetPCIeDeviceObjects = function()
        return {}
    end
    self.drive_collection.on_smbios_status_changed:emit(true)
    lu.assertEquals(drive.AvailableSpare, common_def.INVALID_U8)
    lu.assertEquals(drive.CriticalWarning, common_def.INVALID_U8)
    lu.assertEquals(drive.UsedPercentage, common_def.INVALID_U8)
end

function TestStorage:test_dump_log()
    local path = '/test/'
    storageApp.drive_collection = { nvme_list = {} }
    local ret = pcall(storageApp.dump_log, storageApp, nil, path)
    lu.assertEquals(ret, true)
end

function TestStorage:test_update_drive_sas_smart_info()
    local _, drive = create_drive(self)
    local pd_sas_smart_info = {
        strip_temperature = 2,
        glist_len = 2,
        plist_len = 77,
        manufacture_data = '2',
        blocks_sent = 2,
        blocks_received = 2,
        minutes_left = 2,
    }
    drive:update_drive_sas_smart_info(pd_sas_smart_info)
    local db = c_storageconfig.get_instance().db
    local db_drive_info = db:select(db.Drive):where(db.Drive.Id:eq(drive.Id)):first()
    lu.assertEquals(db_drive_info.ElementsInPrimaryDefectList, 77)
end

function TestStorage:test_update_pd_sas_smart_data_spec()
    -- timeout_call打桩
    local function mock_timeout_call(service, type, funcname, sch_prio, ctrl_idx, ...)
        local info = sml_cmd.get_pd_sas_smart_info_spec(ctrl_idx, ...)
        return info
    end
    sml.modify_timeout_call(mock_timeout_call)

    -- 模拟正确获取
    local _, drive = create_drive(self)
    local smart_info = drive:update_pd_sas_smart_data_spec(0x00)

    lu.assertEquals(smart_info.blocks_sent, 0)
    lu.assertEquals(smart_info.glist_len, 0)
    lu.assertEquals(smart_info.plist_len, 0)
    lu.assertEquals(smart_info.minutes_left, 0)
    lu.assertEquals(smart_info.manufacture_data, '')
    lu.assertEquals(smart_info.blocks_received, 0)
    lu.assertEquals(smart_info.strip_temperature, 0)

    -- 模拟错误场景
    local _, drive_error = create_drive(self)
    sml.get_pd_sas_smart_info_spec = {}
    local _, smart_info_error = pcall(drive_error.update_pd_sas_smart_data_spec, drive_error, 0)
    lu.assertEquals(smart_info_error, false)
end

function TestStorage:test_set_log_auto_collect_config()
    local req = {}
    local _, res = pcall(function()
        return c_ipmi_service.set_log_auto_collect_config(req, construct_ctx())
    end)
    -- 测试ManufacturerId为空，返回错误
    lu.assertEquals(res.CompletionCode, 199)

    req.ManufacturerId = 0x0007DB
    req.Data = '224'
    local _, res = pcall(function()
        return c_ipmi_service.set_log_auto_collect_config(req, construct_ctx())
    end)
    -- 测试 请求参数硬盘日志自动收集开启状态为2，返回错误
    lu.assertEquals(res.CompletionCode, 199)

    req.Data = '1256'
    local _, res = pcall(function()
        return c_ipmi_service.set_log_auto_collect_config(req, construct_ctx())
    end)
    -- 测试 请求参数硬盘日志自动收集周期为256，返回错误
    lu.assertEquals(res.CompletionCode, 199)

    req.Data = '124'
    local success, res = pcall(function()
        return c_ipmi_service.set_log_auto_collect_config(req, construct_ctx())
    end)
    lu.assertEquals(success, true)
    -- 测试 设置开启状态为1，收集周期为24，正常返回
    lu.assertEquals(res.CompletionCode, 0)
end

function TestStorage:test_get_log_auto_collect_config()
    local req = {}
    local _, res = pcall(function()
        return c_ipmi_service.get_log_auto_collect_config(req, construct_ctx())
    end)
    -- 测试ManufacturerId为空，返回错误
    lu.assertEquals(res.CompletionCode, 199)

    req.ManufacturerId = 0x0007DB
    drives_object.get_instance().LogAutoCollectEnable = function()
        return false
    end
    drives_object.get_instance().LogAutoCollectInterval = function()
        return 24
    end
    local _, res = pcall(function()
        return c_ipmi_service.get_log_auto_collect_config(req, construct_ctx())
    end)
    -- 测试开启状态false，收集周期24，正常返回
    lu.assertEquals(res.CompletionCode, 0)
end

function TestStorage:test_start_drive_collect_log()
    local req = {}
    local _, res = pcall(function()
        return c_ipmi_service.start_drive_collect_log(req, construct_ctx())
    end)
    -- 测试ManufacturerId为空，返回错误
    lu.assertEquals(res.CompletionCode, 199)

    req.ManufacturerId = 0x0007DB
    local _, drive = create_drive(self)
    drive.Name = 'Disk1'

    self.drive_collection.get_all_drives = function()
        return {}
    end
    local _, res = pcall(function()
        return c_ipmi_service.start_drive_collect_log(req, construct_ctx())
    end)
    -- 测试 查询硬盘对象列表为空，返回错误
    lu.assertEquals(res.CompletionCode, 213)

    req.DriveId = 255
    self.drive_collection.get_all_drives = function()
        return { drive }
    end
    local _, res = pcall(function()
        return c_ipmi_service.start_drive_collect_log(req, construct_ctx())
    end)
    -- 测试 收集所有硬盘日志，正常返回
    lu.assertEquals(res.CompletionCode, 0)

    req.DriveId = 1
    local _, res = pcall(function()
        return c_ipmi_service.start_drive_collect_log(req, construct_ctx())
    end)
    -- 测试 收集所有硬盘日志，正常返回
    lu.assertEquals(res.CompletionCode, 0)
end

function TestStorage:test_on_import()
    local ctx = {}
    local customize_data = {
        ['BMCSet_DrivesLogAutoCollectEnable'] = { Value = 0 },
        ['BMCSet_DrivesLogCollectInterval'] = { Value = 1 },
    }
    local ret = pcall(c_customize_config.on_import, ctx, customize_data)
    lu.assertEquals(ret, true)

    local customize_data = {
        ['BMCSet_DrivesLogAutoCollectEnable'] = { Value = 2 },
        ['BMCSet_DrivesLogCollectInterval'] = { Value = 256 },
    }
    local ret = pcall(c_customize_config.on_import, ctx, customize_data)
    lu.assertEquals(ret, true)

    local customize_data = {
        ['BMCSet_DrivesLogAutoCollectEnable'] = { Value = '' },
        ['BMCSet_DrivesLogCollectInterval'] = { Value = '' },
    }
    local ret = pcall(c_customize_config.on_import, ctx, customize_data)
    lu.assertEquals(ret, true)

    local customize_data = {
        ['BMCSet_DrivesLogAutoCollectEnable'] = { Value = 0 },
        ['BMCSet_DrivesLogCollectInterval'] = { Value = '' },
    }
    local ret = pcall(c_customize_config.on_import, ctx, customize_data)
    lu.assertEquals(ret, true)

    local customize_data = {}
    local ret = pcall(c_customize_config.on_import, ctx, customize_data)
    lu.assertEquals(ret, true)
end

function TestStorage:test_on_export()
    local ctx = {}
    local ret = pcall(c_customize_config.on_export, ctx, {})
    lu.assertEquals(ret, true)
end

function TestStorage:test_export()
    local ctx = {}
    local export_type = 'custom'
    local ret = pcall(c_export_import_engine.export, ctx, export_type)
    lu.assertEquals(ret, true)
end

function TestStorage:test_import()
    local ctx = {}
    local config_string = '{"ConfigData":{"CustomSettings":{"BMCSet_DrivesLogCollectInterval":{"Value":1},\
        "BMCSet_DrivesLogAutoCollectEnable":{"Value":0}}}}'
    local import_type = 'custom'
    local ret = pcall(c_export_import_engine.import, ctx, config_string, import_type)
    lu.assertEquals(ret, true)
end