-- 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 dpu_object = require 'dpu_service.dpu_object'
local dpu_service = require 'dpu_service.dpu_service'
local log_collector = require 'log_collector'
local utils = require 'mc.utils'
local utils_core = require 'utils.core'
local file_sec = require 'utils.file'
local ipmi = require 'ipmi'
local cc = ipmi.types.Cc
local event = require 'infrastructure.event'
local context = require 'mc.context'
local log = require 'mc.logging'
local sd_bus = require 'sd_bus'

TestDpuObj = {}

local DPUCARD_INTF<const> = 'bmc.kepler.Systems.DPUCard'
local PCIECARD_INTF<const> = 'bmc.kepler.Systems.PCIeDevices.PCIeCard'
local DPUCARD_FAULT_INTF<const> = 'bmc.kepler.Systems.DPUCard.Fault'
local DPUCARD_METRICS_INTF<const> = 'bmc.kepler.Systems.DPUCard.Metrics'
local TLV_OS_STATUS<const> = 6
local TLV_POWER_STATUS<const> = 1 -- 获取DPU卡电源状态

local board_name = 'IT21SHSU'
local description = 'test description information'
local board_id = 1

local mds_obj = {
    BoardID = 165,
    [DPUCARD_METRICS_INTF] = {
        PowerWatts = 0,
        CPUTemperatureCelsius = 0,
        SFP1TemperatureCelsius = 0,
        SFP2TemperatureCelsius = 0,
        Inlet1TemperatureCelsius = 0,
        Outlet1TemperatureCelsius = 0,
    },
    [DPUCARD_FAULT_INTF] = {
        HeartBeatLoss = 0
    },
    [DPUCARD_INTF] = {
        MCUVersion = '0.0.0',
        LogicVersion = '0.0',
        VrdVersion = '0',
        BootSourceOverrideMode = 0,
        BootSourceOverrideEnabled = 0,
        PxeOption = '0',
        UUID = '0',
        M2SlotPresence = 0,
        PfMacInfo = {},
        StorageIpAddr = '0.0.0.0',
        StorageIpVlan = 0,
        SystemLoadedStatus = 0
    },
    [PCIECARD_INTF] = {
        SlotID = 1,
        BoardName = board_name,
        BoardID = board_id,
        Description = description,
        NodeID = 'PCIeCard0'
    },
    MRCLogLevel = 1,
    UEFILogLevel = 1,
    PowerState = '',
    MPUBusyStatus = 1,
    RefChip = require 'test_data.dpu_ref_chip'
}

function TestDpuObj:setupClass()
    self.dpu_object = dpu_object.new(mds_obj, '010101')
end

function TestDpuObj:teardownClass()
end

function TestDpuObj:test_basic_info()
    lu.assertEquals(self.dpu_object.pciecard.BoardName, board_name)
    lu.assertEquals(self.dpu_object:get_description(), description)
    lu.assertEquals(self.dpu_object:get_board_id(), board_id)
end

function TestDpuObj:test_set_sys_time()
    local sys_time = {
        year = 2024,
        month = 8,
        day = 1,
        hour = 23,
        min = 16,
        sec = 29,
        wday = 5
    }
    local ret = self.dpu_object:set_sys_time(sys_time)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_fetch_mcu_fw_ver()
    self.dpu_object:fetch_mcu_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.MCUVersion, '1.2.3')
end

function TestDpuObj:test_update_prop_retry()
    self.dpu_object:update_prop_retry()
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 1)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 2)
    lu.assertEquals(self.dpu_object.inner.PowerState, 'On')
end

function TestDpuObj:test_set_property()
    self.dpu_object:set_property('MRCLogLevel', 2)
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 2)
    self.dpu_object:set_property('UEFILogLevel', 3)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 3)
    self.dpu_object:set_property('PowerState', 'On')
    lu.assertEquals(self.dpu_object.inner.PowerState, 'On')
end

function TestDpuObj:test_set_bios_log_prop()
    self.dpu_object:set_bios_log_prop(1, 3)
    lu.assertEquals(self.dpu_object.inner.MRCLogLevel, 3)
    self.dpu_object:set_bios_log_prop(2, 4)
    lu.assertEquals(self.dpu_object.inner.UEFILogLevel, 4)
end

function TestDpuObj:test_set_bios_log_level()
    local type = 1
    local level = 1
    local ret = self.dpu_object:set_bios_log_level(type, level)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_mrc_log_level()
    local mrc_log_level = self.dpu_object:get_mrc_log_level()
    lu.assertEquals(mrc_log_level.level, 1)
end

function TestDpuObj:test_get_uefi_log_level()
    local uefi_log_level = self.dpu_object:get_uefi_log_level()
    lu.assertEquals(uefi_log_level.level, 2)
end

function TestDpuObj:test_set_sol_switch()
    local sol_switch_type = 7
    local ret = self.dpu_object:set_sol_switch(sol_switch_type)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_sol_switch()
    local data = self.dpu_object:get_sol_switch()
    lu.assertEquals(data.type, 5)
    lu.assertEquals(data.value, 7)
end

function TestDpuObj:test_update_pciecard_nodeid()
    self.dpu_object:update_pciecard_nodeid()
    lu.assertEquals(self.dpu_object.pciecard.NodeID, 'PCIeCard1')
end

function TestDpuObj:test_check_heartbeat_loss()
    self.dpu_object:check_heartbeat_loss()
    lu.assertEquals(self.dpu_object.dpucard_fault.HeartBeatLoss, 0)
end

function TestDpuObj:test_fetch_health()
    local ret = self.dpu_object:fetch_health()
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_generate_alarm()
    self.dpu_object:generate_alarm('HeartBeatLoss', 3)
    lu.assertEquals(self.dpu_object.dpucard_fault.HeartBeatLoss, 3)
end

function TestDpuObj:test_fetch_power()
    self.dpu_object:fetch_power()
    lu.assertEquals(self.dpu_object.dpucard_metrics.PowerWatts, 1800)
end

function TestDpuObj:test_fetch_cpld_fw_ver()
    self.dpu_object:fetch_cpld_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.LogicVersion, '1.02')
end

function TestDpuObj:test_fetch_vrd_fw_ver()
    self.dpu_object:fetch_vrd_fw_ver()
    lu.assertEquals(self.dpu_object.dpucard.VrdVersion, '1')
end

function TestDpuObj:test_update_fw_ver()
    self.dpu_object.cpld_fw_obj = {}
    self.dpu_object.mcu_fw_obj = {}
    self.dpu_object.vrd_fw_obj = {}
    self.dpu_object:update_fw_ver()
    lu.assertEquals(self.dpu_object.mcu_fw_obj.Version, '1.2.3')
    lu.assertEquals(self.dpu_object.cpld_fw_obj.Version, '1.02')
    lu.assertEquals(self.dpu_object.vrd_fw_obj.Version, '1')
end

function TestDpuObj:test_get_cpld_manufacture_id()
    self.dpu_object:get_cpld_manufacture_id()
    lu.assertEquals(self.dpu_object.cpld_manufacture_id, 2)
end

function TestDpuObj:test_get_mcu_manufacture_id()
    self.dpu_object:get_mcu_manufacture_id()
    lu.assertEquals(self.dpu_object.mcu_manufacture_id, 0xA0)
end

function TestDpuObj:test_get_slot_id()
    local slot_id = self.dpu_object:get_slot_id()
    lu.assertEquals(slot_id, 1)
    local dpu_service_obj = {
        dpu_objects = {
            [1] = {
                get_slot_id = function()
                    return 2
                end
            }
        }
    }
    slot_id = dpu_service.get_slot_id(dpu_service_obj, 1)
    lu.assertEquals(slot_id, 2)
end

function TestDpuObj:test_query_upgrade_status()
    local status = self.dpu_object:query_upgrade_status()
    lu.assertEquals(status, 1)
end

function TestDpuObj:test_fetch_mcu_reset_time()
    self.dpu_object:fetch_mcu_reset_time()
    lu.assertEquals(self.dpu_object.mcu_reset_time, 16)
end

function TestDpuObj:test_get_dpu_boot_option()
    local ret, data = self.dpu_object:get_dpu_boot_option()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.save_option, 1)
    lu.assertEquals(data.boot_option, 2)
end

function TestDpuObj:test_fetch_boot_option()
    self.dpu_object:fetch_boot_option()
    lu.assertEquals(self.dpu_object.dpucard.BootSourceOverrideMode, 2)
    lu.assertEquals(self.dpu_object.dpucard.BootSourceOverrideEnabled, 1)
end

function TestDpuObj:test_get_dpu_pxe_option()
    local ret, data = self.dpu_object:get_dpu_pxe_option()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.pxe_option, 4)
end

function TestDpuObj:test_set_dpu_pxe_option()
    local pxe_option = 5
    local ret = self.dpu_object:set_dpu_pxe_option(pxe_option)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_dpu_uuid()
    local ret, uuid = self.dpu_object:get_dpu_uuid()
    lu.assertEquals(ret, true)
    lu.assertEquals(uuid, '010203')
end

function TestDpuObj:test_fetch_pxe_option()
    self.dpu_object:fetch_pxe_option()
    lu.assertEquals(self.dpu_object.dpucard.PxeOption, 'UefiIpv6')
end

function TestDpuObj:test_set_pxe_option()
    local pxe_opt = 'UefiIpv4AndIpv6'
    local ret = self.dpu_object:set_pxe_option(pxe_opt)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_fetch_dpu_uuid()
    self.dpu_object:fetch_dpu_uuid()
    lu.assertEquals(self.dpu_object.dpucard.UUID, '010203')
end

function TestDpuObj:test_set_presence()
    self.dpu_object.dpucard.M2SlotPresence = 0
    self.dpu_object:set_presence('M2SlotPresence', 1, 1)
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 1)
    self.dpu_object:set_presence('M2SlotPresence', 2, 0)
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 1)
end

function TestDpuObj:test_fetch_m2_presence()
    self.dpu_object:fetch_m2_presence()
    lu.assertEquals(self.dpu_object.dpucard.M2SlotPresence, 3)
end

function TestDpuObj:test_fetch_mac_address()
    self.dpu_object:fetch_mac_address()
    local data = self.dpu_object.dpucard.PfMacInfo
    lu.assertEquals(data[1][1], 1)
    lu.assertEquals(data[1][2], 8)
    lu.assertEquals(data[1][3], '00:11:22:33:44:55')
end

function TestDpuObj:test_get_dpu_ip()
    local ret, data = self.dpu_object:get_dpu_ip()
    local ipv4 = string.format('%d.%d.%d.%d', string.unpack('BBBB',string.pack('I', data.ipv4)))
    local mask = string.format('%d.%d.%d.%d', string.unpack('BBBB',string.pack('I', data.mask)))
    lu.assertEquals(ret, true)
    lu.assertEquals(ipv4, '192.168.1.1')
    lu.assertEquals(mask, '192.168.1.2')
    lu.assertEquals(data.vlan, 1)
end

function TestDpuObj:test_fetch_sdi_ip()
    self.dpu_object:fetch_sdi_ip()
    lu.assertEquals(self.dpu_object.dpucard.StorageIpAddr, '192.168.1.1')
    lu.assertEquals(self.dpu_object.dpucard.StorageIpVlan, 1)
end

function TestDpuObj:test_set_sdi_ipv4()
    local ip_type = 0
    local ipv4 = '192.168.2.1'
    local mask = '192.168.2.2'
    local vlan = 1
    local ret = self.dpu_object:set_sdi_ipv4(ip_type, ipv4, mask, vlan)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_sdi_ipv6()
    local ip_type = 0
    local ipv6 = '0000:0000:0000:0000:0000:0000:0000:0001'
    local prefix_length = 1
    local vlan = 1
    local ret = self.dpu_object:set_sdi_ipv6(ip_type, ipv6, prefix_length, vlan)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_sdi_slot()
    local ret = self.dpu_object:set_sdi_slot(1)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_sdi_host_os_status()
    local ret = self.dpu_object:set_sdi_host_os_status(0)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_dpu_boot_option()
    local type = 1
    local value = 2
    local ret = self.dpu_object:set_dpu_boot_option(type, value)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_reset_sdi_card()
    local ret = self.dpu_object:reset_sdi_card()
    lu.assertEquals(ret, true)
end
local function construct_ctx()
    local ctx = {}
    ctx.ChanType = 1
    ctx.get_initiator = function()
        return {}
    end

    return ctx
end

function TestDpuObj:test_set_dpu_nmi()
    local req = {}
    req.ManufactureId = 0x0007DB
    dpu_service.get_dpu_obj_by_device_no = function()
        return false
    end
    local _, res = pcall(function()
        return dpu_service:set_dpu_nmi(req, construct_ctx())
    end)
    -- 1、测试 self:get_dpu_obj_by_device_no(req.DeviceNo) 为空
    lu.assertEquals(res.CompletionCode, cc.DataNotAvailable)

    dpu_service.get_dpu_obj_by_device_no = function()
        return {pciecard = {SlotID = 1, Name = "test"}, set_dpu_nmi = function()
            return false, "test"
        end}
    end
    local _, res1 = pcall(function()
        return dpu_service:set_dpu_nmi(req, construct_ctx())
    end)
    -- 2、测试 dpu:set_dpu_nmi() 为空
    lu.assertEquals(res1.CompletionCode, cc.UnspecifiedError)
end

function TestDpuObj:test_set_dpu_nmi_rpc()
    dpu_service.get_dpu_obj_by_slot_id = function()
        return false
    end
    local ok, _ = pcall(function()
        return dpu_service:set_dpu_nmi_rpc(construct_ctx(), 1)
    end)
    -- 1、测试 self:get_dpu_obj_by_slot_id(slot_id) 为空
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_slot_id = function()
        return {pciecard = {SlotID = 1, Name = "test"}, set_dpu_nmi = function()
            return false, "test"
        end}
    end
    local ok1, _ = pcall(function()
        return dpu_service:set_dpu_nmi_rpc(construct_ctx(), 1)
    end)
    -- 2、测试 dpu:set_dpu_nmi() 为空
    lu.assertEquals(ok1, false)
end

function TestDpuObj:test_set_dpu_power_state()
    local ret = self.dpu_object:set_dpu_power_state(1)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_set_dpu_reset_linkage()
    local ret = self.dpu_object:set_dpu_reset_linkage(1)
    lu.assertEquals(ret, true)
end

function TestDpuObj:test_get_reset_linkage()
    local ret, data = self.dpu_object:get_reset_linkage()
    lu.assertEquals(ret, true)
    lu.assertEquals(data.linkage, 1)
end

function TestDpuObj:test_get_dpu_status()
    local ret, data 
    ret, data = self.dpu_object:get_dpu_status(TLV_POWER_STATUS)
    lu.assertEquals(ret, true)
    lu.assertEquals(data.status, 1)
    ret, data = self.dpu_object:get_dpu_status(TLV_OS_STATUS)
    lu.assertEquals(ret, true)
    lu.assertEquals(data.status, 3)
end

function TestDpuObj:test_get_mpu_busy_status()
    local ret, status = self.dpu_object:get_mpu_busy_status()
    lu.assertEquals(ret, true)
    lu.assertEquals(status, 1)
end

function TestDpuObj:test_fetch_os_status()
    self.dpu_object:fetch_os_status()
    lu.assertEquals(self.dpu_object.dpucard.SystemLoadedStatus, 3)
end

function TestDpuObj:test_fetch_mpu_busy_status()
    self.dpu_object:fetch_mpu_busy_status()
    lu.assertEquals(self.dpu_object.inner.MPUBusyStatus, 0)
end

function TestDpuObj:test_get_error_code()
    local resp = self.dpu_object:get_error_code()
    lu.assertEquals(resp, {})
end

function TestDpuObj:test_fetch_device_temperature()
    self.dpu_object:fetch_device_temperature()
    lu.assertEquals(self.dpu_object.dpucard_metrics.CPUTemperatureCelsius, 26)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP1TemperatureCelsius, 32)
    lu.assertEquals(self.dpu_object.dpucard_metrics.SFP2TemperatureCelsius, 48)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Inlet1TemperatureCelsius, 50)
    lu.assertEquals(self.dpu_object.dpucard_metrics.Outlet1TemperatureCelsius, 43)
end

function TestDpuObj:test_generate_mcu_event()
    local bus = {}
    self.dpu_object:generate_mcu_event(bus, 8358, 'true')
end

function TestDpuObj:test_log_clear()
    log_collector.init()
    log_collector.log_timeout = -1
    -- 用例开始前确保目录干净
    utils.remove_file(log_collector.pcie_card_log_base_dir)
    local file_names = {
        'PCIeCard5(RAID)_RAID',
        'PCIeCard5(SDIV5.0)_SDI5.0',
        'PCIeCard0(SDIV6.0)_SDI5.0',
        'PCIeCard$(SDI5.0)_SDI5.0',
        '_SDI5.0'
    }
    for _, file_name in ipairs(file_names) do
        local log_dir = log_collector.pcie_card_log_base_dir .. file_name
        log_collector.create_dir(log_dir)
        local f_log = file_sec.open_s(log_dir .. '/error_log.bin', 'w+')
        utils.close(f_log, pcall(f_log.write, f_log, 'xxx'))
    end
    -- 目录清理
    log_collector.clear_invaild_log()
    -- 检查目录是否被正确清理
    local expectd_exist_files = {
        'PCIeCard5(RAID)_RAID'
    }
    local files = utils_core.dir(log_collector.pcie_card_log_base_dir)
    lu.assertEquals(#files, #expectd_exist_files)
    table.sort(file_names, function(a, b) return a < b end)
    table.sort(expectd_exist_files, function(a, b) return a < b end)
    for index, _ in pairs(files) do
        lu.assertEquals(files[index], expectd_exist_files[index])
    end

    -- 用例执行结束需要清理残留文件
    utils.remove_file(log_collector.pcie_card_log_base_dir)
end

function TestDpuObj:test_calc_power_read_fail_duration()
    local time = self.dpu_object:calc_power_read_fail_duration()
    lu.assertEquals(time, 0)
end

function TestDpuObj:test_get_sdi_elabel()
    local expected_elabel_value = setmetatable({
        BoardManufacturer = 'Huawei',
        BoardProductName = 'IT21SHSN',
        BoardSerialNumber = '2106310363FSQ6000554',
        BoardPartNumber = '5104-00781'
    },
    {__index = function ()
        -- 未打桩的属性为空串
        return ''
    end})
    local cb = function (elabels)
        for elabel_prop, value in pairs(elabels) do
            lu.assertEquals(value, expected_elabel_value[elabel_prop])
        end
    end
    local elabel_value = self.dpu_object:get_sdi_elabel(cb)
    cb(elabel_value)
end

function TestDpuObj:test_get_latest_alarm_list()
    log.error = function()
        return true
    end
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 1、测试 Event.event_obj为空
    lu.assertEquals(ok, true)
    lu.assertEquals(res, false)

    event.event_obj = {}
    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return false, "error"
        end
    }
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 2、测试 Event.event_obj不为空GetAlarmList_PACKED为空
    lu.assertEquals(ok, true)
    lu.assertEquals(res, false)

    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return true, "error"
        end
    }
    local ok, _ = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 3、测试 GetAlarmList_PACKED返回失败
    lu.assertEquals(ok, true)

    event.event_obj.pcall ={
        GetAlarmList_PACKED = function()
        return true, {EventList = {}}
        end
    }
    local ok, res = pcall(function()
        return event:get_latest_alarm_list({})
    end)
    -- 4、测试 GetAlarmList_PACKED返回成功
    lu.assertEquals(ok, true)
    lu.assertEquals(res, true)
end

function TestDpuObj:test_update_serial_disconnect_alarm()
    local bus = {}
    log.debug = function()
        return true
    end
    dpu_object.pciecard = {name = "test"}
    local ok, _ = pcall(function()
        return dpu_object:update_serial_disconnect_alarm(bus)
    end)
    lu.assertEquals(ok, true)
end

function TestDpuObj:test_update_latest_alarm_list()
    local bus = {}
    log.debug = function()
        return true
    end
    dpu_object.pciecard = {name = "test"}
    local ok, _ = pcall(function()
        return dpu_object:update_latest_alarm_list(bus)
    end)
    lu.assertEquals(ok, true)
end