-- 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 json = require 'cjson'
local c_handler_optical = require 'bma.handles.handler_optical'
local c_handler_optical_diag = require 'bma.handles.handler_optical_diag'
local om_mgmt = require 'device.class.nic_mgmt.om_mgmt'
local log = require 'mc.logging'
-- 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_add_and_update_diag_bias_current()
--     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 = {
--         TxBiasHighAlarmMilliampere = 25.0,
--         TxBiasLowAlarmMilliampere = 1.0
--     }
--     self.bma:update(self.sff_dia_url, new_data)
--     lu.assertEquals(op.BC_TXUpperThresholdCritical, new_data.TxBiasHighAlarmMilliampere)
--     lu.assertEquals(op.BC_TXLowerThresholdCritical, new_data.TxBiasLowAlarmMilliampere)
-- 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)
--     local origin_func = op.update_fault_state
--     op.update_fault_state = function(self, _, fault_state)
--         self.FaultState = fault_state
--     end
--     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, '')
--     op.update_fault_state = origin_func
-- end

-- 覆盖 handler_optical：update_manufacturer_from_bma / update_part_number_from_bma /
-- update_serial_number_from_bma / update_production_date_from_bma /
-- update_wave_length_nanometer_from_bma（经 update_optical 调用）

local function stub_log_quiet()
    local si, se = log.info, log.error
    log.info = function() end
    log.error = function() end
    return function()
        log.info = si
        log.error = se
    end
end

local function make_mock_optical_op()
    return {
        Manufacturer = 'KEEP_MFR',
        PartNumber = 'KEEP_PN',
        SerialNumber = 'KEEP_SN',
        ProductionDate = 'KEEP_PD',
        WaveLengthNanometer = 999,
        TransceiverType = 'tt',
        FiberConnectionType = 'fc',
        SupportedSpeedsMbps = {},
        FaultState = 0,
        PowerState = 1,
        Type = 'typ',
        TypeMatch = true,
        SpeedMatch = true,
        ChannelNum = 0,
        Presence = 0,
        check_port_workload_type = function()
            return false
        end,
        update_fault_state = function() end,
        update_asset_data_info = function() end,
    }
end

--- update_optical 内其它 pcall 也会调 om_mgmt.get_instance，需一并提供空实现
local function new_om_mgmt_recorder()
    local rec = {
        manufacturer = {},
        part_number = {},
        serial_number = {},
        production_date = {},
        wave_length = {},
    }
    local mgr = {
        update_manufacturer_to_device_obj = function(_, op, v)
            rec.manufacturer[#rec.manufacturer + 1] = { op = op, val = v }
        end,
        update_part_number_to_device_obj = function(_, op, v)
            rec.part_number[#rec.part_number + 1] = { op = op, val = v }
        end,
        update_serial_number_to_device_obj = function(_, op, v)
            rec.serial_number[#rec.serial_number + 1] = { op = op, val = v }
        end,
        update_production_date_to_device_obj = function(_, op, v)
            rec.production_date[#rec.production_date + 1] = { op = op, val = v }
        end,
        update_wave_length_nanometer_to_device_obj = function(_, op, v)
            rec.wave_length[#rec.wave_length + 1] = { op = op, val = v }
        end,
        set_fault_state_to_device_obj = function() end,
        set_power_state_to_device_obj = function() end,
        set_type_match_to_device_obj = function() end,
        set_speed_match_to_device_obj = function() end,
        set_presence_to_device_obj = function() end,
    }
    return mgr, rec
end

local function run_update_optical(op, data)
    local orig_gi = om_mgmt.get_instance
    local restore_log = stub_log_quiet()
    local mgr, rec = new_om_mgmt_recorder()
    om_mgmt.get_instance = function()
        return mgr
    end
    c_handler_optical:update_optical(op, data)
    om_mgmt.get_instance = orig_gi
    restore_log()
    return rec
end

function test_bma_optical:test_update_optical_asset_nil_fields_skip_local_updates()
    local op = make_mock_optical_op()
    local rec = run_update_optical(op, {})
    lu.assertEquals(op.Manufacturer, 'KEEP_MFR')
    lu.assertEquals(op.PartNumber, 'KEEP_PN')
    lu.assertEquals(op.SerialNumber, 'KEEP_SN')
    lu.assertEquals(op.ProductionDate, 'KEEP_PD')
    lu.assertEquals(op.WaveLengthNanometer, 999)
    lu.assertEquals(#rec.manufacturer, 0)
    lu.assertEquals(#rec.part_number, 0)
    lu.assertEquals(#rec.serial_number, 0)
    lu.assertEquals(#rec.production_date, 0)
    lu.assertEquals(#rec.wave_length, 0)
end

function test_bma_optical:test_update_manufacturer_from_bma_and_sync()
    local op = make_mock_optical_op()
    op.Manufacturer = ''
    local rec = run_update_optical(op, { VendorName = 'ACME_OPT' })
    lu.assertEquals(op.Manufacturer, 'ACME_OPT')
    lu.assertEquals(#rec.manufacturer, 1)
    lu.assertIs(rec.manufacturer[1].op, op)
    lu.assertEquals(rec.manufacturer[1].val, 'ACME_OPT')
end

function test_bma_optical:test_update_manufacturer_json_null_uses_default()
    local op = make_mock_optical_op()
    op.Manufacturer = 'old'
    local rec = run_update_optical(op, { VendorName = json.null })
    lu.assertEquals(op.Manufacturer, '')
    lu.assertEquals(rec.manufacturer[1].val, '')
end

function test_bma_optical:test_update_part_serial_wave_from_bma()
    local op = make_mock_optical_op()
    op.PartNumber = ''
    op.SerialNumber = ''
    op.WaveLengthNanometer = 0
    local data = {
        PartNumber = 'PN-001',
        SerialNumber = 'SN-xyz',
        Wavelength = 1310,
    }
    local rec = run_update_optical(op, data)
    lu.assertEquals(op.PartNumber, 'PN-001')
    lu.assertEquals(op.SerialNumber, 'SN-xyz')
    lu.assertEquals(op.WaveLengthNanometer, 1310)
    lu.assertEquals(#rec.part_number, 1)
    lu.assertEquals(#rec.serial_number, 1)
    lu.assertEquals(#rec.wave_length, 1)
    lu.assertEquals(rec.part_number[1].val, 'PN-001')
    lu.assertEquals(rec.serial_number[1].val, 'SN-xyz')
    lu.assertEquals(rec.wave_length[1].val, 1310)
end

function test_bma_optical:test_update_production_date_from_bma_parse_and_null()
    local op1 = make_mock_optical_op()
    op1.ProductionDate = ''
    local rec1 = run_update_optical(op1, { ManufactureDate = '230515' })
    lu.assertEquals(op1.ProductionDate, '2023-05-15')
    lu.assertEquals(#rec1.production_date, 1)
    lu.assertEquals(rec1.production_date[1].val, '2023-05-15')

    local op2 = make_mock_optical_op()
    local rec2 = run_update_optical(op2, { ManufactureDate = json.null })
    lu.assertEquals(op2.ProductionDate, '')
    lu.assertEquals(rec2.production_date[1].val, '')
end

function test_bma_optical:test_update_optical_om_sync_pcall_swallows_error()
    local op = make_mock_optical_op()
    op.Manufacturer = ''
    local orig_gi = om_mgmt.get_instance
    local restore_log = stub_log_quiet()
    om_mgmt.get_instance = function()
        return {
            update_manufacturer_to_device_obj = function()
                error('dbus')
            end,
            update_part_number_to_device_obj = function() end,
            update_serial_number_to_device_obj = function() end,
            update_production_date_to_device_obj = function() end,
            update_wave_length_nanometer_to_device_obj = function() end,
            set_fault_state_to_device_obj = function() end,
            set_power_state_to_device_obj = function() end,
            set_type_match_to_device_obj = function() end,
            set_speed_match_to_device_obj = function() end,
            set_presence_to_device_obj = function() end,
        }
    end
    local ok = pcall(function()
        c_handler_optical:update_optical(op, { VendorName = 'OK_MFR' })
    end)
    lu.assertTrue(ok)
    lu.assertEquals(op.Manufacturer, 'OK_MFR')
    om_mgmt.get_instance = orig_gi
    restore_log()
end

-- 覆盖 handler_optical_diag：update_temperature_* / update_tx_bias_* / update_*power* / update_rx_*（94-221，经 update_optical_diag）

local function make_mock_diag_op()
    return {
        get_diagnostic_by_ibma_success = false,
        check_port_workload_type = function()
            return false
        end,
        ReadingCelsius = 25.0,
        TemperatureUpperThresholdCritical = 111,
        TemperatureLowerThresholdCritical = 112,
        SupplyVoltage = 3.3,
        VoltageUpperThresholdCritical = 113,
        VoltageLowerThresholdCritical = 114,
        TXBiasCurrentMilliAmps = { 1 },
        BC_TXUpperThresholdCritical = 115,
        BC_TXLowerThresholdCritical = 116,
        TXOutputPowerMilliWatts = { 2 },
        Power_TXUpperThresholdCritical = 117,
        Power_TXLowerThresholdCritical = 118,
        RXInputPowerMilliWatts = { 3 },
        RXUpperThresholdCritical = 119,
        RXLowerThresholdCritical = 120,
        Presence = 0,
    }
end

local function diag_om_method_names()
    return {
        'set_temperature_to_device_obj',
        'set_temperature_upper_threshold_critical_to_device_obj',
        'set_temperature_lower_threshold_critical_to_device_obj',
        'set_supply_voltage_to_device_obj',
        'set_voltage_upper_threshold_critical_to_device_obj',
        'set_voltage_lower_threshold_critical_to_device_obj',
        'set_tx_bias_current_milli_amps_to_device_obj',
        'set_bias_current_upper_threshold_critical_to_device_obj',
        'set_bias_current_lower_threshold_critical_to_device_obj',
        'set_tx_output_power_milli_watts_to_device_obj',
        'set_tx_power_upper_threshold_critical_to_device_obj',
        'set_tx_power_lower_threshold_critical_to_device_obj',
        'set_rx_input_power_milli_watts_to_device_obj',
        'set_rx_power_upper_threshold_critical_to_device_obj',
        'set_rx_power_lower_threshold_critical_to_device_obj',
        'set_presence_to_device_obj',
    }
end

local function new_diag_om_recorder()
    local rec = {}
    local mgr = {}
    for _, name in ipairs(diag_om_method_names()) do
        mgr[name] = function(_, op, val)
            rec[name] = rec[name] or {}
            rec[name][#rec[name] + 1] = { op = op, val = val }
        end
    end
    return mgr, rec
end

local function count_calls(rec, name)
    local t = rec[name]
    return t and #t or 0
end

local function run_update_optical_diag(op, data)
    local orig_gi = om_mgmt.get_instance
    local restore_log = stub_log_quiet()
    local mgr, rec = new_diag_om_recorder()
    om_mgmt.get_instance = function()
        return mgr
    end
    c_handler_optical_diag:update_optical_diag(op, data)
    om_mgmt.get_instance = orig_gi
    restore_log()
    return rec
end

--- 无诊断阈值/偏置/功率字段时，94-221 各函数提前 return，对应 set_* 不应被调用
function test_bma_optical:test_update_optical_diag_local_from_bma_skip_when_keys_absent()
    local op = make_mock_diag_op()
    local rec = run_update_optical_diag(op, {})
    lu.assertTrue(op.get_diagnostic_by_ibma_success)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, 111)
    lu.assertEquals(op.TemperatureLowerThresholdCritical, 112)
    lu.assertEquals(op.BC_TXUpperThresholdCritical, 115)
    lu.assertEquals(op.RXLowerThresholdCritical, 120)
    lu.assertEquals(count_calls(rec, 'set_temperature_upper_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_temperature_lower_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_tx_bias_current_milli_amps_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_bias_current_upper_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_bias_current_lower_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_tx_output_power_milli_watts_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_tx_power_upper_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_tx_power_lower_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_rx_input_power_milli_watts_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_rx_power_upper_threshold_critical_to_device_obj'), 0)
    lu.assertEquals(count_calls(rec, 'set_rx_power_lower_threshold_critical_to_device_obj'), 0)
end

function test_bma_optical:test_update_optical_diag_all_threshold_and_power_from_bma()
    local op = make_mock_diag_op()
    local data = {
        TempHighAlarmCelsius = 70,
        TempLowAlarmCelsius = -10,
        TxBiasCurrentMilliampere = 6.6,
        TxBiasHighAlarmMilliampere = 12.0,
        TxBiasLowAlarmMilliampere = 0.5,
        TxPowerCurrentMilliwatt = { 1.1, 2.2 },
        TxPowerHighAlarmMilliwatt = 3.3,
        TxPowerLowAlarmMilliwatt = 0.2,
        RxPowerCurrentMilliwatt = { 4.4 },
        RxPowerHighAlarmMilliwatt = 5.5,
        RxPowerLowAlarmMilliwatt = 0.05,
    }
    local rec = run_update_optical_diag(op, data)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, 70)
    lu.assertEquals(op.TemperatureLowerThresholdCritical, -10)
    lu.assertEquals(op.TXBiasCurrentMilliAmps, 6.6)
    lu.assertEquals(op.BC_TXUpperThresholdCritical, 12.0)
    lu.assertEquals(op.BC_TXLowerThresholdCritical, 0.5)
    lu.assertEquals(#op.TXOutputPowerMilliWatts, 2)
    lu.assertEquals(op.TXOutputPowerMilliWatts[1], 1.1)
    lu.assertEquals(op.TXOutputPowerMilliWatts[2], 2.2)
    lu.assertEquals(op.Power_TXUpperThresholdCritical, 3.3)
    lu.assertEquals(op.Power_TXLowerThresholdCritical, 0.2)
    lu.assertEquals(op.RXInputPowerMilliWatts[1], 4.4)
    lu.assertEquals(op.RXUpperThresholdCritical, 5.5)
    lu.assertEquals(op.RXLowerThresholdCritical, 0.05)
    lu.assertEquals(count_calls(rec, 'set_temperature_upper_threshold_critical_to_device_obj'), 1)
    lu.assertIs(rec.set_temperature_upper_threshold_critical_to_device_obj[1].op, op)
    lu.assertEquals(rec.set_temperature_upper_threshold_critical_to_device_obj[1].val, 70)
    lu.assertEquals(count_calls(rec, 'set_temperature_lower_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_tx_bias_current_milli_amps_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_bias_current_upper_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_bias_current_lower_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_tx_output_power_milli_watts_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_tx_power_upper_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_tx_power_lower_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_rx_input_power_milli_watts_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_rx_power_upper_threshold_critical_to_device_obj'), 1)
    lu.assertEquals(count_calls(rec, 'set_rx_power_lower_threshold_critical_to_device_obj'), 1)
end

function test_bma_optical:test_update_optical_diag_json_null_uses_default_65535()
    local op = make_mock_diag_op()
    local data = {
        TempHighAlarmCelsius = json.null,
        TxBiasHighAlarmMilliampere = json.null,
        RxPowerHighAlarmMilliwatt = json.null,
    }
    local rec = run_update_optical_diag(op, data)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, 65535)
    lu.assertEquals(op.BC_TXUpperThresholdCritical, 65535)
    lu.assertEquals(op.RXUpperThresholdCritical, 65535)
    lu.assertEquals(rec.set_temperature_upper_threshold_critical_to_device_obj[1].val, 65535)
    lu.assertEquals(rec.set_bias_current_upper_threshold_critical_to_device_obj[1].val, 65535)
    lu.assertEquals(rec.set_rx_power_upper_threshold_critical_to_device_obj[1].val, 65535)
end

function test_bma_optical:test_update_optical_diag_pcall_swallows_om_error()
    local op = make_mock_diag_op()
    local orig_gi = om_mgmt.get_instance
    local restore_log = stub_log_quiet()
    local base = {}
    for _, name in ipairs(diag_om_method_names()) do
        base[name] = function() end
    end
    base.set_temperature_upper_threshold_critical_to_device_obj = function()
        error('dbus')
    end
    om_mgmt.get_instance = function()
        return base
    end
    local ok = pcall(function()
        c_handler_optical_diag:update_optical_diag(op, { TempHighAlarmCelsius = 88 })
    end)
    lu.assertTrue(ok)
    lu.assertEquals(op.TemperatureUpperThresholdCritical, 88)
    om_mgmt.get_instance = orig_gi
    restore_log()
end

