openstack/os-net-config
Code Issues Proposed changes
27915c67c8
os-net-config/os_net_config/tests/test_utils.py
Karthik S 803840f65f Add SR-IOV, NIC Part support using nmstate provider 
Added support for SR-IOV and NIC Partitioning use cases
using nmstate provider. The implementation still uses the
dual parse of templates similar to the one in ifcfg provider.
Now the udev rules, sriov_config and sriov_config service will
not be used.

Change-Id: Iec98f3cacc061b37c6ecf6d53f453fd078791b16
2023-08-31 12:29:05 +05:30

1016 lines
41 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
# Copyright 2014 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import os.path
import random
import shutil
import tempfile
from unittest import mock
import yaml
from os_net_config import common
from os_net_config import objects
from os_net_config import sriov_config
from os_net_config.tests import base
from os_net_config import utils
from oslo_concurrency import processutils
_PCI_OUTPUT = '''driver: e1000e
version: 3.2.6-k
firmware-version: 0.13-3
expansion-rom-version:
bus-info: 0000:00:19.0
supports-statistics: yes
supports-test: yes
supports-eeprom-access: yes
supports-register-dump: yes
supports-priv-flags: no
'''
_VPPCTL_OUTPUT = '''
Name Idx State Counter Count
GigabitEthernet0/9/0 1 down
local0 0 down
'''
_VPPBOND_OUTPUT = """
Name Idx Link Hardware
BondEthernet0 3 up Slave-Idx: 1 2
TenGigabitEthernet2/0/0 1 slave TenGigabitEthernet2/0/0
TenGigabitEthernet2/0/1 2 slave TenGigabitEthernet2/0/1
"""
_INITIAL_VPP_CONFIG = '''
unix {
nodaemon
log /tmp/vpp.log
full-coredump
}
api-trace {
on
}
api-segment {
gid vpp
}
dpdk {
}
'''
udev_content = '# This file is autogenerated by os-net-config\n'\
'KERNEL=="eth1", RUN+="/bin/os-net-config-sriov -n %k:10"\n' \
'KERNEL=="eth2", RUN+="/bin/os-net-config-sriov -n %k:10"\n'
UDEV_FILE = '/tmp/70-os-net-config-sriov.'
class TestUtils(base.TestCase):
def setUp(self):
super(TestUtils, self).setUp()
rand = str(int(random.random() * 100000))
common.DPDK_MAPPING_FILE = '/tmp/dpdk_mapping_' + rand + '.yaml'
common.SRIOV_CONFIG_FILE = '/tmp/sriov_config_' + rand + '.yaml'
common._LOG_FILE = '/tmp/' + rand + 'os_net_config.log'
sriov_config._UDEV_LEGACY_RULE_FILE = UDEV_FILE + rand
def tearDown(self):
super(TestUtils, self).tearDown()
if os.path.isfile(common.DPDK_MAPPING_FILE):
os.remove(common.DPDK_MAPPING_FILE)
if os.path.isfile(common._LOG_FILE):
os.remove(common._LOG_FILE)
if os.path.isfile(common.SRIOV_CONFIG_FILE):
os.remove(common.SRIOV_CONFIG_FILE)
if os.path.isfile(sriov_config._UDEV_LEGACY_RULE_FILE):
os.remove(sriov_config._UDEV_LEGACY_RULE_FILE)
def test_ordered_active_nics(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
def test_is_available_nic(interface_name, check_active):
return True
self.stub_out('os_net_config.utils._is_available_nic',
test_is_available_nic)
for nic in ['a1', 'em1', 'em2', 'eth2', 'z1',
'enp8s0', 'enp10s0', 'enp1s0f0']:
with open(os.path.join(tmpdir, nic), 'w') as f:
f.write(nic)
nics = utils.ordered_active_nics()
self.assertEqual('em1', nics[0])
self.assertEqual('em2', nics[1])
self.assertEqual('eth2', nics[2])
self.assertEqual('a1', nics[3])
self.assertEqual('enp1s0f0', nics[4])
self.assertEqual('enp8s0', nics[5])
self.assertEqual('enp10s0', nics[6])
self.assertEqual('z1', nics[7])
shutil.rmtree(tmpdir)
def test_update_sriov_pf_map_new(self):
def get_numvfs_stub(pf_name):
return 0
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False)
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_pf_map))
test_sriov_pf_map = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'vdpa': False}]
self.assertListEqual(test_sriov_pf_map, sriov_pf_map)
def test_update_sriov_pf_map_with_same_numvfs(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False)
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_pf_map))
test_sriov_pf_map = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'vdpa': False}]
self.assertListEqual(test_sriov_pf_map, sriov_pf_map)
def test_update_sriov_pf_map_with_diff_numvfs(self):
def get_numvfs_stub(pf_name):
return 12
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
self.assertRaises(sriov_config.SRIOVNumvfsException,
utils.update_sriov_pf_map, 'eth1', 10, False)
def test_update_sriov_pf_map_new_with_promisc(self):
def get_numvfs_stub(pf_name):
return 0
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False, promisc='off')
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_pf_map))
test_sriov_pf_map = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'off',
'vdpa': False}]
self.assertListEqual(test_sriov_pf_map, sriov_pf_map)
def test_update_sriov_pf_map_new_with_vdpa(self):
def get_numvfs_stub(pf_name):
return 0
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False, vdpa=True)
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_pf_map))
test_sriov_pf_map = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'vdpa': True}]
self.assertListEqual(test_sriov_pf_map, sriov_pf_map)
def test_update_sriov_pf_map_exist(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
self.assertRaises(sriov_config.SRIOVNumvfsException,
utils.update_sriov_pf_map, 'eth1', 20, False)
def test_update_sriov_pf_map_exist_with_promisc(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
utils.update_sriov_pf_map('eth1', 10, False, promisc='off')
pf_final = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'off',
'vdpa': False}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(pf_map))
self.assertListEqual(pf_final, pf_map)
def test_update_sriov_pf_map_exist_with_vdpa(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
utils.update_sriov_pf_map('eth1', 10, False, vdpa=True)
pf_final = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': True}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(pf_map))
self.assertListEqual(pf_final, pf_map)
def test_update_sriov_pf_map_new_with_lag_candidate(self):
def get_numvfs_stub(pf_name):
return 0
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False, lag_candidate=True)
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_pf_map))
test_sriov_pf_map = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'vdpa': False,
'lag_candidate': True}]
self.assertListEqual(test_sriov_pf_map, sriov_pf_map)
def test_update_sriov_pf_map_exist_with_lag_candidate(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False, 'lag_candidate': False}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
utils.update_sriov_pf_map('eth1', 10, False, lag_candidate=True)
pf_final = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False, 'lag_candidate': True}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(pf_map))
self.assertListEqual(pf_final, pf_map)
def test_update_sriov_pf_map_exist_with_lag_candidate_not_exist_true(
self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
utils.update_sriov_pf_map('eth1', 10, False, lag_candidate=True)
pf_final = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False, 'lag_candidate': True}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(pf_map))
self.assertListEqual(pf_final, pf_map)
def test_update_sriov_pf_map_exist_with_lag_candidate_not_exist_false(
self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
pf_initial = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, pf_initial)
utils.update_sriov_pf_map('eth1', 10, False, lag_candidate=False)
pf_final = [{'device_type': 'pf', 'link_mode': 'legacy',
'name': 'eth1', 'numvfs': 10, 'promisc': 'on',
'vdpa': False, 'lag_candidate': False}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
pf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(pf_map))
self.assertListEqual(pf_final, pf_map)
def test_update_sriov_vf_map_minimal_new(self):
utils.update_sriov_vf_map('eth1', 2, 'eth1_2')
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_vf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_vf_map))
test_sriov_vf_map = [{'device_type': 'vf', 'name': 'eth1_2',
'min_tx_rate': 0, 'max_tx_rate': 0,
'device': {"name": "eth1", "vfid": 2}}]
self.assertListEqual(test_sriov_vf_map, sriov_vf_map)
def test_udev_rule_for_sriov_vf(self):
def get_numvfs_stub(pf_name):
return 10
self.stub_out('os_net_config.sriov_config.get_numvfs',
get_numvfs_stub)
utils.update_sriov_pf_map('eth1', 10, False)
utils.update_sriov_pf_map('eth2', 10, False)
utils.update_sriov_vf_map('eth1', 2, 'eth1_2')
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_vf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(3, len(sriov_vf_map))
test_sriov_vf_map = {'device': {'name': 'eth1', 'vfid': 2},
'device_type': 'vf', 'max_tx_rate': 0,
'min_tx_rate': 0, 'name': 'eth1_2'}
udev_file = open(sriov_config._UDEV_LEGACY_RULE_FILE, "w")
udev_file.write(udev_content)
udev_file.close()
# sriov_config.add_udev_rule_for_legacy_sriov_pf("eth1", 10)
for dev in sriov_vf_map:
if(dev.keys() == test_sriov_vf_map.keys()):
self.assertEqual(test_sriov_vf_map, dev)
expect = '# This file is autogenerated by os-net-config\n'\
'KERNEL=="eth2", RUN+="/bin/os-net-config-sriov -n %k:10"\n'
utils.nicpart_udev_rules_check()
f = open(sriov_config._UDEV_LEGACY_RULE_FILE, 'r')
self.assertEqual(expect, f.read())
f.close()
def test_update_sriov_vf_map_complete_new(self):
utils.update_sriov_vf_map('eth1', 2, 'eth1_2', vlan_id=10, qos=5,
spoofcheck="on", trust="on", state="enable",
macaddr="AA:BB:CC:DD:EE:FF", promisc="off",
pci_address="0000:80:00.1", max_tx_rate=10)
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
sriov_vf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(sriov_vf_map))
test_sriov_vf_map = [{'device_type': 'vf', 'name': 'eth1_2',
'device': {'name': 'eth1', 'vfid': 2},
'vlan_id': 10, 'qos': 5,
'min_tx_rate': 0, 'max_tx_rate': 10,
'spoofcheck': 'on', 'trust': 'on',
'state': 'enable',
'macaddr': 'AA:BB:CC:DD:EE:FF',
'promisc': 'off',
'pci_address': "0000:80:00.1"}]
self.assertListEqual(test_sriov_vf_map, sriov_vf_map)
def test_update_sriov_vf_map_exist(self):
vf_initial = [{'device_type': 'vf', 'name': 'eth1_2',
'device': {"name": "eth1", "vfid": 2}}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, vf_initial)
utils.update_sriov_vf_map('eth1', 2, 'eth1_2', vlan_id=10, qos=5,
spoofcheck="on", trust="on", state="enable",
macaddr="AA:BB:CC:DD:EE:FF", promisc="off",
pci_address="0000:80:00.1", max_tx_rate=10)
vf_final = [{'device_type': 'vf', 'name': 'eth1_2',
'device': {'name': 'eth1', 'vfid': 2},
'vlan_id': 10, 'qos': 5,
'min_tx_rate': 0, 'max_tx_rate': 10,
'spoofcheck': 'on', 'trust': 'on',
'state': 'enable',
'macaddr': 'AA:BB:CC:DD:EE:FF',
'promisc': 'off',
'pci_address': '0000:80:00.1'}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
vf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(vf_map))
self.assertListEqual(vf_final, vf_map)
def test_update_sriov_vf_map_exist_complete(self):
vf_initial = [{'device_type': 'vf', 'name': 'eth1_2',
'device': {'name': 'eth1', 'vfid': 2},
'vlan_id': 10, 'qos': 5,
'min_tx_rate': 0, 'max_tx_rate': 00,
'spoofcheck': 'on', 'trust': 'on',
'state': 'enable',
'macaddr': 'AA:BB:CC:DD:EE:FF',
'promisc': 'off',
'pci_address': "0000:80:00.1"}]
utils.write_yaml_config(common.SRIOV_CONFIG_FILE, vf_initial)
utils.update_sriov_vf_map('eth1', 2, 'eth1_2', vlan_id=100, qos=15,
spoofcheck="off", trust="off", state="auto",
macaddr="BB:BB:CC:DD:EE:FF", promisc="on",
pci_address="0000:80:00.1", max_tx_rate=40)
vf_final = [{'device_type': 'vf', 'name': 'eth1_2',
'device': {'name': 'eth1', 'vfid': 2},
'vlan_id': 100, 'qos': 15,
'min_tx_rate': 0, 'max_tx_rate': 40,
'spoofcheck': 'off', 'trust': 'off',
'state': 'auto',
'macaddr': 'BB:BB:CC:DD:EE:FF',
'promisc': 'on',
'pci_address': '0000:80:00.1'}]
contents = common.get_file_data(common.SRIOV_CONFIG_FILE)
vf_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(vf_map))
self.assertListEqual(vf_final, vf_map)
def test_get_vf_devname_net_dir_not_found(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
self.assertRaises(utils.SriovVfNotFoundException,
utils.get_vf_devname, "eth1", 1)
shutil.rmtree(tmpdir)
def test_get_vf_devname_vf_dir_found_in_map(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
def test_get_vf_name_from_map(pf_name, vfid):
return pf_name + '_' + str(vfid)
self.stub_out('os_net_config.utils._get_vf_name_from_map',
test_get_vf_name_from_map)
vf_path = os.path.join(common.SYS_CLASS_NET, 'eth1/device/virtfn1')
os.makedirs(vf_path)
self.assertEqual(utils.get_vf_devname("eth1", 1), "eth1_1")
shutil.rmtree(tmpdir)
def test_get_vf_devname_vf_dir_not_found(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
def test_get_vf_name_from_map(pf_name, vfid):
return None
self.stub_out('os_net_config.utils._get_vf_name_from_map',
test_get_vf_name_from_map)
vf_path = os.path.join(common.SYS_CLASS_NET, 'eth1/device/virtfn1')
os.makedirs(vf_path)
self.assertRaises(utils.SriovVfNotFoundException,
utils.get_vf_devname, "eth1", 1)
shutil.rmtree(tmpdir)
def test_get_vf_devname_vf_dir_found(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
vf_path = os.path.join(common.SYS_CLASS_NET,
'eth1/device/virtfn1/net/eth1_1')
os.makedirs(vf_path)
self.assertEqual(utils.get_vf_devname("eth1", 1), "eth1_1")
shutil.rmtree(tmpdir)
def test_get_pci_address_success(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
out = _PCI_OUTPUT
return out, None
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
pci = utils.get_pci_address('nic2', False)
self.assertEqual('0000:00:19.0', pci)
def test_get_pci_address_exception(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
raise processutils.ProcessExecutionError
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
pci = utils.get_pci_address('nic2', False)
self.assertEqual(None, pci)
def test_get_pci_address_error(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
return None, 'Error'
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
pci = utils.get_pci_address('nic2', False)
self.assertEqual(None, pci)
def test_get_stored_pci_address_success(self):
def test_get_dpdk_map():
return [{'name': 'eth1', 'pci_address': '0000:00:09.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
self.stub_out('os_net_config.common.get_dpdk_map', test_get_dpdk_map)
pci = utils.get_stored_pci_address('eth1', False)
self.assertEqual('0000:00:09.0', pci)
def test_get_stored_pci_address_empty(self):
def test_get_dpdk_map():
return []
self.stub_out('os_net_config.common.get_dpdk_map', test_get_dpdk_map)
pci = utils.get_stored_pci_address('eth1', False)
self.assertEqual(None, pci)
def test_get_vendor_id_success(self):
mocked_open = mock.mock_open(read_data='0x15b3\n')
with mock.patch('os_net_config.common.open', mocked_open, create=True):
vendor = common.get_vendor_id('nic2')
self.assertEqual('0x15b3', vendor)
def test_get_vendor_id_exception(self):
mocked_open = mock.mock_open()
mocked_open.side_effect = IOError
with mock.patch('os_net_config.common.open', mocked_open, create=True):
vendor = common.get_vendor_id('nic2')
self.assertEqual(None, vendor)
def test_get_device_id_success(self):
mocked_open = mock.mock_open(read_data='0x1003\n')
with mock.patch('os_net_config.common.open', mocked_open, create=True):
device = common.get_device_id('nic2')
self.assertEqual('0x1003', device)
def test_get_device_id_exception(self):
mocked_open = mock.mock_open()
mocked_open.side_effect = IOError
with mock.patch('os_net_config.utils.open', mocked_open, create=True):
device = common.get_device_id('nic2')
self.assertEqual(None, device)
def test_bind_dpdk_interfaces(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
out = _PCI_OUTPUT
return out, None
if 'driverctl' in name:
return None, None
def test_get_dpdk_mac_address(name):
return '01:02:03:04:05:06'
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
self.stub_out('os_net_config.common._get_dpdk_mac_address',
test_get_dpdk_mac_address)
try:
utils.bind_dpdk_interfaces('nic2', 'vfio-pci', False)
except common.OvsDpdkBindException:
self.fail("Received OvsDpdkBindException unexpectedly")
def test_bind_dpdk_interfaces_fail(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
out = _PCI_OUTPUT
return out, None
if 'driverctl' in name:
return None, 'Error'
def test_get_dpdk_mac_address(name):
return '01:02:03:04:05:06'
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
self.stub_out('os_net_config.common._get_dpdk_mac_address',
test_get_dpdk_mac_address)
self.assertRaises(common.OvsDpdkBindException,
utils.bind_dpdk_interfaces, 'eth1', 'vfio-pci',
False)
def test_bind_dpdk_interfaces_skip_valid_device(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
return None, 'Error'
if 'driverctl' in name:
return None, None
def test_get_dpdk_mac_address(name):
return '01:02:03:04:05:06'
def test_get_dpdk_map():
return [{'name': 'eth1', 'pci_address': '0000:00:09.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
self.stub_out('os_net_config.common.get_dpdk_map', test_get_dpdk_map)
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
self.stub_out('os_net_config.utils_get_dpdk_mac_address',
test_get_dpdk_mac_address)
try:
utils.bind_dpdk_interfaces('eth1', 'vfio-pci', False)
except common.OvsDpdkBindException:
self.fail("Received OvsDpdkBindException unexpectedly")
def test_bind_dpdk_interfaces_fail_invalid_device(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
return None, 'Error'
if 'driverctl' in name:
return None, None
def test_get_dpdk_mac_address(name):
return '01:02:03:04:05:06'
def test_get_dpdk_map():
return [{'name': 'eth1', 'pci_address': '0000:00:09.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
self.stub_out('os_net_config.common.get_dpdk_map',
test_get_dpdk_map)
self.stub_out('oslo_concurrency.processutils.execute',
test_execute)
self.stub_out('os_net_config.utils._get_dpdk_mac_address',
test_get_dpdk_mac_address)
self.assertRaises(common.OvsDpdkBindException,
utils.bind_dpdk_interfaces, 'eth2', 'vfio-pci',
False)
def test_bind_dpdk_interfaces_same_driver(self):
mocked_open = mock.mock_open(read_data='DRIVER=vfio-pci\n')
self.stub_out('os_net_config.utils.open', mocked_open)
mocked_logger = mock.Mock()
self.stub_out('os_net_config.utils.logger.info', mocked_logger)
try:
utils.bind_dpdk_interfaces('eth1', 'vfio-pci', False)
except common.OvsDpdkBindException:
self.fail("Received OvsDpdkBindException unexpectedly")
msg = "Driver (vfio-pci) is already bound to the device (eth1)"
mocked_logger.assert_called_with(msg)
def test_get_interface_driver(self):
mocked_open = mock.mock_open(read_data='DRIVER=vfio-pci\n')
self.stub_out('os_net_config.utils.open', mocked_open)
driver = utils.get_interface_driver('eth1')
self.assertEqual(driver, 'vfio-pci')
def test_get_interface_driver_fail_none(self):
mocked_open = mock.mock_open(read_data='')
self.stub_out('os_net_config.utils.open', mocked_open)
driver = utils.get_interface_driver('eth1')
self.assertFalse(driver)
def test_get_interface_driver_fail_empty(self):
mocked_open = mock.mock_open(read_data='DRIVER\n')
self.stub_out('os_net_config.utils.open', mocked_open)
driver = utils.get_interface_driver('eth1')
self.assertFalse(driver)
def test__update_dpdk_map_new(self):
utils._update_dpdk_map('eth1', '0000:03:00.0', '01:02:03:04:05:06',
'vfio-pci')
contents = common.get_file_data(common.DPDK_MAPPING_FILE)
dpdk_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(dpdk_map))
dpdk_test = [{'name': 'eth1', 'pci_address': '0000:03:00.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
self.assertListEqual(dpdk_test, dpdk_map)
def test_update_dpdk_map_exist(self):
dpdk_test = [{'name': 'eth1', 'pci_address': '0000:03:00.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
utils.write_yaml_config(common.DPDK_MAPPING_FILE, dpdk_test)
utils._update_dpdk_map('eth1', '0000:03:00.0', '01:02:03:04:05:06',
'vfio-pci')
contents = common.get_file_data(common.DPDK_MAPPING_FILE)
dpdk_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(dpdk_map))
self.assertListEqual(dpdk_test, dpdk_map)
def test_update_dpdk_map_value_change(self):
dpdk_test = [{'name': 'eth1', 'pci_address': '0000:03:00.0',
'driver': 'vfio-pci'}]
utils.write_yaml_config(common.DPDK_MAPPING_FILE, dpdk_test)
dpdk_test = [{'name': 'eth1', 'pci_address': '0000:03:00.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
utils._update_dpdk_map('eth1', '0000:03:00.0', '01:02:03:04:05:06',
'vfio-pci')
try:
contents = common.get_file_data(common.DPDK_MAPPING_FILE)
except IOError:
pass
dpdk_map = yaml.safe_load(contents) if contents else []
self.assertEqual(1, len(dpdk_map))
self.assertListEqual(dpdk_test, dpdk_map)
def test_ordered_active_nics_with_dpdk_mapping(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
tmp_pci_dir = tempfile.mkdtemp()
self.stub_out('os_net_config.common._SYS_BUS_PCI_DEV', tmp_pci_dir)
physfn_path = common._SYS_BUS_PCI_DEV + '/0000:05:01.1/physfn'
os.makedirs(physfn_path)
def test_is_available_nic(interface_name, check_active):
return True
self.stub_out('os_net_config.utils._is_available_nic',
test_is_available_nic)
for nic in ['a1', 'em1', 'em2', 'eth2', 'z1',
'enp8s0', 'enp10s0', 'enp1s0f0']:
with open(os.path.join(tmpdir, nic), 'w') as f:
f.write(nic)
utils._update_dpdk_map('eth1', '0000:03:00.0', '01:02:03:04:05:06',
'vfio-pci')
utils._update_dpdk_map('p3p1', '0000:04:00.0', '01:02:03:04:05:07',
'igb_uio')
utils._update_dpdk_map('p3p0_0', '0000:05:01.1', 'AA:02:03:04:05:FF',
'vfio-pci')
nics = utils.ordered_active_nics()
self.assertEqual('em1', nics[0])
self.assertEqual('em2', nics[1])
self.assertEqual('eth1', nics[2]) # DPDK bound nic
self.assertEqual('eth2', nics[3])
self.assertEqual('a1', nics[4])
self.assertEqual('enp1s0f0', nics[5])
self.assertEqual('enp8s0', nics[6])
self.assertEqual('enp10s0', nics[7])
self.assertEqual('p3p1', nics[8]) # DPDK bound nic
self.assertEqual('z1', nics[9])
shutil.rmtree(tmpdir)
shutil.rmtree(tmp_pci_dir)
def test_ordered_active_nics_with_dpdk_mapping_of_vf(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
tmp_pci_dir = tempfile.mkdtemp()
self.stub_out('os_net_config.common._SYS_BUS_PCI_DEV', tmp_pci_dir)
physfn_path = common._SYS_BUS_PCI_DEV + '/0000:05:01.1/physfn'
os.makedirs(physfn_path)
def test_is_available_nic(interface_name, check_active):
return True
self.stub_out('os_net_config.utils._is_available_nic',
test_is_available_nic)
utils._update_dpdk_map('eth2_0', '0000:06:01.1', 'AA:02:03:04:05:FE',
'vfio-pci')
utils.update_sriov_vf_map('eth2', 0, 'eth2_0')
nics = utils.ordered_active_nics()
self.assertEqual(len(nics), 0)
shutil.rmtree(tmpdir)
shutil.rmtree(tmp_pci_dir)
def test_interface_mac_raises(self):
self.assertRaises(IOError, common.interface_mac, 'ens20f2p3')
def test_get_dpdk_devargs_mlnx(self):
def test_execute(name, dummy1, dummy2=None, dummy3=None):
if 'ethtool' in name:
out = _PCI_OUTPUT
return out, None
def test_get_stored_pci_address(ifname, noop):
return "0000:00:07.0"
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
self.stub_out('os_net_config.utils.get_stored_pci_address',
test_get_stored_pci_address)
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
nic = 'p4p1'
nic_path = os.path.join(tmpdir, nic)
os.makedirs(nic_path)
os.makedirs(os.path.join(nic_path, 'device'))
# Testing standard Mellanox Connect-X cards
with open(os.path.join(nic_path, 'operstate'), 'w') as f:
f.write('up')
with open(os.path.join(nic_path, 'address'), 'w') as f:
f.write('00:0f:21:69:39:14')
with open(os.path.join(nic_path, 'device', 'vendor'), 'w') as f:
f.write('0x15b3')
self.assertEqual(utils.get_dpdk_devargs(nic, False),
'0000:00:19.0')
# Testing VFs of Mellanox Connect-X cards
os.makedirs(os.path.join(nic_path, 'device', 'physfn'))
self.assertEqual(utils.get_dpdk_devargs(nic, False),
'0000:00:19.0')
# Check if devargs is derived, when the operstate is down
os.rmdir(os.path.join(nic_path, 'device', 'physfn'))
with open(os.path.join(nic_path, 'operstate'), 'w') as f:
f.write('down')
self.assertEqual(utils.get_dpdk_devargs(nic, False),
'0000:00:19.0')
# now testing the Mellanox CX3
with open(os.path.join(nic_path, 'device', 'device'), 'w') as f:
f.write('0x1007')
self.assertEqual(utils.get_dpdk_devargs(nic, False),
'class=eth,mac=00:0f:21:69:39:14')
with open(os.path.join(nic_path, 'device', 'vendor'), 'w') as f:
f.write('0x15b4')
self.assertEqual(utils.get_dpdk_devargs(nic, False),
'0000:00:07.0')
shutil.rmtree(tmpdir)
def test_is_active_nic_for_sriov_vf(self):
tmpdir = tempfile.mkdtemp()
self.stub_out('os_net_config.common.SYS_CLASS_NET', tmpdir)
# SR-IOV PF = ens802f0
# SR-IOV VF = enp129s2
for nic in ['ens802f0', 'enp129s2']:
nic_path = os.path.join(tmpdir, nic)
os.makedirs(nic_path)
os.makedirs(os.path.join(nic_path, 'device'))
with open(os.path.join(nic_path, 'operstate'), 'w') as f:
f.write('up')
with open(os.path.join(nic_path, 'address'), 'w') as f:
f.write('1.2.3.4')
nic_path = os.path.join(tmpdir, 'enp129s2', 'device', 'physfn')
os.makedirs(nic_path)
self.assertEqual(utils.is_active_nic('ens802f0'), True)
self.assertEqual(utils.is_active_nic('enp129s2'), False)
shutil.rmtree(tmpdir)
def test_get_vpp_interface(self):
def test_execute(name, *args, **kwargs):
if 'systemctl' in name:
return None, None
if 'vppctl' in name:
return _VPPCTL_OUTPUT, None
self.stub_out('oslo_concurrency.processutils.execute',
test_execute)
int_info = utils._get_vpp_interface('0000:00:09.0')
self.assertIsNotNone(int_info)
self.assertEqual('GigabitEthernet0/9/0', int_info['name'])
self.assertEqual('1', int_info['index'])
self.assertIsNone(utils._get_vpp_interface(None))
self.assertIsNone(utils._get_vpp_interface('0000:01:09.0'))
self.assertRaises(utils.VppException,
utils._get_vpp_interface, '0000:09.0')
@mock.patch('os_net_config.utils.processutils.execute',
return_value=('', None))
def test_get_vpp_interface_name_multiple_iterations(self, mock_execute):
self.assertIsNone(utils._get_vpp_interface('0000:00:09.0', 2, 1))
self.assertEqual(4, mock_execute.call_count)
def test_get_vpp_bond(self):
def test_execute(name, *args, **kwargs):
if 'systemctl' in name:
return None, None
if 'vppctl' in name:
return _VPPBOND_OUTPUT, None
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
bond_info = utils._get_vpp_bond(['1', '2'])
self.assertIsNotNone(bond_info)
self.assertEqual('BondEthernet0', bond_info['name'])
self.assertEqual('3', bond_info['index'])
self.assertIsNone(utils._get_vpp_bond(['1']))
self.assertIsNone(utils._get_vpp_bond(['1', '2', '3']))
self.assertIsNone(utils._get_vpp_bond([]))
def test_generate_vpp_config(self):
tmpdir = tempfile.mkdtemp()
config_path = os.path.join(tmpdir, 'startup.conf')
with open(config_path, 'w') as f:
f.write(_INITIAL_VPP_CONFIG)
vpp_exec_path = os.path.join(tmpdir, 'vpp-exec')
utils._VPP_EXEC_FILE = vpp_exec_path
int1 = objects.VppInterface('em1', options="vlan-strip-offload off")
int1.pci_dev = '0000:00:09.0'
int2 = objects.VppInterface('em2')
int2.pci_dev = '0000:00:09.1'
interfaces = [int1, int2]
bonds = []
expected_config = '''
unix {
exec %s
nodaemon
log /tmp/vpp.log
full-coredump
}
api-trace {
on
}
api-segment {
gid vpp
}
dpdk {
dev 0000:00:09.1
uio-driver vfio-pci
dev 0000:00:09.0 {vlan-strip-offload off}
}
''' % vpp_exec_path
self.assertEqual(expected_config,
utils.generate_vpp_config(config_path, interfaces,
bonds))
bonds = [objects.VppBond('net_bonding0', members=interfaces,
bonding_options='mode=2,xmit_policy=l3')]
expected_config = '''
unix {
exec %s
nodaemon
log /tmp/vpp.log
full-coredump
}
api-trace {
on
}
api-segment {
gid vpp
}
dpdk {
vdev net_bonding0,slave=0000:00:09.0,slave=0000:00:09.1,mode=2,xmit_policy=l3
dev 0000:00:09.1
uio-driver vfio-pci
dev 0000:00:09.0 {vlan-strip-offload off}
}
''' % vpp_exec_path
self.assertEqual(expected_config,
utils.generate_vpp_config(config_path, interfaces,
bonds))
def test_update_vpp_mapping(self):
tmpdir = tempfile.mkdtemp()
vpp_exec_path = os.path.join(tmpdir, 'vpp-exec')
utils._VPP_EXEC_FILE = vpp_exec_path
def test_get_dpdk_map():
return [{'name': 'eth1', 'pci_address': '0000:00:09.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'}]
self.stub_out('os_net_config.common.get_dpdk_map', test_get_dpdk_map)
def test_execute(name, *args, **kwargs):
return None, None
self.stub_out('oslo_concurrency.processutils.execute', test_execute)
def test_get_vpp_interface(pci_dev, tries, timeout):
return {'name': 'GigabitEthernet0/9/0', 'index': '1'}
self.stub_out('os_net_config.utils._get_vpp_interface',
test_get_vpp_interface)
int1 = objects.VppInterface('eth1', options="vlan-strip-offload off")
int1.pci_dev = '0000:00:09.0'
int1.hwaddr = '01:02:03:04:05:06'
int2 = objects.VppInterface('eth2')
int2.pci_dev = '0000:00:09.1'
int2.hwaddr = '01:02:03:04:05:07'
interfaces = [int1, int2]
utils.update_vpp_mapping(interfaces, [])
contents = common.get_file_data(common.DPDK_MAPPING_FILE)
dpdk_test = [{'name': 'eth1', 'pci_address': '0000:00:09.0',
'mac_address': '01:02:03:04:05:06',
'driver': 'vfio-pci'},
{'name': 'eth2', 'pci_address': '0000:00:09.1',
'mac_address': '01:02:03:04:05:07',
'driver': 'vfio-pci'}]
dpdk_map = yaml.safe_load(contents) if contents else []
self.assertEqual(2, len(dpdk_map))
self.assertListEqual(dpdk_test, dpdk_map)
View Git Blame Copy Permalink