-- 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_network_adapter = require 'device.class.network_adapter'
local c_object_manage = require 'mc.orm.object_manage'
local c_tasks = require 'mc.orm.tasks'
local c_network_port = require 'device.class.network_port'
local c_network_adapter_db = require 'network_adapter.db'
local hook_tasks = require 'test_common.hook_tasks'
local test_utils = require 'test_utils'
local create_mds_object = test_utils.create_mds_object
local vos = require 'utils.vos'
local utils = require 'mc.utils'
local c_optical_module = require 'device.class.optical_module'
local hi1822 = require 'hardware_config.Hi1822'
local cx4 = require 'hardware_config.CX4'
local cx5 = require 'hardware_config.CX5'
local cx7 = require 'hardware_config.CX7'
local bf3 = require 'hardware_config.BF3'
local bs = require 'mc.bitstring'
local c_network_bonding = require 'device.class.network_bonding'
local c_handler_eth = require 'bma.handles.handler_eth'
local pmu_cmd = require 'imu'

TEST_network_adapter = {}

function TEST_network_adapter:test_RequireConfig()
    local table_1822 = hi1822.smbus().properties
    for _, func in pairs(table_1822) do
        pcall(func.response)
    end
    local table_1822_ncsi = hi1822.mctp().properties
    for _, func in pairs(table_1822_ncsi) do
        pcall(func.response)
    end
    local table_cx4 = cx4.mctp().properties
    for _, func in pairs(table_cx4) do
        pcall(func.response)
    end
    local table_cx5 = cx5.mctp().properties
    for _, func in pairs(table_cx5) do
        pcall(func.response)
    end
    local table_cx7 = cx7.mctp().properties
    for _, func in pairs(table_cx7) do
        pcall(func.response)
    end
    local table_bf3 = bf3.mctp().properties
    for _, func in pairs(table_bf3) do
        pcall(func.response)
    end
end

function TEST_network_adapter: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.position = '00010102'
    self.network_adapter_obj = test_utils.add_network_adapter(self.object_manage, 'NetworkAdapter_1', {
        BoardID = 0x123,
        DeviceLocator = 'some locator',
        Name = "",
        Position = self.position
    })
    self.object_manage.mc:before_add_object(c_network_adapter.__class_name, self.network_adapter_obj)
    self.object_manage.mc:add_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)
    self.object_manage.mc:prepare_ok()
    self:flush()
end

function TEST_network_adapter:CreateVLANs()
end

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

function TEST_network_adapter:flush()
    c_tasks.get_instance():run_all_task()
    self.database.db:flush()
end

function TEST_network_adapter:test_can_construct_network_adapter()
    local adapter = c_network_adapter.collection:find({Position = self.position})

    lu.assertEquals(adapter.Position, self.position)
    lu.assertEquals(adapter.ID, self.position .. 'somelocator')
    lu.assertEquals(adapter.BoardIDHex, '0x0123')
end

function TEST_network_adapter:test_can_deconstruct_network_adapter()
    local adapter = c_network_adapter.collection:find({Position = self.position})

    local r = {ports_deconstruct_called = 0, scheduler_deconstruct_called = 0}
    c_network_port.stop = function()
        r.ports_deconstruct_called = 1
    end

    -- 自发现添加 port 对象
    local mds_network_port = create_mds_object('NetworkPort_1', {PortID = 1, NetDevFuncType = 1},
        self.network_adapter_obj,
        '/bmc/kepler/Systems/1/NetworkAdapters/NetworkAdapter_1_0001010101')
    self.object_manage.mc:before_add_object(c_network_port.__class_name, mds_network_port)
    self.object_manage.mc:add_object(c_network_port.__class_name, mds_network_port, self.position)

    local scheduler_expect_call<const> = 10
    for i = 1, scheduler_expect_call do
        adapter.ncsi_schedulers[i] = {
            deconstruct = function()
                r.scheduler_deconstruct_called = r.scheduler_deconstruct_called + 1
            end
        }
    end
    self.object_manage.mc:add_object_complete(self.position)

    -- 自发现删除对象
    self.object_manage.mc:del_object(c_network_port.__class_name, mds_network_port, self.position)
    self.object_manage.mc:del_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)

    lu.assertEquals(r.ports_deconstruct_called, 1)
    lu.assertEquals(r.scheduler_deconstruct_called, scheduler_expect_call)
end

function TEST_network_adapter:test_network_adapter_collect_optical_info()
    self.network_adapter_obj.Name = 'SP580'
    self.network_portobj = test_utils.add_network_port(self.object_manage, 'NetworkPort_1',
        {PortID = 0, PermanentMACAddress = '04:8C:16:59:C9:3E', MACAddress = '04:8C:16:59:C9:3E',
            LinkStatus = 'LinkDown'},
        self.position, self.network_adapter_obj)
    self.optical_module_obj = test_utils.add_optical_module(self.object_manage, 'OpticalModule_1',
        {FaultState = 0, Presence = 0, Manufacturer = 'Huawei', SerialNumber = 'INHAX4480013'},
        self.position, self.network_portobj)
    test_utils.add_object_complete(self.object_manage, self.position)
    local bma = {handles = {[3] = {objects = {[self.network_portobj] =
        '/redfish/v1/Sms/1/Systems/1/EthernetInterfaces/0000:bc:00.0_0000:bd:00.0'}}}}
    c_network_adapter.collect_optical_info('/var/log', bma)
    lu.assertEquals(vos.get_file_accessible('/var/log/net_card_info'), true)
    self.object_manage.mc:del_object(c_network_adapter.__class_name, self.network_adapter_obj,
        self.position)
    self.object_manage.mc:del_object(c_network_port.__class_name, self.network_portobj, self.position)
    self.object_manage.mc:del_object(c_optical_module.__class_name, self.optical_module_obj, self.position)
    utils.remove_file('/var/log/net_card_info')
end

-- 十六进制字符转ascii字符
local function hex_to_ascii(mac)
    local hex = string.gsub(mac, ":", "")
    -- 非法mac地址置零
    if string.find(hex, "[^%x]") then
        return "\0\0\0\0\0\0"
    end
    local str = {}
    local byte
    -- 十六进制字符转换
    for i = 1, #hex, 2 do
        byte = string.sub(hex, i, i + 1)
        table.insert(str, string.char(tonumber(byte, 16)))
    end
    return table.concat(str)
end

local function get_port_mac_hex(port)
    local str = "\0\0\0\0\0\0"
    if not port then
        return str
    end

    local mac = port.PermanentMACAddress
    -- 非法mac地址置零
    if mac and #mac == 17 then
        str = hex_to_ascii(mac)
    end
    return str
end

local function get_uuid_pack_str(ports)
    table.sort(ports, function(a, b)
        return a.PortID < b.PortID
    end)
    local mac1 = get_port_mac_hex(ports[1])
    local mac2 = get_port_mac_hex(ports[2])
    local mac3 = get_port_mac_hex(ports[3])
    local mac4 = get_port_mac_hex(ports[4])
    -- 网口个数小于等于2个时，取两个网口的mac共96位，剩余32位用0补齐
    -- 网口个数大于2个，取第一个网口的mac（48位），后三个网口的低24位共120位，剩余8位用0补齐
    if #ports <= 2 then
        return string.pack("<c6c6c4", mac1, mac2, string.rep('\0', 4))
    end
    return string.pack("<c6c3c3c3c1", mac1, string.sub(mac2, 4, 6),
                string.sub(mac3, 4, 6), string.sub(mac4, 4, 6), '\0')
end

local function regenerate_uuid_str(pack_string)
    local partten_mac = bs.new([[<<
        custom_a:48/big,
        custom_b:4,
        custom_c:4,
        custom_d:4,
        custom_e:4,
        custom_f:56/big,
        custom_g:8/big>>]])
    local uuid_part = partten_mac:unpack(pack_string)
    local uuid_param = {
        custom_a = uuid_part.custom_a,
        custom_b = uuid_part.custom_c,
        ver = 8,
        custom_c = uuid_part.custom_e,
        custom_d = uuid_part.custom_b,
        custom_e = uuid_part.custom_d,
        var = 8,
        custom_f = uuid_part.custom_f
    }
    local partten_uuid = bs.new([[<<
        custom_a:48/big,
        custom_b:4,
        ver:4,
        custom_c:4,
        custom_d:4,
        custom_e:4,
        var:4,
        custom_f:56/big>>]])
    return partten_uuid:pack(uuid_param)
end

function TEST_network_adapter:test_network_adapter_uuid()
    local network_port_objs = {
        {
            PermanentMACAddress = "00:15:5d:34:80:59",
            PortID = 0
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5A",
            PortID = 1
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5B",
            PortID = 2
        },
        {
            PermanentMACAddress = "00:15:5d:34:80:5C",
            PortID = 3
        }
    }

    local uuid8_1 = regenerate_uuid_str(get_uuid_pack_str({network_port_objs[1]}))
    local uuid8_2 = regenerate_uuid_str(get_uuid_pack_str({network_port_objs[1], network_port_objs[2]}))
    local uuid8_3 = regenerate_uuid_str(
        get_uuid_pack_str({network_port_objs[1], network_port_objs[2], network_port_objs[3]}))
    local uuid8_4 = regenerate_uuid_str(get_uuid_pack_str(network_port_objs))
    local uuid8_4_2 = regenerate_uuid_str(get_uuid_pack_str(network_port_objs))
    lu.assertEquals(uuid8_4, uuid8_4_2)
    lu.assertEquals(uuid8_1, '\x00\x15\x5D\x34\x80\x59\x80\x00\x80\x00\x00\x00\x00\x00\x00\x00')
    lu.assertEquals(uuid8_2, '\x00\x15\x5D\x34\x80\x59\x80\x01\x85\x5d\x34\x80\x5a\x00\x00\x00')
    lu.assertEquals(uuid8_3, '\x00\x15\x5D\x34\x80\x59\x83\x48\x80\x5a\x34\x80\x5b\x00\x00\x00')
    lu.assertEquals(uuid8_4, '\x00\x15\x5D\x34\x80\x59\x83\x48\x80\x5a\x34\x80\x5b\x34\x80\x5c')
end

function TEST_network_adapter:test_network_bonding_update_ports()
    local value = {
        Id = 1,
        Ports = 1
    }
    local ok, _ = pcall(function()
        c_network_bonding.update_ports(value)
    end)
    lu.assertEquals(ok, true)
end

function TEST_network_adapter:test_update_network_port_linkstatus()
    local port = {}
    local data = {}
    local oem_huawei = {}

    port.update_ipv6_default_gateway = function()
    end
    port.set_link_status_numeric = function()
    end

    data.LinkStatus = ''
    local ok, _ = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.LinkStatus, 'N/A')

    data.LinkStatus = 'LinkUp'
    local ok, _ = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(port.LinkStatus, 'LinkUp')
end

function TEST_network_adapter:test_mctp()
    c_network_adapter.ncsi_schedulers = {}
    c_network_adapter.ncsi_config_obj = {
        DisableHardwareArbitration = function()
            return {
                value = function() return true end
            }
        end,
        ChipTemp = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
        LinkAbility = function()
            return {
                value = function() return {} end
            }
        end,
        LinkInfo = function()
            return {
                value = function() return {} end
            }
        end,
        FaultStatCode = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end,
        VendorID = function()
            return {
                on_data_change = { on = function(data) end },
                start = function () end
            }
        end
    }
    c_network_adapter:disable_hardware_arbittration()
    cx5.mctp().properties.DisableHardwareArbitration.response('\x00\x00\x00\x00')
    local res = hi1822.mctp().properties.OpticalModuleInfo.response(
        string.pack('<c85','aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' .. '\x01' ..
            'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'))
    lu.assertEquals(res.MediumType, 'Optical')
    c_network_adapter:update_chip_temp_by_ncsi()
    c_network_adapter:update_link_ability_by_ncsi()
    c_network_adapter:update_link_status_by_ncsi()
    c_network_adapter:update_error_code_by_ncsi()
    c_network_adapter:update_ncsi_firmware({})
end

function TEST_network_adapter:test_get_max_npu_cdr_temp_from_imu()
    c_network_adapter:get_max_npu_cdr_temp_from_imu()
end

function TEST_network_adapter:test_update_max_npu_cdr_temp_from_imu()
    local network_port_objs = {
        { PortID = 0 }, { PortID = 1 }, { PortID = 2 }, { PortID = 3 }
    }
    for _, port in ipairs(network_port_objs) do
        port.get_npu_cdr_temp_from_imu = function()
            return 40
        end
    end
    c_network_adapter:update_max_npu_cdr_temp_from_imu(network_port_objs)
    lu.assertEquals(c_network_adapter.TemperatureCelsius, 40)
end
function TEST_network_adapter:test_register_npu_listen_callback()
    local client = require 'network_adapter.client'
    local func = client.OnFruCtrlPropertiesChanged
    client.values = {
        SysResetDetected = {
            value = function()
                return 1
            end
        },
        PowerState = {
            value = function()
                return "OFF"
            end
        }
    }
    client.OnFruCtrlPropertiesChanged = function(self, cb)
        cb(self.values, 1, 2)
    end
    local collection = c_network_port.collection
    c_network_port.collection = {
        obj = {
            get_parent = function()
                return {
                    Model = "NPU"
                }
            end
        }
    }
    c_network_port.collection.fold = function(self, cb)
        cb(1, self.obj)
    end
    c_network_adapter.register_npu_listen_callback()
    client.OnFruCtrlPropertiesChanged = func
    c_network_port.collection = collection
end

function TEST_network_adapter:test_firmware_version()
    local properties = hi1822.mctp().properties
    local vendor = properties.VendorID
    local response =vendor.response
    local tmp = bs.new
    bs.new = function()
        return {unpack = function(data, flag)
            return {
                ncsi_version_major=8,
                ncsi_version_minor=8, 
                ncsi_version_update=8, 
                ncsi_version_alpha1=8, 
                ncsi_version_alpha2=8,
                firmware_version_0=8,
                firmware_version_1=8,
                firmware_version_2=8,
                firmware_version_3=8,
                pci_vid=16,
                pci_did=16,
                pci_svid=16,
                pci_ssid=16
            }
        end, pack = function()
            return "test"
        end}
    end
    local ok, _ = pcall(function()
        return response(1)
    end)
    lu.assertEquals(ok, true)
    bs.new = tmp
end

function TEST_network_adapter:test_update_ncsi_properties()
    local ports = {
        ['BF3'] = {
            set_ncsi_config_obj = function ()
                return nil
            end,
            set_package_id = function(port, package_id)
                port.package_id = package_id
            end,
            initialize_ncsi_channel = function ()
                return nil
            end,
            update_ncsi_properties = function ()
                return nil
            end
        }
    }
    c_network_adapter.update_network_adapter = function ()
        return nil
    end
    c_network_adapter.package_id = 0
    c_network_adapter.Type = 'test'
    c_network_adapter.Model = 'BF3'
    pcall(c_network_adapter.update_ncsi_properties, c_network_adapter, ports)
    lu.assertEquals(ports['BF3'].package_id, 0)
end

function TEST_network_adapter:test_set_npu_max_sfp_temp()
    local c_next_tick = c_network_adapter.next_tick
    local c_update_npu_max_sfp_temp = c_network_adapter.update_npu_max_sfp_temp
    c_network_adapter.update_npu_max_sfp_temp = function ()
        error('end')
    end
    c_network_adapter.next_tick = function(_, cb)
        return cb()
    end
    local ok, _ = pcall(function()
        return c_network_adapter:set_npu_max_sfp_temp()
    end)
    c_network_adapter.next_tick = c_next_tick
    c_network_adapter.update_npu_max_sfp_temp = c_update_npu_max_sfp_temp
    lu.assertEquals(ok, false)
end

function TEST_network_adapter:test_update_npu_max_sfp_temp()
    local ops = {
        { ReadingCelsius = 20 }, { ReadingCelsius = 40 }, { ReadingCelsius = 10 }, { ReadingCelsius = 0 }
    }
    c_network_adapter:update_npu_max_sfp_temp(ops)
    lu.assertEquals(c_network_adapter.SFPMaxTemperatureCelsius, 40)
end

function TEST_network_adapter:test_update_bdf()
    c_network_adapter.Bus = 1
    c_network_adapter.Device = 2
    c_network_adapter.Function = 3
    c_network_adapter:update_root_bdf()
    lu.assertEquals(c_network_adapter.RootBDF, '0000:01:02.3')
end

local pcie_info = {
    system_id = 1,
    is_local = false,
    cpu_id = 1,
    bus_num = 1,
    device_num = 1,
    function_num = 1,
    did = 0,
    vid = 0,
    sdid = 0,
    svid = 0
}

-- 测试四元组信息解析成功
function TEST_network_adapter:test_parse_quater_info_succ()
    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertEquals(pcie_info.did, 0xd120)
    lu.assertEquals(pcie_info.svid, 0x19e5)
end

-- 测试四元组信息校验失败
function TEST_network_adapter:test_update_quater_info_fail()
    pmu_cmd.get_info_from_pmu = function()
        return {0xff, 0xff, 0xff, 0xff}
    end
    local ret = c_network_adapter:verify_vid_did_svid_sdid(pcie_info)
    lu.assertIsFalse(ret)
end

-- 测试获取四元组信息
function TEST_network_adapter:test_update_quater_info()
    pmu_cmd.get_info_from_pmu = function()
        return {0xff, 0xff, 0xff, 0xff}
    end
    local ret = c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertIsFalse(ret)

    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    ret = c_network_adapter:get_vid_did_svid_sdid(pcie_info)
    lu.assertEquals(pcie_info.did, 0xd120)
    lu.assertIsTrue(ret)
end

-- 测试四元组信息校验成功
function TEST_network_adapter:test_verify_quater_info()
    pmu_cmd.get_info_from_pmu = function()
        return {0xe5, 0x19, 0x20, 0xd1}
    end
    pcie_info.sdid = 0
    local ret = c_network_adapter:verify_vid_did_svid_sdid(pcie_info)
    lu.assertIsTrue(ret)
    lu.assertNotEquals(pcie_info.sdid, 0xd120)
end