-- 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 cmn = require 'common'
local c_device_service = require 'device.device_service'
local c_pcie_device = require 'device.class.pcie_device'
local c_ocp_card = require 'device.class.ocp_card'
local c_pcie_card = require 'device.class.pcie_card'
local c_pcie_card_flash_checker = require 'device.class.pcie_card_flash_checker'
local utils = require 'mc.utils'
local client = require 'pcie_device.client'
local log = require 'mc.logging'
local c_pcie_addr_info = require 'biz_topo.class.pcie_addr_info'
local event = require 'infrastructure.event'
local cjson = require 'cjson'

TestDeviceService = {}

local DATA_PATH<const> = './test_data/'
local PCIECARD_INTERFACE <const> = 'bmc.kepler.Systems.PCIeDevices.PCIeCard'

function TestDeviceService:test_method_card_info_dump()
    client.GetPackageObjects = function()
        return {}
    end
    local pcie_card_list = {
        [1] = {
            get_prop = function(_, prop)
                local props = {
                    SlotID = 3,
                    LaneOwner = 0,
                    VendorID = 4096,
                    DeviceID = 4322,
                    SubVendorID = 4096,
                    SubDeviceID = 16400,
                    Description = 'MegaRAID 9560-8i',
                    BoardID = 0,
                    PcbVersion = 'N/A',
                    PartNumber = '06030622',
                    Manufacturer = 'Broadcom',
                    Position = 'PCIeRiser1',
                    Name = '9560-8i'
                }
                return props[prop]
            end
        }
    }
    local ocp_card_list = {
        [1] = {
            get_prop = function(_, prop)
                local props = {
                    SlotID = 3,
                    LaneOwner = 0,
                    VendorID = 4097,
                    DeviceID = 4323,
                    SubVendorID = 4097,
                    SubDeviceID = 16401,
                    Description = 'MT2892 Family [ConnectX-6 Dx]',
                    BoardID = 65535,
                    PcbVersion = '.A',
                    PartNumber = '06310299',
                    Manufacturer = 'Mellanox',
                    Position = 'EXU',
                    Name = 'MCX623436MN-CDAB'
                }
                return props[prop]
            end
        }
    }
    local cpu_table = {
        {LogicalId = 2, PhysicalId = 3},
        {LogicalId = 0, PhysicalId = 1}
    }
    local c_ForeachCPUObjects = client.ForeachCPUObjects
    client.ForeachCPUObjects = function(self, cb)
        for _, obj in pairs(cpu_table) do
            cb(obj)
        end
    end

    c_device_service.pcie_device_list = {}
    c_device_service.pcie_card_list = pcie_card_list
    c_device_service.ocp_card_list = ocp_card_list
    c_device_service:method_card_info_dump(nil, utils.realpath(DATA_PATH))
    client.ForeachCPUObjects = c_ForeachCPUObjects
    local test_file_path = DATA_PATH .. 'card_info'
    local file = io.open(test_file_path, 'r')
    local str = file:read("a")
    file:close()
    lu.assertEquals(str:sub(1, 14), 'Pcie Card Info')
    lu.assertEquals(str:sub(798, 810), 'OCP Card Info')
    os.execute('rm -f ' .. test_file_path)
end

function TestDeviceService:test_on_add_ocp_card()
    c_device_service.ocp_card_list = {}
    c_device_service:on_add_object('OCPCard', {
        PcbID = 1,
        PcbVersion = ''
    }, 'position')
    lu.assertEquals(#c_device_service.ocp_card_list, 1)

    local pcie_device_obj = c_pcie_device.new({
        DeviceName = "pcie-device",
        DeviceType = 8,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        property_changed = {
            on = function()
            end
        }
    }, "position", {})

    cmn.skynet.fork_loop = function(_, cb)
        cb()
    end
    c_device_service:register_get_pcie_info_task(pcie_device_obj, "position")

    local position = "position1"
    local mdb_obj = {
        DeviceName = "ocp1",
        DeviceType = 8,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        PcbID = 1,
        PcbVersion = ''
    }
    local bus = {}
    local reset_local_db = {}
    local ocp_obj = c_ocp_card.new(mdb_obj, position, bus, reset_local_db)
    ocp_obj.get_pcie_lang_info = function()
    end
    ocp_obj.sync_info_to_card = function()
    end
    ocp_obj.update_pcie_vpd_info = function()
    end
    local sleep_bak = cmn.skynet.sleep
    cmn.skynet.sleep = function()
        log:raise("break")
    end
    local ok, res = pcall(function()
        c_device_service.pcie_and_ocp_task(ocp_obj, {ocp_obj}, "position1")
    end)
    lu.assertEquals(ok, false)
    lu.assertNotEquals(res, nil)
    cmn.skynet.sleep = sleep_bak
end

function TestDeviceService:test_sync_eth_device_bdf_info()
    local addr_obj = {
        get_prop = function() return true end
    }
    local biz_topo = {
        pfid_map = {[1] = {[0] = 0, [1] = 4}}
    }
    c_device_service.service_manager = {
        get_service = function() return {
            biz_topo = biz_topo,
            method_get_target_conn_by_addr_obj = function() return true end
        } end,
    }
    obj_list = {
        ethobj1 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 0, sync_ready = true,
            mds_obj = { name = "port1", PortId = 0, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj2 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 0, sync_ready = true,
            mds_obj = { name = "port2", PortId = 1, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj3 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 1, sync_ready = true,
            mds_obj = { name = "port3", PortId = 2, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj4 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 1, sync_ready = true,
            mds_obj = { name = "port4", PortId = 3, Device = 0,
                Function = 0, PortFunctionId = 0 }
        }
    }
    cmn.skynet.fork = function(cb) cb() end
    c_device_service:sync_eth_device_bdf_info(obj_list, addr_obj)
    lu.assertEquals(obj_list.ethobj1.mds_obj.PortFunctionId, 0)
    lu.assertEquals(obj_list.ethobj2.mds_obj.PortFunctionId, 1)
    lu.assertEquals(obj_list.ethobj3.mds_obj.PortFunctionId, 4)
    lu.assertEquals(obj_list.ethobj4.mds_obj.PortFunctionId, 5)
    lu.assertEquals(obj_list.ethobj1.mds_obj.Function, 0)
    lu.assertEquals(obj_list.ethobj2.mds_obj.Function, 1)
    lu.assertEquals(obj_list.ethobj3.mds_obj.Function, 0)
    lu.assertEquals(obj_list.ethobj4.mds_obj.Function, 1)
end

function TestDeviceService:test_sync_pcie_addr_info_to_device()
    local pcie_addr_info_obj = c_pcie_addr_info.new({
        ComponentType = 83,
        SlotID = 1,
        path = '/bmc/kepler/Systems/1'
    }, "position", {})

    c_device_service.service_manager = {
        get_service = function()
            return {
                biz_topo = {
                    get_objs = function()
                        return {
                            pcie_addr_info_obj
                        }
                    end
                }
            }
        end
    }
    cmn.skynet.fork = function(cb)
        cb()
    end
    local pcie_device_obj = c_pcie_device.new({
        DeviceName = "pcie-device",
        DeviceType = 83,
        SlotID = 1,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        property_changed = {
            on = function()
            end
        }
    }, "position", {})

    pcie_device_obj.sync_pcie_addr_info = function()
    end
    c_device_service.update_pcie_type = function()
    end
    local sleep_bak = cmn.skynet.sleep
    cmn.skynet.sleep = function()
        log:raise("break")
    end
    local ok, res = pcall(function()
        c_device_service:sync_pcie_addr_info_to_device(pcie_device_obj)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(res, nil)
    cmn.skynet.sleep = sleep_bak
end

function TestDeviceService:test_on_add_flash_checker()
    c_device_service.flash_checker_list = {}
    c_device_service:on_add_object('PCIeCardFlashChecker', {
        prop = ''
    }, 'position')
    lu.assertEquals(#c_device_service.flash_checker_list, 1)

    c_device_service:on_delete_object('PCIeCardFlashChecker', {
        prop = ''
    }, 'position')
    lu.assertEquals(#c_device_service.flash_checker_list, 0)
end

function TestDeviceService:test_update_flash_checker()
    local position = 'position'
    local mdb_obj = {
        PcbID = 1,
        PcbVersion = '',
        SerialNumber = '',
        PartNumber = '',
        CardForServers = {},
        LinkSpeed = '',
        LinkSpeedCapability = '',
        LinkWidth = '',
        LinkWidthAbility = '',
        Protocol = '',
        MaxFrameLen = '',
        RefChip = '',
        DeviceName = "pcie-card",
        DeviceType = 8,
        VendorID = 0x1111,
        DeviceID = 0x2222,
        SlotID = 1,
        [PCIECARD_INTERFACE] = {
        }
    }
    local bus = {}
    local reset_local_db = {}
    local pcie_card_obj = c_pcie_card.new(mdb_obj, position, bus, reset_local_db)

    local pcie_card_flash_checker_obj = c_pcie_card_flash_checker.new({
        ["FaultStatus"] = 0,
        ["PCIeQuadrupleList"] = {{'1111_2222','card-des'}}
    }, "position", {})

    local flash_checker_list = {pcie_card_flash_checker_obj}

    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 1)

    pcie_card_obj:set_prop('VendorID', 123)
    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 0)

    flash_checker_list = {}
    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 0)
end

function TestDeviceService:test_update_init_abnormal_event()
    local get_latest_alarm_list_bak = event.get_latest_alarm_list
    event.get_latest_alarm_list = function()
        return {{
            EventCode = '0x080000A5',
            MessageArgs = cjson.encode({'', 1, 'card description', "Flash fault"}),
        }}
    end

    local bus = {}
    local status = 1
    local slot = 0
    local card_des = 'card description'
    local device_name = 'card device_name'
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    status = 0
    slot = 1
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    local generate_event_bak = cmn.generate_event
    cmn.generate_event = function()
    end
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    cmn.generate_event = function()
        log:raise('error')
    end
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    status = 1
    slot = 2
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    cmn.generate_event = function()
    end
    local ok, res = pcall(function()
        c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(res, nil)
    
    event.get_latest_alarm_list = get_latest_alarm_list_bak
    cmn.generate_event = generate_event_bak
end

function TestDeviceService:test_clear_flash_checker_event()
    local sleep_bak = cmn.skynet.sleep
    local update_init_abnormal_event_bak = c_device_service.update_init_abnormal_event
    local get_latest_alarm_list_bak = c_device_service.get_latest_alarm_list
    cmn.skynet.sleep = function()
    end
    c_device_service.update_init_abnormal_event = function()
    end
    c_device_service.get_latest_alarm_list = function()
        return {[1] = true}
    end

    local ok, _ = pcall(function()
        c_device_service.clear_flash_checker_event({}, 0)
    end)
    lu.assertEquals(ok, true)
    cmn.skynet.sleep = sleep_bak
    c_device_service.update_init_abnormal_event = update_init_abnormal_event_bak
    c_device_service.get_latest_alarm_list = get_latest_alarm_list_bak
end