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

-- Description: power_mgmt test
require 'skynet.manager'
local skynet = require 'skynet'
local sd_bus = require 'sd_bus'
local log = require 'mc.logging'
local lu = require 'luaunit'
local mdb = require 'mc.mdb'
local cilent = require 'power_mgmt.client'
local test_common = require 'test_common.utils'
local utils = require 'mc.utils'
local datas = require 'power_mgmt.datas'
local ctx = require 'mc.context'
local test_manufacture = require 'test_manufacture'
local test_metric = require 'test_metric'
local utils_core = require 'utils.core'
local psu_def = require 'macros.psu_def'
local crc8 = require 'mc.crc8'
local test_power_ipmi = require 'test_power_ipmi'
local test_power_mgmt_rpc = require 'test_power_mgmt_rpc'
local base_msg = require 'messages.base'
local test_customize_config = require 'test_customize_config'
require 'power_mgmt.json_types.OnePower'
require 'power_mgmt.json_types.Status'
require 'power_mgmt.json_types.BlackBox'
require 'power_mgmt.json_types.PowerSupplies'
require 'power_mgmt.json_types.EfficiencyCurve'
require 'power_mgmt.json_types.Capacitor'
require 'power_mgmt.json_types.CapacitorMetrics'
require 'power_mgmt.json_types.OutputControl'

POWER_PATH = '/bmc/kepler/Systems/1/PowerMgmt/'
POWER_DEBUG_PATH = '/bmc/kepler/Debug/Chassis/1/PowerMgmt/'
POWER_INTF = 'bmc.kepler.Systems.PowerMgmt.OnePower'
POWER_INTF_METRICS = 'bmc.kepler.Systems.PowerMgmt.OnePower.Metrics'
POWER_INTF_TEMPERATURES = 'bmc.kepler.Systems.PowerMgmt.OnePower.Temperatures'
POWER_INTF_STATUS = 'bmc.kepler.Systems.PowerMgmt.OnePower.Status'
POWER_INTF_BLK_BOX = 'bmc.kepler.Debug.BlackBox'
POWER_INTF_EFFICIENCY_CURVE = 'bmc.kepler.Systems.PowerMgmt.OnePower.EfficiencyCurve'
POWER_INTF_CAPACITOR = 'bmc.kepler.Chassis.PowerSubsystem.Capacitor'
POWER_INTF_CAPACITOR_METRICS = 'bmc.kepler.Chassis.PowerSubsystem.Capacitor.Metrics'
POWER_INTF_OUTPUT_CONTROL = 'bmc.kepler.Chassis.PowerSubsystem.PowerSupply.OutputControl'


local project_dir = os.getenv('PROJECT_DIR')
local test_dir = project_dir .. '/test/integration/.test_temp_data/'
skynet.setenv('PROG_APP_PATH', test_dir .. 'apps')
skynet.setenv('PROG_CSR_PATH', test_dir .. 'sr')
skynet.setenv('DATA_CSR_PATH', test_dir .. 'data/sr')

local function prepare_test_data()
    log:notice('================ prepare test data start ================')
    local test_data_dir = skynet.getenv('TEST_DATA_DIR')
    os.execute('mkdir -p ' .. test_data_dir)
    local dir_list = { 'apps/power_mgmt/mds', 'apps/hwproxy/mds', 'apps/hwdiscovery/mds', 'apps/bmc_time/mds', 'sr',
        'data', 'usr/lib64', 'apps/chassis/mds', 'apps/ipmi_core/mds', 'apps/product_mgmt/mds' }
    local absolute_test_data_dir = os.getenv('PROJECT_DIR') .. '/test/integration/.test_temp_data/'
    skynet.setenv('KSF_PATH', absolute_test_data_dir)
    skynet.setenv('KSF_BAK_PATH', absolute_test_data_dir)
    skynet.setenv('KSF_DEFAULT_PATH', absolute_test_data_dir)
    utils.copy_file('test/integration/test_data/datatocheck_default.dat',
        test_data_dir .. '/datatocheck_default.dat')
    for _, path in pairs(dir_list) do
        os.execute('mkdir -p ' .. test_data_dir .. '/' .. path)
    end
    os.execute('cp -r ' .. 'test/integration/test_data/mockdata' .. ' ' .. test_data_dir .. 'data')
    -- 拷贝本地sr
    utils.copy_file('test/integration/test_data/14100513_00000001010302044492.sr',
        test_data_dir .. '/sr/14100513_00000001010302044492.sr')
    utils.copy_file('test/integration/test_data/14100513_00000001010302044491.sr',
        test_data_dir .. '/sr/14100513_00000001010302044491.sr')
    utils.copy_file('test/integration/test_data/14100513_00000001010302023922.sr',
        test_data_dir .. '/sr/14100513_00000001010302023922.sr')
    utils.copy_file('test/integration/test_data/14191046_PSU_0.sr',
        test_data_dir .. '/sr/14191046_PSU_0.sr')
    utils.copy_file('test/integration/test_data/14100513_Bmctime_01.sr',
        test_data_dir .. '/sr/14100513_Bmctime_01.sr')
    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')
    -- 拷贝模块自身的schema
    utils.copy_file('mds/schema.json', test_data_dir .. '/apps/power_mgmt/mds/schema.json')
    -- 拷贝依赖模块的schema
    utils.copy_file('temp/opt/bmc/apps/chassis/mds/schema.json',
        test_data_dir .. '/apps/chassis/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/hwproxy/mds/schema.json',
        test_data_dir .. '/apps/hwproxy/mds/schema.json')
        utils.copy_file('temp/opt/bmc/apps/product_mgmt/mds/schema.json',
        test_data_dir .. '/apps/product_mgmt/mds/schema.json')
    utils.copy_file('temp/opt/bmc/apps/bmc_time/mds/schema.json',
        test_data_dir .. '/apps/bmc_time/mds/schema.json')
    utils.copy_file('temp/opt/bmc/apps/ipmi_core/mds/schema.json',
        test_data_dir .. '/apps/ipmi_core/mds/schema.json')
    utils.copy_file('temp/opt/bmc/apps/event/mds/schema.json',
        test_data_dir .. '/apps/event/mds/schema.json')
    utils.copy_file('temp/usr/lib64/mock/libsoc_adapter_it.so',
        test_data_dir .. '/usr/lib64/libsoc_adapter.so')
end

local function clear_test_data(exit_test)
    log:notice('- clear test data')
    local test_data_dir = utils.realpath('.') .. '/' .. 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 test_status(bus, onepower_obj)
    log:notice('================ test power status ================')
    -- test DeepSleep
    local ok, status_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. 'OnePower_0_010109', POWER_INTF_STATUS
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(status_obj['SleepMode'], 'Normal')
    local ok, rsp = status_obj.pcall:SetSleepMode(ctx.new(), 'DeepSleep')
    lu.assertTrue(ok)
    lu.assertEquals(rsp, 0)
    skynet.sleep(500)
    lu.assertEquals(status_obj['SleepMode'], 'DeepSleep')
    -- 深度休眠下显示StandbySpare
    lu.assertEquals(onepower_obj['WorkMode'], 'StandbySpare')

    local ok, eeprom_obj = pcall(
        mdb.get_object, bus, '/bmc/kepler/Chip/Eeprom/Eeprom_PsuChip1_0101', 'bmc.kepler.Chip.BlockIO'
    )
    lu.assertTrue(ok)
    -- 模拟异常，待继续补充
    eeprom_obj.pcall:Write(ctx.new(), 0x81, {128, 51})
    eeprom_obj.pcall:Write(ctx.new(), 0x7c, {0x90, 190})
    skynet.sleep(16000)
    lu.assertEquals(status_obj['InputVoltageFault'], 144)
    lu.assertEquals(status_obj['Fan1Fault'], 1)
    lu.assertEquals(status_obj['Health'], 144)
    -- 恢复异常
    eeprom_obj.pcall:Write(ctx.new(), 0x81, {0x00, 186})
    eeprom_obj.pcall:Write(ctx.new(), 0x7c, {0x00, 71})
    skynet.sleep(16000)
    lu.assertEquals(status_obj['Fan1Fault'], 0)
    lu.assertEquals(status_obj['InputVoltageFault'], 0)
    lu.assertEquals(status_obj['Health'], 0)
    log:notice('================ test power status complete ================')
end

local function test_black_box(bus, object_name)
    local ok, onepower_obj = pcall(
        mdb.get_object, bus, POWER_DEBUG_PATH .. object_name, POWER_INTF_BLK_BOX
    )
    lu.assertTrue(ok, true)
    ok = onepower_obj.pcall:DumpPsuBlackbox_PACKED(ctx.new())
    lu.assertTrue(ok)
    skynet.sleep(200)
end

local function test_psu1(bus)
    local object_name = 'OnePower_0_010109'
    local ok, onepower_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(onepower_obj['Manufacturer'], 'HUAWEI')
    lu.assertEquals(onepower_obj['Model'], 'PAC900S12-B2')
    lu.assertEquals(onepower_obj['PartNumber'], '02312XWK')
    lu.assertEquals(onepower_obj['WorkMode'], 'Enabled')
    lu.assertEquals(onepower_obj['Protocol'], 'pmbus')
    lu.assertEquals(onepower_obj['SerialNumber'], '2102312XWK10N8112866')
    lu.assertEquals(onepower_obj['RelaySwitchedCount'], 0)
    -- 设置单个电源Slot:1, 备用工作模式：1
    local ok, rsp = onepower_obj.pcall:SetPowerWorkMode_PACKED(ctx.new(), 1)
    -- 设置成功，返回结果：0
    lu.assertTrue(ok)
    lu.assertEquals(rsp.CmdResult, 0)
    skynet.sleep(2000)
    lu.assertEquals(onepower_obj['WorkMode'], 'StandbySpare')

    -- 恢复主用
    ok, rsp = onepower_obj.pcall:SetPowerWorkMode_PACKED(ctx.new(), 0)
    -- 设置成功，返回结果：0
    lu.assertTrue(ok)
    lu.assertEquals(rsp.CmdResult, 0)
    skynet.sleep(2000)
    lu.assertEquals(onepower_obj['WorkMode'], 'Enabled')

    test_status(bus, onepower_obj)
    object_name = 'OnePowerDebug_0_010109'
    test_black_box(bus, object_name)
end

local function test_psu2(bus)
    local object_name = 'OnePower_0_010209'
    local ok, onepower_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(onepower_obj['Manufacturer'], 'HUAWEI')
    lu.assertEquals(onepower_obj['Model'], 'PAC900S12-B2')
    lu.assertEquals(onepower_obj['PartNumber'], '02312XWK')
    lu.assertEquals(onepower_obj['WorkMode'], 'Enabled')
    lu.assertEquals(onepower_obj['Protocol'], 'pmbus')
    lu.assertEquals(onepower_obj['SerialNumber'], '2102312XWK10N7123736')
end

local function test_efficiency_curve(bus)
    local object_name = 'OnePower_0_010109'
    local ok, efficiency_curve_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF_EFFICIENCY_CURVE
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(efficiency_curve_obj['PowerLevel'], '80PLUS Platinum')
end

local function test_capacitor(bus)
    local object_name = 'OnePower_0_010109'
    local ok, capacitor_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF_CAPACITOR
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(capacitor_obj['HardwareVersion'], '')
    -- 设置VOUTDebounceMilliseconds，当前pmbus不支持
    local method_ok, rsp = capacitor_obj.pcall:SetVOUTDebounceMilliseconds_PACKED(ctx.new(), 10)
    lu.assertEquals(method_ok, false)
    lu.assertEquals(rsp.name, base_msg.ActionNotSupportedMessage.Name)
    -- 设置DepthOfDischargeVolts，当前pmbus不支持
    method_ok, rsp = capacitor_obj.pcall:SetDepthOfDischargeVolts_PACKED(ctx.new(), 60)
    lu.assertEquals(method_ok, false)
    lu.assertEquals(rsp.name, base_msg.ActionNotSupportedMessage.Name)
    ok, capacitor_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF_CAPACITOR_METRICS
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(capacitor_obj['CapacityMicrofarads'], 0)
end

local function test_outpt_control(bus)
    local object_name = 'OnePower_0_010109'
        local ok, output_control_obj = pcall(
            mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF_OUTPUT_CONTROL
        )
        lu.assertEquals(ok, true)
        local method_ok, rsp = output_control_obj.pcall:SetOutputPowerLimitWatts_PACKED(ctx.new(), 1)
        lu.assertEquals(method_ok, false)
        lu.assertEquals(rsp.name, base_msg.ActionNotSupportedMessage.Name)
    end

local function test_turbo_alarm_status(bus)
    local object_name = 'OnePower_0_010109'
    local ok, status_obj = pcall(
        mdb.get_object, bus, POWER_PATH .. object_name, POWER_INTF_STATUS
    )
    lu.assertTrue(ok, true)
    lu.assertEquals(status_obj['TPSUStatusHigh'], 0)
    lu.assertEquals(status_obj['TPSUStatusLow'], 0)
end

local function test_power_info(bus)
    test_psu1(bus)
    test_psu2(bus)
    test_efficiency_curve(bus)
    test_capacitor(bus)
    test_turbo_alarm_status(bus)
    test_outpt_control(bus)
end

local function test_power_supplies(bus)
    local ok, power_supplies = pcall(
        mdb.get_object, bus, '/bmc/kepler/Chassis/1/PowerSubsystem/PowerSupplies',
        'bmc.kepler.Chassis.PowerSubsystem.PowerSupplies'
    )
    if not ok then
        log:error(power_supplies)
    end
    lu.assertTrue(ok, true)
    ok, power_supplies = pcall(function()
        return power_supplies:SetPsusFanMinPWM(ctx.new(), 50)
    end)
    if not ok then
        log:error(power_supplies)
    end
    lu.assertTrue(ok, true)
end

local function test_power_mgmt()
    log:notice('================ test start ================')
    local bus = sd_bus.open_user(true)
    test_power_info(bus)
    test_power_supplies(bus)
    test_manufacture.test_all_dft_method(bus)
    test_metric.test_all_metric_method(bus)
    test_power_ipmi.test_set_power_status(bus)
    test_power_ipmi.test_get_voltage(bus)
    test_power_ipmi.test_get_current(bus)
    test_power_ipmi.test_get_power(bus)
    test_power_ipmi.test_get_power_supply_info(bus)
    test_power_ipmi.test_set_psu_fru_config(bus)
    test_power_ipmi.test_power_sleep_mode_1(bus)
    test_power_ipmi.test_power_sleep_mode_2(bus)
    test_power_mgmt_rpc.main(bus)
    test_customize_config.main(bus)
    log:notice('================ test complete ================')
end

skynet.register_protocol {
    name = "test_mode",
    id = 99,
    pack = skynet.pack,
    unpack = skynet.unpack
}

skynet.start(function()
    clear_test_data()
    prepare_test_data()
    test_common.dbus_launch()
    skynet.uniqueservice('sd_bus')
    skynet.uniqueservice('persistence/service/main')
    skynet.uniqueservice('key_mgmt/service/main')
    skynet.uniqueservice('maca/service/main')
    skynet.uniqueservice('hwproxy/service/main')
    skynet.uniqueservice('product_mgmt/service/main')
    skynet.uniqueservice('ipmi_core/service/main')
    skynet.uniqueservice('bmc_time/service/main')
    skynet.uniqueservice('chassis/service/main')
    skynet.uniqueservice('main')
    skynet.uniqueservice('hwdiscovery/service/main')
    -- 线程起太快会一直失败
    skynet.sleep(4000)
    skynet.fork(function()
        local ok, err = pcall(test_power_mgmt)
        clear_test_data(true)
        if not ok then
            error(err)
        end
    end)
end)
