-- Copyright (c) 2026 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 ub_cable_detection = require 'ub_cable.ub_cable_detection'
local client = require 'pcie_device.client'
local event = require 'ub_cable.infrastructure.event'
local mdb_service = require 'mc.mdb.mdb_service'
local utils_core = require 'utils.core'
local mdb = require 'mc.mdb'

TestUbCableDetection = {}

function TestUbCableDetection:test_init()
    ub_cable_detection.listen_inter_node_update = function()
        return
    end
    local ok, _ = pcall(ub_cable_detection.init, ub_cable_detection)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_update_guid_map()
    local port_obj = {}
    local port_props = {
        NodeId = "node-001",
        CPUId = "cpu-2",
        DieId = "die-1",
        MacroId = "macro-1",
        InterfaceDeviceSlotId = "slot-2",
        InterfaceDevicePortId = "port-2"
    }
    port_obj = {
        prop = port_props,

        src_guid = "guid-11111",

        get_prop = function(_, key)
            return port_obj.prop[key]
        end
    }
    ub_cable_detection.guid_map = {
        [1] = {
            node_id = nil,
            cpu_id = nil,
            die_id = nil,
            slot_id = nil,
            port_id = nil,
            guid = nil
        }
    }
    local ok, _ = pcall(ub_cable_detection.update_guid_map, ub_cable_detection, port_obj)
    lu.assertEquals(ok, true)

    port_obj.prop['NodeId'] = 255
    ok, _ = pcall(ub_cable_detection.update_guid_map, ub_cable_detection, port_obj)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_update_guid_node_id_map()
    local guid = 'abc'
    local node_id = '222222222222'
    ub_cable_detection.node_id_guid_map = {}
    ub_cable_detection.guid_node_id_map = {
        [1] = {
            abc = nil
        }
    }
    local ok, ret = pcall(ub_cable_detection.update_guid_node_id_map, ub_cable_detection, guid, node_id)
    if not ok then
        print(ret)
    end
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_get_props_rpc()
    local mdb_bak = mdb.get_object

    mdb.get_object = function()
        return {
            GetProperties = function()
                return {
                    {
                        [1] = 63,
                        [2] = {
                            [1] = {
                                [1] = 'Name',
                                [2] = {
                                    ["PropertyValue"] = "NodePortAddr_0_0_6_1_3_2_8000B417B30000000000000200A008CC",
                                    ["Expired"] = "false"
                                }
                            }
                        }
                    }
                }
            end
        }
    end
    ub_cable_detection.node_port_addr_list = {
        [1] = {
            update_target_info = function()
            end
        }
    }
    local ok, ret = pcall(ub_cable_detection.get_props_rpc, ub_cable_detection, 1, 'object_name', 'interface')
    if not ok then
        print(ret)
    end
    lu.assertEquals(ok, true)

    mdb.get_object = mdb_bak
end

function TestUbCableDetection:test_listen_inter_node_update()
    client.SubscribeDataSyncPropertyUpdated = function()
        return
    end
    local ok, _ = pcall(ub_cable_detection.listen_inter_node_update, ub_cable_detection, client)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_update_peer_info()
    local port_obj = {}
    local peer_info = {
        remote_link_status_list = {
            [1] = {
                remote_port_guid = '1111',
                remote_port_id = '22222'
            }
        }
    }
    local port_props = {
        NodeId = "node-001",
        CPUId = "cpu-2",
        DieId = "die-1",
        MacroId = "macro-1",
        InterfaceDeviceSlotId = "slot-2",
        InterfaceDevicePortId = "port-2"
    }
    ub_cable_detection.guid_node_id_map = {}
    ub_cable_detection.guid_map = {}
    port_obj = {
        prop = port_props,

        src_guid = "guid-11111",

        get_prop = function(key)
            return port_obj.prop[key]
        end,
        clear_peer_info = function()
            return
        end,
        update_peer_info = function()
            return
        end
    }
    local ok, ret = pcall(ub_cable_detection.update_peer_info, ub_cable_detection, port_obj, peer_info)
    if not ok then
        print(ret)
    end
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_on_add_object()
    ub_cable_detection.cable_cfg_list = {}
    ub_cable_detection.node_port_addr_list = {}
    ub_cable_detection.init_debounce_value = function()
        return
    end
    local class_name = 'InterNodeCableConfiguration'
    local mds_obj = {
        name = 'abc'
    }
    local position = '1'
    local ok, _ = pcall(ub_cable_detection.on_add_object, ub_cable_detection, class_name, mds_obj, position)
    lu.assertEquals(ok, true)

    class_name = 'NodePortAddr'
    mds_obj = {
        name = 'abc'
    }
    position = '1'
    ok, _ = pcall(ub_cable_detection.on_add_object, ub_cable_detection, class_name, mds_obj, position)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_update_alarm()
    local alarm_list = {{EventCode = "0x28000039", ComponentName = 'NodePortAddr_1'}}

    local temp_get_event_list = event.get_latest_alarm_list
    event.get_latest_alarm_list = function ()
        return false
    end
    local ok = pcall(function ()
        ub_cable_detection:update_latest_alarm_list()
    end)
    lu.assertEquals(ok, false)

    event.get_latest_alarm_list = function ()
        return alarm_list
    end
    ok = pcall(function ()
        ub_cable_detection:update_latest_alarm_list()
    end)
    lu.assertEquals(ok, true)
    event.get_latest_alarm_list = temp_get_event_list
end

function TestUbCableDetection:test_generate_soft_event()
    local ok = pcall(function ()
        ub_cable_detection:generate_soft_event('111111111',true)
    end)
    lu.assertEquals(ok, false)
    ok = pcall(function ()
        ub_cable_detection:generate_soft_event('111111111',false)
    end)
    lu.assertEquals(ok, false)
    
end

function TestUbCableDetection:test_set_ubinterface_fault_led()
    local c_getsystemobj = client.GetSystemsBoardObjects
    client.GetSystemsBoardObjects = function ()
        return nil
    end
    local led_on_port_objs = {['1_1'] = true}
    local ok, ret = pcall(function ()
        ub_cable_detection:set_ubinterface_fault_led(led_on_port_objs)
    end)
    if not ok then
        print(ret)
    end
    lu.assertEquals(ok, true)
    
    led_on_port_objs = {
        ['3_2'] = true
    }
    ok = pcall(function ()
        ub_cable_detection:set_ubinterface_fault_led(led_on_port_objs)
    end)
    lu.assertEquals(ok, true)
    client.GetSystemsBoardObjects = function ()
        return {{Slot = 3}}
    end
    ok = pcall(function ()
        ub_cable_detection:set_ubinterface_fault_led(led_on_port_objs)
    end)
    lu.assertEquals(ok, true)
    local c_get_sub_paths = mdb_service.get_sub_paths
    mdb_service.get_sub_paths = function ()
        return {SubPaths = {'/bmc/kepler/Chassis/:ChassisId/UBInterfaceCards/1/Ports/1'}}
    end
    local c_match = utils_core.g_regex_match
    utils_core.g_regex_match = function ()
        return true
    end
    ok = pcall(function ()
        ub_cable_detection:set_ubinterface_fault_led(led_on_port_objs)
    end)
    lu.assertEquals(ok, true)
    local c_get_obj = mdb.get_object
    local ubport_obj = {PortId = 2, CableDetection = 0}
    mdb.get_object = function ()
        return ubport_obj
    end
    ok = pcall(function ()
        ub_cable_detection:set_ubinterface_fault_led(led_on_port_objs)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(ubport_obj.CableDetection, 1)
    client.GetSystemsBoardObjects = c_getsystemobj
    mdb_service.get_sub_paths = c_get_sub_paths
    utils_core.g_regex_match = c_match
    mdb.get_object = c_get_obj
end

function TestUbCableDetection:test_on_dump()
    local c_ub = ub_cable_detection.new({})
    c_ub.node_port_addr_list = {
        [1] = {
            src_info = 'src_info',
            target_info = 'target_info',
            peer_info = 'peer_info'
        }
    }
    local ok, _ = pcall(c_ub.on_dump, c_ub)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_parse_name()
    local sync_data = {
        {
            [1] = 63,
            [2] = {
                [1] = {
                    [1] = 'Name',
                    [2] = {
                        ["PropertyValue"] = "NodePortAddr_0_0_6_1_3_2_8000B417B30000000000000200A008CC",
                        ["Expired"] = "false"
                    }
                }
            }
        }
    }
    local ok, _ = pcall(ub_cable_detection.parse_name, ub_cable_detection, sync_data)
    lu.assertEquals(ok, true)
end

function TestUbCableDetection:test_read_ub_cable_info()
    local imu_cmd = {
        get_ub_port_info = function()
            return {
                link_status_list = {
                    1,2,3
                }
            }
        end
    }
    ub_cable_detection.node_port_addr_list = {
        [1] = {
            get_prop = function()
                return 1
            end
        }
    }
    local ok, _ = pcall(ub_cable_detection.read_ub_cable_info, ub_cable_detection, imu_cmd)
    lu.assertEquals(ok, true)
    ub_cable_detection.node_port_addr_list = {
        [1] = {
            get_prop = function(prop)
                if prop == 'InterfaceDevicePortId' then
                    return nil
                end
                return 1
            end
        }
    }
    ok, _ = pcall(ub_cable_detection.read_ub_cable_info, ub_cable_detection, imu_cmd)
    lu.assertEquals(ok, true)
end


