-- 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_bma = require 'bma'
local test_data = require 'test_datas.bma_nic'
local c_network_adapter_db = require 'network_adapter.db'
local c_vlan_ipv4_address = require 'device.class.vlan.ipv4_address'
local c_vlan_ipv6_address = require 'device.class.vlan.ipv6_address'
local c_vlan = require 'device.class.vlan'
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 test_common = require 'test_common.utils'

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

TEST_bma_vlan = {}

function TEST_bma_vlan:setUp()
    self.bus = test_common.dbus_launch('../.dbus', nil, true)
    self.database = c_network_adapter_db(':memory:')

    hook_tasks.hook()
    self.object_manage = c_object_manage.new(self.database, self.bus)
    self.object_manage.app = self
    self.tasks = c_tasks.new()

    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 = 0xbd,
            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.eth_data = json.decode(test_data.EthernetInterface)
    self.vlan_data = json.decode(test_data.VLanNetworkInterface)
    self.eth_url = self.eth_data['@odata.id']
    self.vlan_url = self.vlan_data['@odata.id']
end

function TEST_bma_vlan:CreateVLANs()
end

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

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

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

function TEST_bma_vlan:CreateVLANIPv4Address(SystemID, ID1, ID2, ID3, ID4, prop_setting_cb)
    return test_utils.CreateVLANIPv4Address(SystemID, ID1, ID2, ID3, ID4, prop_setting_cb)
end

function TEST_bma_vlan:CreateVLANIPv6Address(SystemID, ID1, ID2, ID3, ID4, prop_setting_cb)
    return test_utils.CreateVLANIPv6Address(SystemID, ID1, ID2, ID3, ID4, prop_setting_cb)
end

function TEST_bma_vlan:CreateVLAN(SystemID, ID1, ID2, ID3, prop_setting_cb)
    return test_utils.CreateVLAN(SystemID, ID1, ID2, ID3, prop_setting_cb)
end

function TEST_bma_vlan:test_bma_add_vlan()
    self.object_manage.mc:prepare_ok()

    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)

    local port = self.network_port_obj
    local vlans = c_vlan.get_vlans(port.PortID, port.NetworkAdapterId)
    lu.assertEquals(#vlans, 1)
    lu.assertEquals(vlans[1].VLANId, self.vlan_data.VLANId)
end

function TEST_bma_vlan:test_bma_update_ipv4_address()
    self.object_manage.mc:prepare_ok()

    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)
    local old_ipv4_addrs = c_vlan_ipv4_address.get_addresses()

    local data = self.vlan_data

    -- 修改旧的 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.vlan_url, data)

    local new_ipv4_addrs = c_vlan_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.vlan_url, data)
    local last_ipv4_addrs = c_vlan_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.vlan_url, data)
    lu.assertEquals(c_vlan_ipv4_address.get_addresses(), {})
end

function TEST_bma_vlan:test_bma_add_and_update_ipv4_address()
    self.object_manage.mc:prepare_ok()

    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)
    local old_ipv4_addrs = c_vlan_ipv4_address.get_addresses()

    local data = self.vlan_data

    -- 部分更新，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.vlan_url, new_data)

    local new_ipv4_addrs = c_vlan_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.vlan_url, new_data)
    local last_ipv4_addrs = c_vlan_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.vlan_url, new_data)
    lu.assertEquals(c_vlan_ipv4_address.get_addresses(), {})
end

function TEST_bma_vlan:test_bma_update_ipv6_address()
    self.object_manage.mc:prepare_ok()

    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)
    local old_ipv6_addrs = c_vlan_ipv6_address.get_addresses()
    local data = self.vlan_data

    -- 修改旧的 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.vlan_url, data)

    local new_ipv6_addrs = c_vlan_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.vlan_url, data)
    local last_ipv6_addrs = c_vlan_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.vlan_url, data)
    lu.assertEquals(c_vlan_ipv6_address.get_addresses(), {})
end

-- 测试先加载 BMA 资源，再添加 VLAN 对象
function TEST_bma_vlan:test_bma_reload_vlan_resource()
    local resources = {[self.eth_url] = self.eth_data, [self.vlan_url] = self.vlan_data}
    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)

    self.object_manage.mc:prepare_ok()

    local np = self.network_port_obj

    -- 注册信号监听资源重新加载
    local reload_resources = {}
    self.bma.on_need_resource:on(function(path)
        for _, v in pairs(reload_resources) do
            if v == path then
                return
            end 
        end
        reload_resources[#reload_resources + 1] = path
        local data = resources[path]
        if data then
            self.bma:add(path, data)
        end
    end)

    -- 修改 BDF，触发重新加载资源, 这里应当触发卸载
    np.BDF = '0000:ff:00.0'
    self.tasks:run_all_task()
    lu.assertEquals(reload_resources, {self.eth_url, self.vlan_url})

    local port = self.network_port_obj
    local vlans = c_vlan.get_vlans(port.PortID, port.NetworkAdapterId)
    lu.assertEquals(#vlans, 1)
    lu.assertEquals(vlans[1].VLANId, self.vlan_data.VLANId)
end

function TEST_bma_vlan:test_bma_reset_vlan()
    self.object_manage.mc:prepare_ok()

    self.bma:add(self.eth_url, self.eth_data)
    self.bma:add(self.vlan_url, self.vlan_data)

    local port = self.network_port_obj
    local vlans = c_vlan.get_vlans(port.PortID, port.NetworkAdapterId)
    lu.assertEquals(#vlans, 1)
    lu.assertEquals(vlans[1].VLANId, self.vlan_data.VLANId)
    local ipv4 = c_vlan_ipv4_address.get_addresses()
    lu.assertEquals(#ipv4, 1)
    local ipv6 = c_vlan_ipv6_address.get_addresses()
    lu.assertEquals(#ipv6, 1)
    self.bma:on_reset()
    vlans = c_vlan.get_vlans(port.PortID, port.NetworkAdapterId)
    lu.assertEquals(#vlans, 0)
    ipv4 = c_vlan_ipv4_address.get_addresses()
    lu.assertEquals(#ipv4, 0)
    ipv6 = c_vlan_ipv6_address.get_addresses()
    lu.assertEquals(#ipv6, 0)
end
