-- 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 dpu_service = require 'dpu_service.dpu_service'
local lu = require 'luaunit'
local dpu_object = require 'dpu_service.dpu_object'
local file_sec = require 'utils.file'
local utils = require 'dpu_service.utils'
local utils_core = require 'utils.core'
local skynet = require 'skynet'
local mdb = require 'mc.mdb'
local client = require 'general_hardware.client'

local COMPUTING_BOARD_PORT0<const> = 3
local COMPUTING_BOARD_PORT1<const> = 4
local COMPUTING_BOARD_PORT2<const> = 5
local LLDP_ENABLE<const> = 1
local LLDP_DISABLE<const> = 0
local LLDP_STATUS_TYPE<const> = 5
local TLV_OS_STATUS<const> = 6
local E_OK = 0 -- 函数执行成功返回0
local E_FAILED = -1

local ILLEGAL_TYPE<const> = 6
local PCIECARD_INTF <const> = 'bmc.kepler.Systems.PCIeDevices.PCIeCard'
TestDpuService = {}

function TestDpuService:setupClass()
end

function TestDpuService:teardownClass()
end


function TestDpuService:test_fetch_dpu_hardware_info()
    local mock = skynet.sleep
    skynet.sleep = function () end
    local obj = {
        fetch_bios_version = function ()
            return 'version'
        end,
        fetch_cpu_arch = function ()
            return 'cpu_arch'
        end,
        fetch_cpu_cores = function ()
            return 'cpu_cores'
        end,
        fetch_disk_size = function ()
            return 'disk_size'
        end,
        fetch_mem_size = function ()
            return 'mem_size'
        end
    }
    local ok, _ = pcall(dpu_service.fetch_dpu_hardware_info, dpu_service, obj)
    lu.assertEquals(ok, true)
    skynet.sleep = mock
end

function TestDpuService:test_is_dpu_mcu()
    local ret = dpu_service:is_dpu_mcu('type', '')
    lu.assertEquals(ret, false)

    local mock1 = utils.get_firmware_uid
    utils.get_firmware_uid =function(...)
        return false
    end
    local ret = dpu_service:is_dpu_mcu('Mcu', '')
    lu.assertEquals(ret, false)

    utils.get_firmware_uid =function(...)
        return true
    end
    local ret = dpu_service:is_dpu_mcu('Mcu', '')
    lu.assertEquals(ret, true)

    utils.get_firmware_uid = mock1
end

function TestDpuService:test_on_upgrade_prepare()
    local mock1 = file_sec.check_real_path_s
    dpu_service.upgrade_state = {}
    dpu_service.upgrade_state['DPUCpld'] = true
    local parameters = {}
    parameters.UpgradeMode = "Parallel"
    local ok, _ = pcall(dpu_service.on_upgrade_prepare, dpu_service, nil, 0, 'DPUCpld', "/tmp/", nil, parameters)
    lu.assertEquals(ok, false)

    dpu_service.upgrade_state['DPUCpld'] = false
    file_sec.check_real_path_s = function()
        return E_FAILED
    end
    local ok, _ = pcall(dpu_service.on_upgrade_prepare, dpu_service, nil, 0, 'DPUCpld', "/tmp/", nil, "param")
    lu.assertEquals(ok, true)

    file_sec.check_real_path_s = function()
        return E_OK
    end
    local mock2 = utils.uncompress_firmware
    utils.uncompress_firmware = function()
        return ''
    end
    local ok, _ = pcall(dpu_service.on_upgrade_prepare, dpu_service, nil, 0, 'DPUCpld', "/tmp/", nil, "param")
    lu.assertEquals(ok, true)

    utils.uncompress_firmware = mock2
    local ok, _ = pcall(dpu_service.on_upgrade_prepare, dpu_service, nil, 0, 'DPUCpld', "/tmp/", nil, "param")
    lu.assertEquals(ok, true)
    file_sec.check_real_path_s = mock1

    local PFileFileChownMock = client.PFileFileChown
    client.PFileFileChown = function ()
        return false
    end

    dpu_service.upgradable_cfgs = {}
    local ok, _ = pcall(dpu_service.on_upgrade_process, dpu_service, nil, 0, 'DPUCpld', "/tmp/", "param")
    lu.assertEquals(ok, true)
    local ok, _ = pcall(dpu_service.on_upgrade_finish, dpu_service, nil, 0, 'DPUCpld', "param")
    lu.assertEquals(ok, true)

    dpu_service.upgradable_cfgs = {"firmware"}
    local ok, _ = pcall(dpu_service.on_upgrade_process, dpu_service, nil, 0, 'DPUCpld', "/tmp/", "param")
    lu.assertEquals(ok, false)
    local ok, _ = pcall(dpu_service.on_upgrade_finish, dpu_service, nil, 0, 'DPUCpld', "param")
    lu.assertEquals(ok, false)
    client.PFileFileChown = PFileFileChownMock
end

function TestDpuService:test_get_LLDP_status()
    local obj = {path = 'path'}
    local ok, _ = pcall(dpu_service.get_LLDP_status, dpu_service, obj, 'ctx', COMPUTING_BOARD_PORT0)
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_path = function()
        return nil
    end
    local ok, _ = pcall(dpu_service.get_LLDP_status, dpu_service, obj, 'ctx', COMPUTING_BOARD_PORT2)
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_path = function()
        return {LLDP_status = TLV_OS_STATUS}
    end
    local ok, _ = pcall(dpu_service.get_LLDP_status, dpu_service, obj, 'ctx', COMPUTING_BOARD_PORT2)
    lu.assertEquals(ok, true)
end

function TestDpuService:test_set_LLDP_status()
    local obj = {
        path = 'path'
    }
    local ctx
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT0, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT2, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    -- dpu is nill
    dpu_service.get_dpu_obj_by_path = function()
        return nil
    end
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT2, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    --call get_dpu_LLDP_status  fail
    dpu_service.get_dpu_obj_by_path = function()
        return {
            get_dpu_LLDP_status = function()
                return nil
            end
        }
    end
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT2, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    -- call get_dpu_LLDP_status suc, but value.type is illegal
    dpu_service.get_dpu_obj_by_path = function()
        return {
            get_dpu_LLDP_status = function()
                return true, {type = ILLEGAL_TYPE}
            end
        }
    end
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT2, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    -- call get_dpu_LLDP_status suc, and value.type is right
    dpu_service.get_dpu_obj_by_path = function()
        return {
            get_dpu_LLDP_status = function()
                return true, {type = LLDP_STATUS_TYPE, status = LLDP_ENABLE}
            end,
            set_dpu_LLDP_status = function()
                return false, LLDP_ENABLE
            end
        }
    end
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT2, LLDP_DISABLE)
    lu.assertEquals(ok, false)

    -- call get_dpu_LLDP_status suc, and value.type is right
    dpu_service.get_dpu_obj_by_path = function()
        return {
            get_dpu_LLDP_status = function()
                return true, {type = LLDP_STATUS_TYPE, status = LLDP_ENABLE}
            end,
            set_dpu_LLDP_status = function()
                return true, LLDP_ENABLE
            end
        }
    end
    local ok, _ = pcall(dpu_service.set_LLDP_status, dpu_service, obj, ctx, COMPUTING_BOARD_PORT1, LLDP_DISABLE)
    lu.assertEquals(ok, true)
end

function TestDpuService:test_set_serial_direction()
    local obj = {
        path = 'path'
    }
    local ctx
    local ok, _ = pcall(dpu_service.set_serial_direction, dpu_service, obj, ctx, 0)
    lu.assertEquals(ok, false)

    dpu_service.get_dpu_obj_by_path = function()
        return nil
    end
    ok, _ = pcall(dpu_service.set_serial_direction, dpu_service, obj, ctx, 1)
    lu.assertEquals(ok, false)
end

function TestDpuService:test_dpu_plug_out_record()
    local original_db = dpu_service.db
    dpu_service.db = {
        DPUCardMapping = {},
        delete = function()
            return {
                where = function()
                    return {
                        exec = function() end
                    }
                end
            }
        end
    }

    local obj_info = nil
    local position = "0101"
    dpu_service:dpu_plug_out_record(obj_info, position)

    obj_info = {
        slot_id = 0,
        vid = 1,
        did = 1,
        sub_did = 1,
        sub_vid = 1,
        sn = 'N/A',
        pn = 'PN111'
    }
    dpu_service:dpu_plug_out_record(obj_info, position)

    obj_info.slot_id = 1
    dpu_service:dpu_plug_out_record(obj_info, position)
    -- 还原
    dpu_service.db = original_db
end

function TestDpuService:test_dpu_plug_in_record()
    local original_db = dpu_service.db

    local obj = nil
    local position = "0101"
    dpu_service:dpu_plug_in_record(obj, position)

    obj = {
        [PCIECARD_INTF] = {
            SlotID = 1,
            VendorID = 1,
            DeviceID = 1,
            SubDeviceID = 1,
            SubVendorID = 1,
            SerialNumber = 'SN111',
            PartNumber = ''
        }
    }
    obj[PCIECARD_INTF].property_changed = {
        on = function(_, cb)
            cb('VendorID', 2)
        end
    }
    dpu_service.db = {
        DPUCardMapping = function ()
            return {
                save = function() end
            }
        end,
        select = function()
            return {
                where = function()
                    return {
                        first = function()
                            return true
                        end
                    }
                end
            }
        end
    }
    dpu_service:dpu_plug_in_record(obj, position)

    obj[PCIECARD_INTF].property_changed = {
        on = function(_, cb)
            cb('SlotID', 1)
        end
    }
    dpu_service.db.select = function()
        return {
            where = function()
                return {
                    first = function() end
                }
            end
        }
    end
    dpu_service:dpu_plug_in_record(obj, position)

    -- 还原
    dpu_service.db = original_db
end

function TestDpuService:test_get_uart_presence()
    local smc_obj

    dpu_service.dpu_detection_list = {
        [1] = {
            UartPresence = 0
        }
    }
    local ret = dpu_service:get_uart_presence(smc_obj)
    lu.assertEquals(ret, 0)

    smc_obj = {
        Read = function ()
            return {
                byte = function ()
                    return 1
                end
            }
        end
    }
    dpu_service.dpu_detection_list = {}
    ret = dpu_service:get_uart_presence(smc_obj)
    lu.assertEquals(ret, 1)

    smc_obj = {
        Read = function ()
            return {
                byte = function ()
                    error('get uart presence failed')
                end
            }
        end
    }
    dpu_service.dpu_detection_list = {}
    ret = dpu_service:get_uart_presence(smc_obj)
    lu.assertEquals(ret, nil)
end

function TestDpuService:test_monitor_serial_connection()
    local original_get_object = mdb.get_object

    local dpu_obj = {
        inner = {
            IgnoreSerialCableAlarm = true
        },
        general_serial_disconnect_event = function() end
    }
    dpu_service:monitor_serial_connection(dpu_obj)

    dpu_obj.inner.IgnoreSerialCableAlarm = false
    dpu_service.monitor_serial_task = true
    dpu_service:monitor_serial_connection(dpu_obj)

    dpu_obj.inner.IgnoreSerialCableAlarm = false
    dpu_service.monitor_serial_task = false
    dpu_service.dpu_detection_list = {}
    mdb.get_object = function ()
        return nil
    end
    dpu_service:monitor_serial_connection(dpu_obj)
    lu.assertEquals(dpu_service.monitor_serial_task, false)

    -- 还原
    mdb.get_object = original_get_object
end

function TestDpuService:test_dpu_set_linkage_powerstate()
    local dpu_obj = {
        inner = {
            IndependentPowerControlEnabled = true
        },
        pciecard = {
            ['SlotID'] = 5
        },
        port_id = 16,
        socket_id = 0,
        hotplug = {
            HotplugNotify = 0
        },
        _set_dpu_power_state = function()
            return true
        end,
        dpu_set_linkage_powerstate = function()
            return true
        end
    }
    local tmp_dpu_objects_1 = dpu_obj
    dpu_service.dpu_objects = {
        [1] = dpu_obj
    }
    local ok = pcall(function()
        dpu_service:dpu_set_linkage_powerstate_cb(0)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(dpu_obj.hotplug.HotplugNotify, 0)

    dpu_service.dpu_objects[1] = tmp_dpu_objects_1
end

function TestDpuService:test_set_power_state()
    local ok, _ = pcall(dpu_service.set_power_state, nil, nil, 2)
    lu.assertEquals(ok, false)
end

function TestDpuService:test_on_add_object()
    local obj = {
        HotplugNotify = 0
    }
    local dpu_obj = {
        set_hotplug = function(_)
            return true
        end
    }
    local tmp_dpu_objs = dpu_service.dpu_objects
    dpu_service.dpu_objects = {dpu_obj}
    local ok = pcall(dpu_service.on_add_object, dpu_service, 'DPUCardHotplug', obj, 'position')
    lu.assertEquals(ok, true)
    dpu_service.dpu_objects = tmp_dpu_objs
end

function TestDpuService:test_on_add_object_sr_upgrade_with_dpu_object()
    local position = '0101'
    local mock_sr_upgrade = { Type = 'BF3' }
    local set_dpu_type_called = false
    local set_dpu_type_arg = nil

    dpu_service.sr_upgrade_list = {}
    dpu_service.dpu_objects = {
        [position] = {
            set_dpu_type = function(_, dpu_type)
                set_dpu_type_called = true
                set_dpu_type_arg = dpu_type
            end
        }
    }

    dpu_service:on_add_object('SRUpgrade', mock_sr_upgrade, position)

    lu.assertEquals(dpu_service.sr_upgrade_list[position], mock_sr_upgrade)
    lu.assertTrue(set_dpu_type_called)
    lu.assertEquals(set_dpu_type_arg, 'BF3')
end

function TestDpuService:test_on_add_object_sr_upgrade_without_dpu_object()
    local position = '0102'
    local mock_sr_upgrade = { Type = 'BF2' }

    dpu_service.sr_upgrade_list = {}
    dpu_service.dpu_objects = {}

    dpu_service:on_add_object('SRUpgrade', mock_sr_upgrade, position)

    lu.assertEquals(dpu_service.sr_upgrade_list[position], mock_sr_upgrade)
end

function TestDpuService:test_on_delete_object_sr_upgrade_with_dpu_object()
    local position = '0101'
    local set_dpu_type_called = false
    local set_dpu_type_arg = nil

    dpu_service.sr_upgrade_list = { [position] = { Type = 'BF3' } }
    dpu_service.dpu_objects = {
        [position] = {
            set_dpu_type = function(_, dpu_type)
                set_dpu_type_called = true
                set_dpu_type_arg = dpu_type
            end
        }
    }

    dpu_service:on_delete_object('SRUpgrade', {}, position)

    lu.assertNil(dpu_service.sr_upgrade_list[position])
    lu.assertTrue(set_dpu_type_called)
    lu.assertEquals(set_dpu_type_arg, '')
end

function TestDpuService:test_on_delete_object_sr_upgrade_without_dpu_object()
    local position = '0102'

    dpu_service.sr_upgrade_list = { [position] = { Type = 'BF2' } }
    dpu_service.dpu_objects = {}

    dpu_service:on_delete_object('SRUpgrade', {}, position)

    lu.assertNil(dpu_service.sr_upgrade_list[position])
end