-- Copyright (c) 2025 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- 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 om_mgmt = require 'device.class.nic_mgmt.om_mgmt'
local comm_fun = require 'device.class.nic_mgmt.comm_fun'
local comm_defs = require 'device.class.nic_mgmt.comm_defs'
local ctx = require 'mc.context'
local mdb = require 'mc.mdb'
local c_factory = require 'mc.orm.factory'
local c_optical_module = require 'device.class.optical_module'
local port_mgmt = require 'device.class.nic_mgmt.port_mgmt'
local om_init = require 'device.class.nic_mgmt.om.om_init'
local log = require 'mc.logging'

local origs = {}
local function save_orig(tbl, key)
    origs[tbl] = origs[tbl] or {}
    origs[tbl][key] = tbl[key]
end
local function stub(tbl, key, value)
    save_orig(tbl, key)
    tbl[key] = value
end
local function restore_all()
    for tbl, keys in pairs(origs) do
        for k, v in pairs(keys) do tbl[k] = v end
    end
end

TestOmMgmt = {}

local sync_cases = {
    { func = 'set_temperature_to_device_obj', interface_key = 'OPTICAL_MODULE_TEMPERATURE_CELSIUS_INTERFACE', field = 'TemperatureCelsius' },
    { func = 'set_presence_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'Presence' },
    { func = 'set_speed_match_to_device_obj', interface_key = 'OPTICAL_MODULE_STATUS_INTERFACE', field = 'SpeedMatch' },
    { func = 'set_type_match_to_device_obj', interface_key = 'OPTICAL_MODULE_STATUS_INTERFACE', field = 'TypeMatch' },
    { func = 'set_supply_voltage_to_device_obj', interface_key = 'OPTICAL_MODULE_VOLTAGE_INTERFACE', field = 'SupplyVoltage' },
    { func = 'set_voltage_lower_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_VOLTAGE_INTERFACE', field = 'VoltageLowerThresholdCritical' },
    { func = 'set_voltage_upper_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_VOLTAGE_INTERFACE', field = 'VoltageUpperThresholdCritical' },
    { func = 'set_bias_current_upper_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_CURRENT_INTERFACE', field = 'TXBiasCurrentUpperThresholdCritical' },
    { func = 'set_bias_current_lower_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_CURRENT_INTERFACE', field = 'TXBiasCurrentLowerThresholdCritical' },
    -- BMA → 设备树：厂商/料号/序列号/日期/波长（om_mgmt.lua 234-257）
    { func = 'update_manufacturer_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'Manufacturer' },
    { func = 'update_part_number_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'PartNumber' },
    { func = 'update_serial_number_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'SerialNumber' },
    { func = 'update_production_date_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'ProductionDate' },
    { func = 'update_wave_length_nanometer_to_device_obj', interface_key = 'OPTICAL_MODULE_DEVICE_INTERFACE', field = 'WaveLengthNanometer' },
    -- 温度/功率/电流阈值与诊断量（om_mgmt.lua 313-356）
    { func = 'set_temperature_upper_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_TEMPERATURE_CELSIUS_INTERFACE', field = 'TemperatureUpperThresholdCritical' },
    { func = 'set_temperature_lower_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_TEMPERATURE_CELSIUS_INTERFACE', field = 'TemperatureLowerThresholdCritical' },
    { func = 'set_tx_bias_current_milli_amps_to_device_obj', interface_key = 'OPTICAL_MODULE_CURRENT_INTERFACE', field = 'TXBiasCurrentMilliAmps' },
    { func = 'set_tx_output_power_milli_watts_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'TXOutputPowerMilliWatts' },
    { func = 'set_rx_input_power_milli_watts_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'RXInputPowerMilliWatts' },
    { func = 'set_rx_power_upper_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'RXPowerUpperThresholdCritical' },
    { func = 'set_rx_power_lower_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'RXPowerLowerThresholdCritical' },
    { func = 'set_tx_power_upper_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'TXPowerUpperThresholdCritical' },
    { func = 'set_tx_power_lower_threshold_critical_to_device_obj', interface_key = 'OPTICAL_MODULE_POWER_INTERFACE', field = 'TXPowerLowerThresholdCritical' },
}

local NIL = {}

local function make_optical_obj(overrides)
    local obj = {
        CreatedByDeviceObject = true,
        device_path = '/dev/optical',
        NetworkAdapterId = 'nid',
        PortID = '1'
    }
    if overrides then
        for k, v in pairs(overrides) do
            if v == NIL then
                obj[k] = nil
            else
                obj[k] = v
            end
        end
    end
    return obj
end

local function set_device_obj_stub(fn)
    rawset(comm_fun, 'get_device_obj_by_path_interface', fn)
end

local function reset_device_obj_stub()
    rawset(comm_fun, 'get_device_obj_by_path_interface', function()
        return {}
    end)
end

function TestOmMgmt:setUp()
    -- mock comm_defs
    stub(comm_defs, 'CARD_DEVICE_PATH_PATTERN', '/mock/pattern')
    stub(comm_defs, 'OPTICAL_MODULE_DEVICE_INTERFACE', 'iface_om')
    stub(comm_defs, 'OPTICAL_MODULE_CURRENT_INTERFACE', 'iface_cur')
    stub(comm_defs, 'OPTICAL_MODULE_POWER_INTERFACE', 'iface_pwr')
    stub(comm_defs, 'MACA_SERVICE', 'svc')
    stub(comm_defs, 'MDB_PATH', '/mdb')
    stub(comm_defs, 'MDB_INTERFACE', 'mdb_iface')
    -- mock ctx
    stub(ctx, 'get_context_or_default', function() return {} end)
    -- mock mdb
    stub(mdb, 'register_interface', function() end)
    stub(mdb, 'get_object', function()
        return {
            get_property = function(self, prop)
                return 0, 'val'
            end
        }
    end)
    -- mock c_optical_module
    stub(c_optical_module, 'create_mdb_object', function(tbl) return tbl end)
    stub(c_optical_module, 'collection', {
        find = function(self, q)
            return { children = {}, PortID = q and q.PortID or 123,
                NetworkAdapterId = q and q.NetworkAdapterId or 'nid',
                parent = { ObjectName = 'obj' }, ObjectName = 'portobj' }
        end,
        del_object = function() end
    })
    -- mock c_factory
    stub(c_factory, 'get_instance', function() return { create_object = function() end } end)
    -- mock port_mgmt.get_instance
    stub(port_mgmt, 'get_instance', function() return {
        get_orm_obj_by_device_path = function(_, path)
            if path == '/parent/path' then return { PortID = 123, NetworkAdapterId = 'nid',
                parent = { ObjectName = 'obj' }, ObjectName = 'portobj', FirmwareVersion = 'fw', children = {} } end
        end,
        register_children = function() end
    } end)
    -- mock comm_fun
    stub(comm_fun, 'get_parent_path', function(bus, optical_device_path)
        if optical_device_path == '/fail' then return nil end
        return '/parent/path'
    end)
    stub(comm_fun, 'get_object_name_by_device_path', function(bus, optical_device_path, rep)
        if optical_device_path == '/fail_obj' then return nil end
        return 'OpticalModule1'
    end)
    stub(comm_fun, 'get_object_identifier_by_device_path', function(bus, device_path)
        return {1, 'test_manager', 'test_chassis', 'test_position'}
    end)
    stub(comm_fun, 'get_all_device_paths', function() return { '/om1', '/om2' } end)
    stub(comm_fun, 'set_interface_add_signal', function(bus, sig_slot, path_pattern, interface_name, cb)
        sig_slot[#sig_slot + 1] = 123
        -- mock bus:match
        if not bus.match then
            bus.match = function(self, sig, cb)
                return 123
            end
        end
        cb('/mock/path')
    end)
    stub(comm_fun, 'get_device_obj_by_path_interface', function()
        return {}
    end)
    -- mock om_init
    stub(om_init, 'init', function() end)
    -- mock log
    stub(log, 'notice', function() end)
    stub(log, 'error', function() end)
    stub(log, 'info', function() end)
    reset_device_obj_stub()
end

function TestOmMgmt:tearDown()
    restore_all()
end

function TestOmMgmt:test_ctor_and_init()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    lu.assertIs(mgr.bus, bus)
    lu.assertIsTable(mgr.objects)
    lu.assertIsTable(mgr.sig_slot)
end

function TestOmMgmt:test_init_obj_normal()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init_obj('/om1')
    lu.assertIsTable(mgr.objects['/om1'])
end

function TestOmMgmt:test_init_obj_already_exists()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = { ['/om1'] = { id = 1 } }
    mgr.sig_slot = {}
    mgr:init_obj('/om1')
    lu.assertEquals(mgr.objects['/om1'].id, 1)
end

function TestOmMgmt:test_init_obj_no_parent_path()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init_obj('/fail')
    lu.assertNil(mgr.objects['/fail'])
end

function TestOmMgmt:test_init_obj_no_parent_orm()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    -- mock port_mgmt.get_instance返回无效对象
    port_mgmt.get_instance = function() return {
        get_orm_obj_by_device_path = function(_, path) return nil end,
        register_children = function() end
    } end
    mgr:init_obj('/om1')
    lu.assertNil(mgr.objects['/om1'])
end

function TestOmMgmt:test_init_obj_no_object_name()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init_obj('/fail_obj')
    lu.assertNil(mgr.objects['/fail_obj'])
end

function TestOmMgmt:test_init_obj_create_optical_orm_object_fail()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    -- mock c_optical_module.collection:find返回nil
    c_optical_module.collection = { find = function() return nil end }
    mgr:init_obj('/om1')
    lu.assertNil(mgr.objects['/om1'])
end

function TestOmMgmt:test_init()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init()
    lu.assertIsTable(mgr.objects['/om1'])
    lu.assertIsTable(mgr.objects['/om2'])
end

function TestOmMgmt:test_get_orm_obj_by_device_path()
    local bus = {}
    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = { ['/om1'] = { id = 1 } }
    local obj = mgr:get_orm_obj_by_device_path('/om1')
    lu.assertEquals(obj.id, 1)
    local obj2 = mgr:get_orm_obj_by_device_path('/notfound')
    lu.assertNil(obj2)
end 

function TestOmMgmt:test_sync_device_obj_skip_when_not_created_by_device()
    local mgr = om_mgmt
    mgr.bus = {}
    for _, case in ipairs(sync_cases) do
        local call_count = 0
        set_device_obj_stub(function(...)
            call_count = call_count + 1
            return {}
        end)
        mgr[case.func](mgr, make_optical_obj({ CreatedByDeviceObject = false }), 'val')
        lu.assertEquals(call_count, 0)
    end
    reset_device_obj_stub()
end

function TestOmMgmt:test_sync_device_obj_skip_when_device_path_missing()
    local mgr = om_mgmt
    mgr.bus = {}
    for _, case in ipairs(sync_cases) do
        local call_count = 0
        set_device_obj_stub(function(...)
            call_count = call_count + 1
            return {}
        end)
        mgr[case.func](mgr, make_optical_obj({ device_path = '' }), 'val')
        lu.assertEquals(call_count, 0)
    end

    reset_device_obj_stub()
end

function TestOmMgmt:test_sync_device_obj_skip_when_device_obj_missing()
    local mgr = om_mgmt
    mgr.bus = {}
    local call_count = 0
    set_device_obj_stub(function(...)
        call_count = call_count + 1
        return nil
    end)
    for _, case in ipairs(sync_cases) do
        mgr[case.func](mgr, make_optical_obj(), 'val')
    end
    lu.assertEquals(call_count, #sync_cases)
    reset_device_obj_stub()
end

function TestOmMgmt:test_sync_device_obj_success_sets_field()
    local mgr = om_mgmt
    mgr.bus = {}
    for _, case in ipairs(sync_cases) do
        local device_obj = {}
        local call_count = 0
        set_device_obj_stub(function(_, path, interface_name)
            call_count = call_count + 1
            lu.assertEquals(path, '/dev/optical')
            lu.assertEquals(interface_name, comm_defs[case.interface_key])
            return device_obj
        end)
        mgr[case.func](mgr, make_optical_obj(), 'new-value')
        lu.assertEquals(call_count, 1)
        lu.assertEquals(device_obj[case.field], 'new-value')
    end
    reset_device_obj_stub()
end

function TestOmMgmt:test_del_obj_remove_optical_module()
    local mgr = om_mgmt
    mgr.objects = { ['/mock/optical'] = { id = 1 } }

    local removed_obj
    local orig_del = c_optical_module.collection.del_object
    c_optical_module.collection.del_object = function(_, obj)
        removed_obj = obj
    end

    mgr:del_obj('/mock/optical')

    lu.assertEquals(removed_obj.id, 1)
    lu.assertNil(mgr.objects['/mock/optical'])

    c_optical_module.collection.del_object = orig_del
end

-- 覆盖 handle_special_property_mapping：经 init_obj → synchronize_property（om_mgmt.lua 85-106）
function TestOmMgmt:test_synchronize_special_property_mapping_basic()
    local iface_om = comm_defs.OPTICAL_MODULE_DEVICE_INTERFACE
    local iface_cur = comm_defs.OPTICAL_MODULE_CURRENT_INTERFACE
    local iface_pwr = comm_defs.OPTICAL_MODULE_POWER_INTERFACE
    local dbus_tx = { 0.11 }
    local dbus_rx = { 0.22 }
    local dbus_txpwr = { 0.33 }

    local orig_get_object = mdb.get_object
    mdb.get_object = function(bus, path, interface)
        return {
            get_property = function(_, prop)
                if interface == iface_om and prop == 'TransmissionDistance' then
                    return 0, 42
                end
                if interface == iface_cur and prop == 'TXBiasCurrentMilliAmps' then
                    return 0, dbus_tx
                end
                if interface == iface_pwr and prop == 'RXInputPowerMilliWatts' then
                    return 0, dbus_rx
                end
                if interface == iface_pwr and prop == 'TXOutputPowerMilliWatts' then
                    return 0, dbus_txpwr
                end
                return -1, nil
            end,
        }
    end

    local bus = {}
    bus.match = function()
        return 1
    end
    bus.call = function(_, svc, devpath, org_iface, method, ...)
        if method == 'GetObject' then
            return { svc = { iface_om, iface_cur, iface_pwr } }
        end
        if method == 'GetAll' then
            local _, interface = ...
            if interface == iface_om then
                return { TransmissionDistance = true }
            end
            if interface == iface_cur then
                return { TXBiasCurrentMilliAmps = true }
            end
            if interface == iface_pwr then
                return { RXInputPowerMilliWatts = true, TXOutputPowerMilliWatts = true }
            end
        end
        return {}
    end

    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init_obj('/om1')
    mdb.get_object = orig_get_object

    local orm = mgr.objects['/om1']
    lu.assertNotNil(orm)
    lu.assertEquals(orm.TransmissionDistance, '42')
    lu.assertEquals(orm.TXBiasCurrentMilliAmps, dbus_tx)
    lu.assertEquals(orm.RXInputPowerMilliWatts, dbus_rx)
    lu.assertEquals(orm.TXOutputPowerMilliWatts, dbus_txpwr)
end

-- 多通道且 iBMA 诊断已成功时，不以设备树覆盖 BMA 侧 TX/RX/TX 功率
function TestOmMgmt:test_synchronize_special_property_mapping_skips_multi_channel_when_ibma_ok()
    local iface_cur = comm_defs.OPTICAL_MODULE_CURRENT_INTERFACE
    local iface_pwr = comm_defs.OPTICAL_MODULE_POWER_INTERFACE

    local keep_tx = { 1, 2 }
    local keep_rx = { 3, 4 }
    local keep_txpwr = { 5, 6 }

    local coll = c_optical_module.collection
    local orig_find = coll.find
    coll.find = function(self, q)
        return {
            children = {},
            PortID = q and q.PortID or 123,
            NetworkAdapterId = q and q.NetworkAdapterId or 'nid',
            parent = { ObjectName = 'obj' },
            ObjectName = 'portobj',
            TXBiasCurrentMilliAmps = keep_tx,
            RXInputPowerMilliWatts = keep_rx,
            TXOutputPowerMilliWatts = keep_txpwr,
            get_diagnostic_by_ibma_success = true,
        }
    end

    local orig_get_object = mdb.get_object
    mdb.get_object = function(bus, path, interface)
        return {
            get_property = function(_, prop)
                if interface == iface_cur and prop == 'TXBiasCurrentMilliAmps' then
                    return 0, { 99 }
                end
                if interface == iface_pwr and prop == 'RXInputPowerMilliWatts' then
                    return 0, { 98 }
                end
                if interface == iface_pwr and prop == 'TXOutputPowerMilliWatts' then
                    return 0, { 97 }
                end
                return -1, nil
            end,
        }
    end

    local bus = {}
    bus.match = function()
        return 1
    end
    bus.call = function(_, svc, devpath, org_iface, method, ...)
        if method == 'GetObject' then
            return { svc = { iface_cur, iface_pwr } }
        end
        if method == 'GetAll' then
            local _, interface = ...
            if interface == iface_cur then
                return { TXBiasCurrentMilliAmps = true }
            end
            if interface == iface_pwr then
                return { RXInputPowerMilliWatts = true, TXOutputPowerMilliWatts = true }
            end
        end
        return {}
    end

    local mgr = om_mgmt
    mgr.bus = bus
    mgr.objects = {}
    mgr.sig_slot = {}
    mgr:init_obj('/om1')
    mdb.get_object = orig_get_object
    coll.find = orig_find

    local orm = mgr.objects['/om1']
    lu.assertNotNil(orm)
    lu.assertEquals(orm.TXBiasCurrentMilliAmps, keep_tx)
    lu.assertEquals(orm.RXInputPowerMilliWatts, keep_rx)
    lu.assertEquals(orm.TXOutputPowerMilliWatts, keep_txpwr)
end

return TestOmMgmt