-- 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_eth = require 'bma.handles.handler_eth'
local c_network_adapter = require 'device.class.network_adapter'
local card_mgmt = require 'device.class.nic_mgmt.card_mgmt'
local port_mgmt = require 'device.class.nic_mgmt.port_mgmt'
local log = require 'mc.logging'
-- local c_bma = require 'bma'
-- local test_data = require 'test_datas.bma_nic'
-- local c_network_adapter_db = require 'network_adapter.db'
-- local c_ipv4_address = require 'device.class.ipv4_address'
-- local c_ipv6_address = require 'device.class.ipv6_address'
-- local c_network_bonding = require 'device.class.network_bonding'
-- 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'
-- local client = require 'network_adapter.client'

-- local ipv4_obj = test_utils.ipv4_obj
-- local ipv6_obj = test_utils.ipv6_obj
-- local find_address = test_utils.find_address

TEST_bma_eth = {}

-- function TEST_bma_eth:setUp()
--     client.OnDPUCardInterfacesAdded = function()
--         return {}
--     end
--     client.ForeachDPUCardObjects = function()
--         return {}
--     end
--     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 = 0xbc,
--             Device = 0,
--             Function = 0,
--             DevBus = 0xbc,
--             DevDevice = 0,
--             DevFunction = 1,
--             SpecialPcieCard = true
--         }, position)
--     self.network_port_obj = test_utils.add_network_port(self.object_manage, 'NetworkPort_1',
--         {PortID = 0, NetDevFuncType = 1}, position, self.network_adapter_obj)
--     test_utils.add_object_complete(self.object_manage, position)

--     self.object_manage.mc:prepare_ok()

--     local eth_data = json.decode(test_data.EthernetInterface)
--     self.url = eth_data['@odata.id']
-- end

-- function TEST_bma_eth:CreateVLANs()
-- end

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

-- function TEST_bma_eth:CreateIPv4Address(SystemID, ID1, ID2, ID3, prop_setting_cb)
--     return test_utils.CreateIPv4Address(SystemID, ID1, ID2, ID3, prop_setting_cb)
-- end

-- function TEST_bma_eth:CreateIPv6Address(SystemID, ID1, ID2, ID3, prop_setting_cb)
--     return test_utils.CreateIPv6Address(SystemID, ID1, ID2, ID3, prop_setting_cb)
-- end

-- function TEST_bma_eth: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.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.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)

--     local ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#ipv4_addrs, #data.IPv4Addresses)
--     for _, v in ipairs(ipv4_addrs) do
--         lu.assertEquals(ipv4_obj(v), ipv4_obj(find_address(data.IPv4Addresses, v.Address)))
--     end

--     local ipv6_addrs = c_ipv6_address.get_addresses()
--     lu.assertEquals(#ipv6_addrs, #data.IPv6Addresses)
--     for _, v in ipairs(ipv6_addrs) do
--         lu.assertEquals(ipv6_obj(v), ipv6_obj(find_address(data.IPv6Addresses, v.Address)))
--     end
-- end

-- function TEST_bma_eth:test_bma_update_ipv4_address()
--     local data = json.decode(test_data.EthernetInterface)

--     -- 装入初始化测试数据
--     self.bma:add(self.url, data)
--     local old_ipv4_addrs = c_ipv4_address.get_addresses()

--     -- 修改旧的 ipv4 地址信息
--     data.IPv4Addresses[1].SubnetMask = '0.0.0.1'
--     -- 再插入一个新 ipv4 地址
--     data.IPv4Addresses[#data.IPv4Addresses + 1] = {
--         Address = '127.0.0.0',
--         AddressOrigin = 'Static',
--         Gateway = {'196.128.0.0'},
--         SubnetMask = '0.0.0.0'
--     }
--     self.bma:add(self.url, data)

--     local new_ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#new_ipv4_addrs, #old_ipv4_addrs + 1)
--     for _, v in ipairs(new_ipv4_addrs) do
--         lu.assertEquals(ipv4_obj(v), ipv4_obj(find_address(data.IPv4Addresses, v.Address)))
--     end

--     -- 删除第一个地址
--     table.remove(data.IPv4Addresses, 1)
--     self.bma:add(self.url, data)
--     local last_ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#last_ipv4_addrs, #new_ipv4_addrs - 1)
--     for _, v in ipairs(last_ipv4_addrs) do
--         lu.assertEquals(ipv4_obj(v), ipv4_obj(find_address(data.IPv4Addresses, v.Address)))
--     end

--     -- 清空地址
--     data.IPv4Addresses = {}
--     self.bma:add(self.url, data)
--     lu.assertEquals(c_ipv4_address.get_addresses(), {})
-- end

-- function TEST_bma_eth:test_bma_add_and_update_ipv4_address()
--     local data = json.decode(test_data.EthernetInterface)

--     -- 装入初始化测试数据
--     self.bma:add(self.url, data)
--     local old_ipv4_addrs = c_ipv4_address.get_addresses()

--     -- 部分更新，new_data 仅包含要更新的信息
--     local new_data = {IPv4Addresses = test_utils.table_clone(data.IPv4Addresses)}
--     new_data.IPv4Addresses[1].SubnetMask = '0.0.0.1'
--     new_data.IPv4Addresses[#new_data.IPv4Addresses + 1] = {
--         Address = '127.0.0.0',
--         AddressOrigin = 'Static',
--         Gateway = {'196.128.0.0'},
--         SubnetMask = '0.0.0.0'
--     }
--     self.bma:update(self.url, new_data)

--     local new_ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#new_ipv4_addrs, #old_ipv4_addrs + 1)
--     for _, v in ipairs(new_ipv4_addrs) do
--         lu.assertEquals(ipv4_obj(v), ipv4_obj(find_address(new_data.IPv4Addresses, v.Address)))
--     end

--     -- 删除第一个地址
--     table.remove(new_data.IPv4Addresses, 1)
--     self.bma:update(self.url, new_data)
--     local last_ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#last_ipv4_addrs, #new_ipv4_addrs - 1)
--     for _, v in ipairs(last_ipv4_addrs) do
--         lu.assertEquals(ipv4_obj(v), ipv4_obj(find_address(new_data.IPv4Addresses, v.Address)))
--     end

--     -- 清空地址
--     new_data.IPv4Addresses = {}
--     self.bma:update(self.url, new_data)
--     lu.assertEquals(c_ipv4_address.get_addresses(), {})
-- end

-- function TEST_bma_eth:test_bma_update_ipv6_address()
--     local data = json.decode(test_data.EthernetInterface)

--     -- 装入初始化测试数据
--     self.bma:add(self.url, data)
--     local old_ipv6_addrs = c_ipv6_address.get_addresses()

--     -- 修改旧的 ipv6 地址信息
--     data.IPv6Addresses[1].PrefixLength = '128'
--     -- 再插入一个新 ipv6 地址
--     data.IPv6Addresses[#data.IPv6Addresses + 1] = {
--         Address = '8888:1871:1:0:f40:3935:cf5c:aaaa',
--         AddressOrigin = 'SLAAC',
--         AddressState = 'Deprecated',
--         PrefixLength = '64'
--     }
--     self.bma:add(self.url, data)

--     local new_ipv6_addrs = c_ipv6_address.get_addresses()
--     lu.assertEquals(#new_ipv6_addrs, #old_ipv6_addrs + 1)
--     for _, v in ipairs(new_ipv6_addrs) do
--         lu.assertEquals(ipv6_obj(v), ipv6_obj(find_address(data.IPv6Addresses, v.Address)))
--     end

--     -- 删除第一个地址
--     table.remove(data.IPv6Addresses, 1)
--     self.bma:add(self.url, data)
--     local last_ipv6_addrs = c_ipv6_address.get_addresses()
--     lu.assertEquals(#last_ipv6_addrs, #new_ipv6_addrs - 1)
--     for _, v in ipairs(last_ipv6_addrs) do
--         lu.assertEquals(ipv6_obj(v), ipv6_obj(find_address(data.IPv6Addresses, v.Address)))
--     end

--     -- 清空地址
--     data.IPv6Addresses = {}
--     self.bma:add(self.url, data)
--     lu.assertEquals(c_ipv6_address.get_addresses(), {})
-- end

-- function TEST_bma_eth:test_bma_add_team()
--     local data = json.decode(test_data.Team)
--     local url = data['@odata.id']
--     -- 装入初始化测试数据
--     self.bma:add(url, data)
--     local team_objs = c_network_bonding.collection:fetch({Id = 'team00'})
--     for _, v in pairs(team_objs) do
--         lu.assertEquals(v.Name, 'teamA')
--     end
--     -- ports更新测试
--     data['Ports'] = {'eno1'}
--     self.bma:add(url, data)
--     team_objs = c_network_bonding.collection:fetch({Id = 'team00'})
--     for _, v in pairs(team_objs) do
--         lu.assertEquals(v.Ports[1], 'eno1')
--     end
-- end

-- function TEST_bma_eth:test_bma_reset_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.url, data)
--     self.bma:on_reset()

--     lu.assertEquals(na.DisplayName, '')
--     lu.assertEquals(na.FirmwareVersion, '')
--     lu.assertEquals(na.DriverName, '')
--     lu.assertEquals(na.DriverVersion, '')
--     lu.assertEquals(na.RootBDF, huawei.BDFNumber.RootBDF)

--     lu.assertEquals(np.Name, '')
--     lu.assertEquals(np.OSLinkStatus, '')
--     lu.assertEquals(np.MACAddress, string.upper(data.MACAddress))
--     lu.assertEquals(np.BDF, huawei.BDFNumber.BDF)
--     lu.assertEquals(np.DriverName, '')
--     lu.assertEquals(np.DriverVersion, '')

--     local ipv4_addrs = c_ipv4_address.get_addresses()
--     lu.assertEquals(#ipv4_addrs, 0)
--     local ipv6_addrs = c_ipv6_address.get_addresses()
--     lu.assertEquals(#ipv6_addrs, 0)
-- end

-- function TEST_bma_eth:test_bma_update_speed_mbps_with_null()
--     local np = self.network_port_obj
--     local data = json.decode(test_data.EthernetInterface)

--     -- 测试正常的 SpeedMbps 值
--     data.SpeedMbps = 1000
--     self.bma:add(self.url, data)
--     lu.assertEquals(np.SpeedMbps, 1000)

--     -- 测试 SpeedMbps 为 json.null 时，应该设置为 4294967295
--     data.SpeedMbps = json.null
--     self.bma:add(self.url, data)
--     lu.assertEquals(np.SpeedMbps, 4294967295)

--     -- 测试 SpeedMbps 从 json.null 恢复为有效值
--     data.SpeedMbps = 10000
--     self.bma:add(self.url, data)
--     lu.assertEquals(np.SpeedMbps, 10000)

--     -- 测试 SpeedMbps 为 0 的情况
--     data.SpeedMbps = 0
--     self.bma:add(self.url, data)
--     lu.assertEquals(np.SpeedMbps, 0)

--     -- 再次测试 json.null 确保逻辑稳定
--     data.SpeedMbps = json.null
--     self.bma:add(self.url, data)
--     lu.assertEquals(np.SpeedMbps, 4294967295)
-- end

-- 覆盖 handler_eth.update_firmware_version_from_bma（经 update_network_card 调用）
local ETH_VIR_TYPE<const> = 6 -- 与 handler_eth.BUSINESSPORT_CARD_TYPE_VIR 一致
local NA<const> = 'N/A'

local function stub_log_quiet()
    local saved_info, saved_error = log.info, log.error
    log.info = function() end
    log.error = function() end
    return function()
        log.info = saved_info
        log.error = saved_error
    end
end

--- 构造最小 port / na / data，使 update_network_card 走到 update_firmware_version_from_bma
local function make_port_and_na(na_fw_initial)
    local na = {
        NodeId = 'ut_na_1',
        path = '/ut/na',
        FirmwareVersion = na_fw_initial,
        Model = 'ut_model_locked',
        Manufacturer = 'ut_mfr_locked',
        DisplayName = 'ut_display',
        DriverName = 'ut_drv',
        DriverVersion = 'ut_drv_ver',
        Description = 'ut_desc',
        RootBDF = '0000:00:00.0',
    }
    local port = {
        Type = 0,
        WorkloadType = 0,
        path = '/ut/port',
        CreatedByDeviceObject = false,
        get_parent = function()
            return { NodeId = 'ut_na_1' }
        end,
    }
    local data = { ['@odata.id'] = '/redfish/v1/ut/EthernetInterfaces/0' }
    return port, na, data
end

function TEST_bma_eth:test_update_firmware_version_from_bma_early_return_when_nil()
    local port, na, data = make_port_and_na('must_keep')
    local oem_huawei = { FirmwareVersion = nil, NICName = 'n' }

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function(_)
        return na
    end
    card_mgmt.get_instance = function()
        error('should not call card_mgmt when FirmwareVersion is nil')
    end

    c_handler_eth:update_network_card(port, data, oem_huawei)

    lu.assertEquals(na.FirmwareVersion, 'must_keep')
    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

function TEST_bma_eth:test_update_firmware_version_from_bma_sets_fw_and_calls_device_sync()
    local port, na, data = make_port_and_na('')
    local oem_huawei = { FirmwareVersion = 'BMA_FW_9.9', NICName = 'n' }
    local sync_calls = {}

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function(_)
        return na
    end
    card_mgmt.get_instance = function()
        return {
            update_firmware_version_to_device_obj = function(_, na_arg, ver)
                sync_calls[#sync_calls + 1] = { na_arg = na_arg, ver = ver }
            end,
        }
    end

    c_handler_eth:update_network_card(port, data, oem_huawei)

    lu.assertEquals(na.FirmwareVersion, 'BMA_FW_9.9')
    lu.assertEquals(#sync_calls, 1)
    lu.assertIs(sync_calls[1].na_arg, na)
    lu.assertEquals(sync_calls[1].ver, 'BMA_FW_9.9')

    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

function TEST_bma_eth:test_update_firmware_version_from_bma_keeps_existing_configured_fw()
    local port, na, data = make_port_and_na('already_from_sr')
    local oem_huawei = { FirmwareVersion = 'BMA_SHOULD_NOT_REPLACE', NICName = 'n' }
    local sync_calls = {}

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function(_)
        return na
    end
    card_mgmt.get_instance = function()
        return {
            update_firmware_version_to_device_obj = function(_, _, ver)
                sync_calls[#sync_calls + 1] = ver
            end,
        }
    end

    c_handler_eth:update_network_card(port, data, oem_huawei)

    lu.assertEquals(na.FirmwareVersion, 'already_from_sr')
    lu.assertEquals(#sync_calls, 1)
    lu.assertEquals(sync_calls[1], 'already_from_sr')

    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

function TEST_bma_eth:test_update_firmware_version_from_bma_json_null_maps_to_na()
    local port, na, data = make_port_and_na('')
    local oem_huawei = { FirmwareVersion = json.null, NICName = 'n' }
    local sync_calls = {}

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function(_)
        return na
    end
    card_mgmt.get_instance = function()
        return {
            update_firmware_version_to_device_obj = function(_, _, ver)
                sync_calls[#sync_calls + 1] = ver
            end,
        }
    end

    c_handler_eth:update_network_card(port, data, oem_huawei)

    lu.assertEquals(na.FirmwareVersion, NA)
    lu.assertEquals(#sync_calls, 1)
    lu.assertEquals(sync_calls[1], NA)

    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

function TEST_bma_eth:test_update_firmware_version_from_bma_pcall_swallows_card_mgmt_error()
    local port, na, data = make_port_and_na('')
    local oem_huawei = { FirmwareVersion = 'FW_OK', NICName = 'n' }

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function(_)
        return na
    end
    card_mgmt.get_instance = function()
        return {
            update_firmware_version_to_device_obj = function()
                error('dbus down')
            end,
        }
    end

    local ok = pcall(function()
        c_handler_eth:update_network_card(port, data, oem_huawei)
    end)
    lu.assertTrue(ok)
    lu.assertEquals(na.FirmwareVersion, 'FW_OK')

    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

function TEST_bma_eth:test_update_network_card_skips_virtual_and_workload()
    local na = { NodeId = 'x', FirmwareVersion = '', Model = 'm', Manufacturer = 'mf' }
    local data = { ['@odata.id'] = '/x' }
    local oem_huawei = { FirmwareVersion = 'FW', NICName = 'n' }
    local touched = false

    local orig_find = c_network_adapter.collection.find
    local orig_gi = card_mgmt.get_instance
    local restore_log = stub_log_quiet()
    c_network_adapter.collection.find = function()
        touched = true
        return na
    end
    card_mgmt.get_instance = function()
        error('should not run')
    end

    local port_vir = {
        Type = ETH_VIR_TYPE,
        WorkloadType = 0,
        path = '/v',
        get_parent = function() return { NodeId = 'x' } end,
    }
    c_handler_eth:update_network_card(port_vir, data, oem_huawei)
    lu.assertFalse(touched)

    local port_wl = {
        Type = 0,
        WorkloadType = 1,
        path = '/w',
        get_parent = function() return { NodeId = 'x' } end,
    }
    c_handler_eth:update_network_card(port_wl, data, oem_huawei)
    lu.assertFalse(touched)

    c_network_adapter.collection.find = orig_find
    card_mgmt.get_instance = orig_gi
    restore_log()
end

-- 覆盖 update_full_duplex_from_bma / update_speed_mbps_from_bma / update_auto_negotiation_from_bma
local function make_minimal_network_port()
    return {
        path = '/ut/np',
        Name = 'eno_ut',
        NetDevFuncType = 1,
        WorkloadType = 0,
        PortID = 0,
        NetworkAdapterId = 'ut_na',
        na_object_name = 'na_o',
        np_object_name = 'np_o',
        FullDuplex = 0,
        SpeedMbps = 100,
        AutoSpeedNegotiation = 0,
        OSLinkStatus = 'N/A',
        set_link_status_numeric = function() end,
        update_ipv6_default_gateway = function() end,
    }
end

local function stub_update_network_port_collateral()
    local saved = {
        bridge = c_handler_eth.update_physical_port_to_bridge_and_bonding,
        ip = c_handler_eth.update_ip_addresses,
        bdf = c_handler_eth.update_network_port_bdf,
    }
    c_handler_eth.update_physical_port_to_bridge_and_bonding = function() end
    c_handler_eth.update_ip_addresses = function() end
    c_handler_eth.update_network_port_bdf = function() end
    return function()
        c_handler_eth.update_physical_port_to_bridge_and_bonding = saved.bridge
        c_handler_eth.update_ip_addresses = saved.ip
        c_handler_eth.update_network_port_bdf = saved.bdf
    end
end

local function new_port_mgmt_recorder()
    local rec = { duplex = {}, speed = {}, autoneg = {} }
    local mgr = {
        set_link_status_to_device_obj = function() end,
        update_bdf_to_device_obj = function() end,
        update_full_duplex_to_device_obj = function(_, p, v)
            rec.duplex[#rec.duplex + 1] = { port = p, val = v }
        end,
        update_speed_mbps_to_device_obj = function(_, p, v)
            rec.speed[#rec.speed + 1] = { port = p, val = v }
        end,
        update_auto_negotiation_to_device_obj = function(_, p, v)
            rec.autoneg[#rec.autoneg + 1] = { port = p, val = v }
        end,
    }
    return mgr, rec
end

local function run_update_network_port(port, data, oem_huawei, recorder_factory)
    local orig_pm = port_mgmt.get_instance
    local restore_collateral = stub_update_network_port_collateral()
    local restore_log = stub_log_quiet()
    local mgr, rec = recorder_factory()
    port_mgmt.get_instance = function()
        return mgr
    end
    c_handler_eth:update_network_port(port, data, oem_huawei)
    restore_collateral()
    port_mgmt.get_instance = orig_pm
    restore_log()
    return rec
end

function TEST_bma_eth:test_update_full_duplex_from_bma_nil_skips()
    local port = make_minimal_network_port()
    port.FullDuplex = 88
    local data = { ['@odata.id'] = '/redfish/v1/x' }
    local oem_huawei = { NICName = port.Name }
    local rec = run_update_network_port(port, data, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port.FullDuplex, 88)
    lu.assertEquals(#rec.duplex, 0)
end

function TEST_bma_eth:test_update_full_duplex_from_bma_json_null_true_false()
    local oem_huawei = { NICName = 'eno_ut' }

    local port1 = make_minimal_network_port()
    local rec1 = run_update_network_port(port1,
        { ['@odata.id'] = '/x', FullDuplex = json.null }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port1.FullDuplex, 255)
    lu.assertEquals(#rec1.duplex, 1)
    lu.assertEquals(rec1.duplex[1].val, 255)

    local port2 = make_minimal_network_port()
    local rec2 = run_update_network_port(port2,
        { ['@odata.id'] = '/x', FullDuplex = true }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port2.FullDuplex, 1)
    lu.assertEquals(rec2.duplex[1].val, 1)

    local port3 = make_minimal_network_port()
    local rec3 = run_update_network_port(port3,
        { ['@odata.id'] = '/x', FullDuplex = false }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port3.FullDuplex, 0)
    lu.assertEquals(rec3.duplex[1].val, 0)
end

function TEST_bma_eth:test_update_full_duplex_from_bma_pcall_swallows_port_mgmt_error()
    local port = make_minimal_network_port()
    local data = { ['@odata.id'] = '/x', FullDuplex = true }
    local oem_huawei = { NICName = port.Name }
    local orig_pm = port_mgmt.get_instance
    local restore_collateral = stub_update_network_port_collateral()
    local restore_log = stub_log_quiet()
    port_mgmt.get_instance = function()
        return {
            set_link_status_to_device_obj = function() end,
            update_bdf_to_device_obj = function() end,
            update_full_duplex_to_device_obj = function()
                error('dbus')
            end,
            update_speed_mbps_to_device_obj = function() end,
            update_auto_negotiation_to_device_obj = function() end,
        }
    end
    local ok = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertTrue(ok)
    lu.assertEquals(port.FullDuplex, 1)
    restore_collateral()
    port_mgmt.get_instance = orig_pm
    restore_log()
end

function TEST_bma_eth:test_update_speed_mbps_from_bma_nil_skips()
    local port = make_minimal_network_port()
    port.SpeedMbps = 333
    local data = { ['@odata.id'] = '/x' }
    local oem_huawei = { NICName = port.Name }
    local rec = run_update_network_port(port, data, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port.SpeedMbps, 333)
    lu.assertEquals(#rec.speed, 0)
end

function TEST_bma_eth:test_update_speed_mbps_from_bma_json_null_and_numeric()
    local oem_huawei = { NICName = 'eno_ut' }

    local port1 = make_minimal_network_port()
    local rec1 = run_update_network_port(port1,
        { ['@odata.id'] = '/x', SpeedMbps = json.null }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port1.SpeedMbps, 4294967295)
    lu.assertEquals(#rec1.speed, 1)
    lu.assertEquals(rec1.speed[1].val, 4294967295)

    local port2 = make_minimal_network_port()
    port2.SpeedMbps = 100
    local rec2 = run_update_network_port(port2,
        { ['@odata.id'] = '/x', SpeedMbps = 25000 }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port2.SpeedMbps, 25000)
    lu.assertEquals(rec2.speed[1].val, 25000)
end

function TEST_bma_eth:test_update_speed_mbps_from_bma_pcall_swallows_port_mgmt_error()
    local port = make_minimal_network_port()
    local data = { ['@odata.id'] = '/x', SpeedMbps = 5000 }
    local oem_huawei = { NICName = port.Name }
    local orig_pm = port_mgmt.get_instance
    local restore_collateral = stub_update_network_port_collateral()
    local restore_log = stub_log_quiet()
    port_mgmt.get_instance = function()
        return {
            set_link_status_to_device_obj = function() end,
            update_bdf_to_device_obj = function() end,
            update_full_duplex_to_device_obj = function() end,
            update_speed_mbps_to_device_obj = function()
                error('dbus')
            end,
            update_auto_negotiation_to_device_obj = function() end,
        }
    end
    local ok = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertTrue(ok)
    lu.assertEquals(port.SpeedMbps, 5000)
    restore_collateral()
    port_mgmt.get_instance = orig_pm
    restore_log()
end

function TEST_bma_eth:test_update_auto_negotiation_from_bma_nil_skips()
    local port = make_minimal_network_port()
    port.AutoSpeedNegotiation = 44
    local data = { ['@odata.id'] = '/x' }
    local oem_huawei = { NICName = port.Name }
    local rec = run_update_network_port(port, data, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port.AutoSpeedNegotiation, 44)
    lu.assertEquals(#rec.autoneg, 0)
end

function TEST_bma_eth:test_update_auto_negotiation_from_bma_json_null_true_false()
    local oem_huawei = { NICName = 'eno_ut' }

    local port1 = make_minimal_network_port()
    local rec1 = run_update_network_port(port1,
        { ['@odata.id'] = '/x', AutoNeg = json.null }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port1.AutoSpeedNegotiation, 255)
    lu.assertEquals(rec1.autoneg[1].val, 255)

    local port2 = make_minimal_network_port()
    local rec2 = run_update_network_port(port2,
        { ['@odata.id'] = '/x', AutoNeg = true }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port2.AutoSpeedNegotiation, 1)
    lu.assertEquals(rec2.autoneg[1].val, 1)

    local port3 = make_minimal_network_port()
    local rec3 = run_update_network_port(port3,
        { ['@odata.id'] = '/x', AutoNeg = false }, oem_huawei, new_port_mgmt_recorder)
    lu.assertEquals(port3.AutoSpeedNegotiation, 0)
    lu.assertEquals(rec3.autoneg[1].val, 0)
end

function TEST_bma_eth:test_update_auto_negotiation_from_bma_pcall_swallows_port_mgmt_error()
    local port = make_minimal_network_port()
    local data = { ['@odata.id'] = '/x', AutoNeg = true }
    local oem_huawei = { NICName = port.Name }
    local orig_pm = port_mgmt.get_instance
    local restore_collateral = stub_update_network_port_collateral()
    local restore_log = stub_log_quiet()
    port_mgmt.get_instance = function()
        return {
            set_link_status_to_device_obj = function() end,
            update_bdf_to_device_obj = function() end,
            update_full_duplex_to_device_obj = function() end,
            update_speed_mbps_to_device_obj = function() end,
            update_auto_negotiation_to_device_obj = function()
                error('dbus')
            end,
        }
    end
    local ok = pcall(function()
        c_handler_eth:update_network_port(port, data, oem_huawei)
    end)
    lu.assertTrue(ok)
    lu.assertEquals(port.AutoSpeedNegotiation, 1)
    restore_collateral()
    port_mgmt.get_instance = orig_pm
    restore_log()
end
