-- 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.

require 'skynet.manager'
local skynet = require 'skynet'
local pcie_device_client = require 'pcie_device.client'
local sd_bus = require 'sd_bus'
local log = require 'mc.logging'
local test_common = require 'test_common.utils'
local utils = require 'mc.utils'
local lu = require 'luaunit'
local cmn = require 'common'
local mdb = require 'mc.mdb'
local ctx = require 'mc.context'
local tool = require 'ipmi_core.tool'
local pcie_addr_info = require 'test_pcie_addr_info'

-- RPC方法调用对象
local pcie_device

local function prepare_test_data()
    log:info('== prepare test data')
    local test_data_dir = skynet.getenv('TEST_DATA_DIR')
    os.execute('mkdir -p ' .. test_data_dir)
    os.execute('mkdir -p ' .. test_data_dir .. 'usr/lib64')
    utils.copy_file('temp/usr/lib64/mock/libsoc_adapter_it.so',
        test_data_dir .. 'usr/lib64/libsoc_adapter.so')

    local test_data_real_path = os.getenv('PROJECT_DIR') .. '/test/integration/.test_temp_data/'
    skynet.setenv('PROG_APP_PATH', test_data_real_path .. 'apps')
    skynet.setenv('PROG_CSR_PATH', test_data_real_path .. 'sr')
    skynet.setenv('DATA_CSR_PATH', test_data_real_path .. 'data/sr')
    -- 创建目录
    local dir_list = {'apps/pcie_device/mds', 'apps/hwdiscovery/mds', 'apps/ipmi_core/mds', 'sr'}
    for _, path in pairs(dir_list) do
        os.execute('mkdir -p ' .. test_data_dir .. '/' .. path)
    end
    utils.copy_file('test/integration/test_data/root.sr', test_data_dir .. '/sr/root.sr')
    utils.copy_file('test/integration/test_data/platform.sr', test_data_dir .. '/sr/platform.sr')
    utils.copy_file('test/integration/test_data/pcie_device.sr', test_data_dir .. '/sr/pcie_device.sr')
    utils.copy_file('mds/schema.json', test_data_dir .. '/apps/pcie_device/mds/schema.json')
    utils.copy_file('temp/opt/bmc/apps/hwdiscovery/mds/schema.json',
        test_data_dir .. '/apps/hwdiscovery/mds/schema.json')
    utils.copy_file('temp/opt/bmc/apps/ipmi_core/mds/schema.json',
        test_data_dir .. '/apps/ipmi_core/mds/schema.json')
end

local function clear_test_data(exit_test)
    log:info('== clear test data')
    local test_data_dir = skynet.getenv('TEST_DATA_DIR')
    if exit_test then
        skynet.timeout(0, function()
            skynet.sleep(20)
            skynet.abort()
            utils.remove_file(test_data_dir)
        end)
    else
        utils.remove_file(test_data_dir)
    end
end

local function bus_get_all(bus, uri, interface)
    return bus:call('bmc.kepler.pcie_device', uri, 'org.freedesktop.DBus.Properties', 'GetAll', 's', interface)
end

local function test_mdb_pcie_device(bus)
    require 'pcie_device.json_types.PCIeDevice'
    skynet.sleep(500)
    local objs = mdb.get_sub_objects(bus, '/bmc/kepler/Systems/1/PCIeDevices',
        'bmc.kepler.Systems.PCIeDevices.PCIeDevice')
    local obj = objs:fold(function (obj)
        if obj.path:match('PCIeDevice_1') then
            return obj, false
        end
    end)

    lu.assertEquals(obj.SlotID, 1)
    lu.assertEquals(obj.Segment, 1)
    lu.assertEquals(obj.SocketID, 1)
    lu.assertEquals(obj.Bus, 1)
    lu.assertEquals(obj.Device, 1)
    lu.assertEquals(obj.Function, 1)
    lu.assertEquals(obj.DevBus, 2)
    lu.assertEquals(obj.DevDevice, 2)
    lu.assertEquals(obj.DevFunction, 2)
    lu.assertEquals(obj.DiagnosticFault, 0)
    lu.assertEquals(obj.PredictiveFault, 0)

    local status = pcie_device:SetDiagnosticFault(ctx.new(), 1, 1, 1, 1, 1, 1)
    lu.assertEquals(status, 0)
    lu.assertEquals(obj.DiagnosticFault, 1)
    local status = pcie_device:SetPredictiveFault(ctx.new(), 1, 1, 1, 1, 1, 1)
    lu.assertEquals(status, 0)
    lu.assertEquals(obj.PredictiveFault, 1)
    local status, device_name = pcie_device:GetDeviceName(ctx.new(), 1, 1, 1, 1, 1)
    lu.assertEquals(status, 0)
    lu.assertEquals(device_name, 'PCIeCard1')
end

local function test_mdb_pcie_card(bus)
    require 'pcie_device.json_types.PCIeCard'
    local objs = mdb.get_sub_objects(bus, '/bmc/kepler/Systems/1/PCIeDevices/PCIeCards',
        'bmc.kepler.Systems.PCIeDevices.PCIeCard')
    local obj = objs:fold(function (obj)
        if obj.path:match('PCIeCard_1') then
            return obj, false
        end
    end)
    lu.assertEquals(obj.Name, 'Tesla T4')
    lu.assertEquals(obj.Description, 'Tesla T4')
    lu.assertEquals(obj.Position, 'PCIe Riser 1')
    lu.assertEquals(obj.Manufacturer, 'Huawei')
    lu.assertEquals(obj.LaneOwner, 1)
    lu.assertEquals(obj.FirmwareVersion, '1.1')
    lu.assertEquals(obj.PartNumber, 'AA0123456')
    lu.assertEquals(obj.SerialNumber, 'AA111')
    local num = pcie_device:GetPCIeCardNum(ctx.new())
    lu.assertEquals(num, 1)
end

local function test_pcie_addr_info(ipmi_tool)
    pcie_addr_info:test_set_info_by_ipmi(ipmi_tool)
end

local function test_pcie_device()
    log:info('================ test start ================')
    local bus = sd_bus.open_user(true)
    local ipmi_tool = tool.new(sd_bus.open_user(true))
    -- 初始化RPC方法调用对象
    require 'pcie_device.json_types.PCIeDevices'
    pcie_device = mdb.get_object(bus, '/bmc/kepler/Systems/1/PCIeDevices', 'bmc.kepler.Systems.PCIeDevices')

    --  the functions to be test
    test_mdb_pcie_device(bus)
    test_mdb_pcie_card(bus)
    log:info('================ test Set PcieAddr info ================')
    test_pcie_addr_info(ipmi_tool)

    skynet.call('pcie_device', 'lua', 'exit')
    log:info('================ test complete ================')
end

skynet.start(function()
    clear_test_data()
    prepare_test_data()
    test_common.dbus_launch()
    skynet.uniqueservice('sd_bus')
    skynet.uniqueservice('persistence/service/main')
    skynet.uniqueservice('maca/service/main')
    skynet.uniqueservice('hwdiscovery/service/main')
    skynet.uniqueservice('ipmi_core/service/main')
    skynet.sleep(500)
    skynet.uniqueservice('main')
    skynet.fork(function()
        local ok, err = pcall(test_pcie_device)
        clear_test_data(true)
        if not ok then
            error(err)
        end
    end)
end)
