-- 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 c_bma = require 'bma'
local test_data = require 'test_datas.bma_1822'
local c_network_adapter_db = require 'network_adapter.db'
local hook_tasks = require 'test_common.hook_tasks'
local c_object_manage = require 'mc.orm.object_manage'
local c_tasks = require 'mc.orm.tasks'
local test_utils = require 'test_utils'
local json = require 'cjson'

test_bma_optical = {}

function test_bma_optical:setUp()
    self.database = c_network_adapter_db(':memory:')

    hook_tasks.hook()
    self.object_manage = c_object_manage.new(self.database)
    self.object_manage.app = self

    self.bma = c_bma.new()

    local position = '00010102'
    self.network_adapter_obj = test_utils.add_network_adapter(self.object_manage,
        'NetworkAdapter_1', {
            BoardID = 0x123,
            DeviceLocator = 'some locator',
            Position = 'position',
            Type = 'Physical',
            Model = 'MyMode',
            Name = "a network card",
            Bus = 5,
            Device = 1,
            Function = 0,
            DevBus = 5,
            DevDevice = 1,
            DevFunction = 1
        }, position)
    self.network_port_obj = test_utils.add_network_port(self.object_manage, 'NetworkPort_1',
        {PortID = 0, NetDevFuncType = 1}, position, self.network_adapter_obj)
    self.optical_module_obj = test_utils.add_optical_module(self.object_manage, 'OpticalModule_1',
        {FaultState = 0}, position, self.network_port_obj)
    test_utils.add_object_complete(self.object_manage, position)

    self.object_manage.mc:prepare_ok()

    local eth_data = json.decode(test_data.EthernetInterface)
    local sff_data = json.decode(test_data.OemEthernetInterfaceSff)
    local sff_dia_data = json.decode(test_data.OemEthernetInterfaceSffDiagnostic)
    local eth_sta_data = json.decode(test_data.OemEthernetInterfaceStatistics)
    self.eth_url = eth_data['@odata.id']
    self.sff_url = sff_data['@odata.id']
    self.sff_dia_url = sff_dia_data['@odata.id']
    self.eth_sta_url = eth_sta_data['@odata.id']
end

function test_bma_optical:CreateVLANs()
end

function test_bma_optical:tearDown()
    c_bma.destroy()
    c_object_manage.destroy()
    self.database.db:close()
    hook_tasks.unhook()
end

function test_bma_optical:test_bma_update_eth()
    local na = self.network_adapter_obj
    local np = self.network_port_obj
    local data = json.decode(test_data.EthernetInterface)
    local huawei = data.Oem.Huawei

    self.bma:add(self.eth_url, data)

    lu.assertEquals(na.Model, 'MyMode') -- Model 以 sr 配置为准，除非 sr 中没有配置
    lu.assertEquals(na.Manufacturer, data.Manufacturer) -- Manufacturer 以 sr 配置为准，除非 sr 中没有配置
    lu.assertEquals(na.DisplayName, huawei.NICName)
    lu.assertEquals(na.FirmwareVersion, huawei.FirmwareVersion)
    lu.assertEquals(na.DriverName, huawei.DriverInfo.DriverName)
    lu.assertEquals(na.DriverVersion, huawei.DriverInfo.DriverVersion)
    lu.assertEquals(na.Description, data.Description)
    lu.assertEquals(na.RootBDF, huawei.BDFNumber.RootBDF)

    lu.assertEquals(np.Name, huawei.NICName)
    lu.assertEquals(np.OSLinkStatus, data.LinkStatus)
    lu.assertEquals(np.MACAddress, string.upper(data.MACAddress))
    lu.assertEquals(np.BDF, huawei.BDFNumber.BDF)
    lu.assertEquals(np.DriverName, huawei.DriverInfo.DriverName)
    lu.assertEquals(np.DriverVersion, huawei.DriverInfo.DriverVersion)
end

function test_bma_optical:test_bma_update_optical()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))

    local op = self.optical_module_obj
    local data = json.decode(test_data.OemEthernetInterfaceSff)

    lu.assertEquals(op.Manufacturer, data.VendorName)
    lu.assertEquals(op.PartNumber, data.PartNumber)
    lu.assertEquals(op.SerialNumber, data.SerialNumber)
    lu.assertEquals(op.ProductionDate,
                    string.format('20%s-%s-%s',
                                  string.sub(data.ManufactureDate, 1, 2),
                                  string.sub(data.ManufactureDate, 3, 4),
                                  string.sub(data.ManufactureDate, 5, 6)))
    lu.assertEquals(op.TransceiverType, data.TransceiverType)
    lu.assertEquals(op.WaveLengthNanometer, data.Wavelength)
    lu.assertEquals(op.FiberConnectionType, data.MediumMode)
    lu.assertEquals(op.SupportedSpeedsMbps, {25000})
    lu.assertEquals(op.Type, data.type)
end

function test_bma_optical:test_bma_add_and_update_optical()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))

    local op = self.optical_module_obj

    local new_data = {speed = {'3G'}, MediumMode = 'MediumMode', PartNumber = 'TR-PY85S-N11'}
    self.bma:update(self.sff_url, new_data)
    lu.assertEquals(op.FiberConnectionType, new_data.MediumMode)
    lu.assertEquals(op.SupportedSpeedsMbps, {3 * 1000})
    lu.assertEquals(op.PartNumber, new_data.PartNumber)
end

function test_bma_optical:test_bma_update_eth_statistics()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.eth_sta_url, json.decode(test_data.OemEthernetInterfaceStatistics))

    local port = self.network_port_obj
    local data = json.decode(test_data.OemEthernetInterfaceStatistics)

    lu.assertEquals(port.RXFrames, tostring(data.RxPackets))
    lu.assertEquals(port.TXFrames, tostring(data.TxPackets))
end

function test_bma_optical:test_bma_add_and_update_eth_statistics()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.eth_sta_url, json.decode(test_data.OemEthernetInterfaceStatistics))

    local port = self.network_port_obj

    local new_data = {RxPackets = 123}
    self.bma:update(self.eth_sta_url, new_data)
    lu.assertEquals(port.RXFrames, tostring(new_data.RxPackets))
end

function test_bma_optical:test_bma_update_diag()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))
    self.bma:add(self.sff_dia_url, json.decode(test_data.OemEthernetInterfaceSffDiagnostic))

    local op = self.optical_module_obj
    local data = json.decode(test_data.OemEthernetInterfaceSffDiagnostic)

    -- bmc.kepler.Systems.OpticalModule.Temperature
    lu.assertEquals(op.ReadingCelsius, data.TempCurrentCelsius)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, data.TempHighAlarmCelsius)
    lu.assertEquals(op.TemperatureLowerThresholdCritical, data.TempLowAlarmCelsius)

    -- bmc.kepler.Systems.OpticalModule.Voltage
    lu.assertEquals(op.SupplyVoltage, data.VccCurrentVolts)
    lu.assertEquals(op.VoltageUpperThresholdCritical, data.VccHighAlarmVolts)
    lu.assertEquals(op.VoltageLowerThresholdCritical, data.VccLowAlarmVolts)

    -- bmc.kepler.Systems.OpticalModule.BiasCurrent
    lu.assertEquals(op.TXBiasCurrentMilliAmps, data.TxBiasCurrentMilliampere)
    lu.assertEquals(op.BC_TXUpperThresholdCritical, data.TxBiasHighAlarmMilliampere)
    lu.assertEquals(op.BC_TXLowerThresholdCritical, data.TxBiasLowAlarmMilliampere)

    -- bmc.kepler.Systems.OpticalModule.Power
    lu.assertEquals(op.TXOutputPowerMilliWatts, data.TxPowerCurrentMilliwatt)
    lu.assertEquals(op.Power_TXUpperThresholdCritical, data.TxPowerHighAlarmMilliwatt)
    lu.assertEquals(op.Power_TXLowerThresholdCritical, data.TxPowerLowAlarmMilliwatt)
    lu.assertEquals(op.RXInputPowerMilliWatts, data.RxPowerCurrentMilliwatt)
    lu.assertEquals(op.RXUpperThresholdCritical, data.RxPowerHighAlarmMilliwatt)
    lu.assertEquals(op.RXLowerThresholdCritical, data.RxPowerLowAlarmMilliwatt)
end

function test_bma_optical:test_bma_add_and_update_diag()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))
    self.bma:add(self.sff_dia_url, json.decode(test_data.OemEthernetInterfaceSffDiagnostic))

    local op = self.optical_module_obj

    local new_data = {TempCurrentCelsius = 37.0}
    self.bma:update(self.sff_dia_url, new_data)
    lu.assertEquals(op.ReadingCelsius, new_data.TempCurrentCelsius)
end

function test_bma_optical:test_bma_reset_stats()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))
    self.bma:add(self.sff_dia_url, json.decode(test_data.OemEthernetInterfaceSffDiagnostic))

    local op = self.optical_module_obj
    local data = json.decode(test_data.OemEthernetInterfaceSffDiagnostic)

    -- bmc.kepler.Systems.OpticalModule.Temperature
    lu.assertEquals(op.ReadingCelsius, data.TempCurrentCelsius)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, data.TempHighAlarmCelsius)
    lu.assertEquals(op.TemperatureLowerThresholdCritical, data.TempLowAlarmCelsius)

    -- bmc.kepler.Systems.OpticalModule.Voltage
    lu.assertEquals(op.SupplyVoltage, data.VccCurrentVolts)
    lu.assertEquals(op.VoltageUpperThresholdCritical, data.VccHighAlarmVolts)
    lu.assertEquals(op.VoltageLowerThresholdCritical, data.VccLowAlarmVolts)

    -- bmc.kepler.Systems.OpticalModule.BiasCurrent
    lu.assertEquals(op.TXBiasCurrentMilliAmps, data.TxBiasCurrentMilliampere)
    lu.assertEquals(op.BC_TXUpperThresholdCritical, data.TxBiasHighAlarmMilliampere)
    lu.assertEquals(op.BC_TXLowerThresholdCritical, data.TxBiasLowAlarmMilliampere)

    -- bmc.kepler.Systems.OpticalModule.Power
    lu.assertEquals(op.TXOutputPowerMilliWatts, data.TxPowerCurrentMilliwatt)
    lu.assertEquals(op.Power_TXUpperThresholdCritical, data.TxPowerHighAlarmMilliwatt)
    lu.assertEquals(op.Power_TXLowerThresholdCritical, data.TxPowerLowAlarmMilliwatt)
    lu.assertEquals(op.RXInputPowerMilliWatts, data.RxPowerCurrentMilliwatt)
    lu.assertEquals(op.RXUpperThresholdCritical, data.RxPowerHighAlarmMilliwatt)
    lu.assertEquals(op.RXLowerThresholdCritical, data.RxPowerLowAlarmMilliwatt)

    self.bma:on_reset()
    -- bmc.kepler.Systems.OpticalModule.Temperature
    lu.assertEquals(op.ReadingCelsius, 0xFFFF)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, 0xFFFF)
    lu.assertEquals(op.TemperatureLowerThresholdCritical, 0xFFFF)

    -- bmc.kepler.Systems.OpticalModule.Voltage
    lu.assertEquals(op.SupplyVoltage, 0xFFFF)
    lu.assertEquals(op.VoltageUpperThresholdCritical, 0xFFFF)
    lu.assertEquals(op.VoltageLowerThresholdCritical, 0xFFFF)

    -- bmc.kepler.Systems.OpticalModule.BiasCurrent
    lu.assertEquals(op.TXBiasCurrentMilliAmps, {0xFFFF})
    lu.assertEquals(op.BC_TXUpperThresholdCritical, 0xFFFF)
    lu.assertEquals(op.BC_TXLowerThresholdCritical, 0xFFFF)

    -- bmc.kepler.Systems.OpticalModule.Power
    lu.assertEquals(op.TXOutputPowerMilliWatts, {0xFFFF})
    lu.assertEquals(op.Power_TXUpperThresholdCritical, 0xFFFF)
    lu.assertEquals(op.Power_TXLowerThresholdCritical, 0xFFFF)
    lu.assertEquals(op.RXInputPowerMilliWatts, {0xFFFF})
    lu.assertEquals(op.RXUpperThresholdCritical, 0xFFFF)
    lu.assertEquals(op.RXLowerThresholdCritical, 0xFFFF)
end

function test_bma_optical:test_bma_reset_optical()
    self.bma:add(self.eth_url, json.decode(test_data.EthernetInterface))
    self.bma:add(self.sff_url, json.decode(test_data.OemEthernetInterfaceSff))

    local op = self.optical_module_obj
    local data = json.decode(test_data.OemEthernetInterfaceSff)

    lu.assertEquals(op.Manufacturer, data.VendorName)
    lu.assertEquals(op.PartNumber, data.PartNumber)
    lu.assertEquals(op.SerialNumber, data.SerialNumber)
    lu.assertEquals(op.ProductionDate,
                    string.format('20%s-%s-%s',
                                  string.sub(data.ManufactureDate, 1, 2),
                                  string.sub(data.ManufactureDate, 3, 4),
                                  string.sub(data.ManufactureDate, 5, 6)))
    lu.assertEquals(op.TransceiverType, data.TransceiverType)
    lu.assertEquals(op.WaveLengthNanometer, data.Wavelength)
    lu.assertEquals(op.FiberConnectionType, data.MediumMode)
    lu.assertEquals(op.SupportedSpeedsMbps, {25000})
    lu.assertEquals(op.Type, data.type)
    self.bma:on_reset()
    lu.assertEquals(op.Manufacturer, '')
    lu.assertEquals(op.PartNumber, '')
    lu.assertEquals(op.SerialNumber, '')
    lu.assertEquals(op.ProductionDate, '')
    lu.assertEquals(op.TransceiverType, '')
    lu.assertEquals(op.WaveLengthNanometer, '')
    lu.assertEquals(op.FiberConnectionType, '')
    lu.assertEquals(op.SupportedSpeedsMbps, {})
    lu.assertEquals(op.Type, '')
end
