# Path: vif_plug_ovs/constants.py
# PLUGIN_NAME = 'ovs'
# OVS_VHOSTUSER_INTERFACE_TYPE = 'dpdkvhostuser'
# OVS_VHOSTUSER_CLIENT_INTERFACE_TYPE = 'dpdkvhostuserclient'
# OVS_VHOSTUSER_PREFIX = 'vhu'
# OVS_DATAPATH_SYSTEM = 'system'
# OVS_DATAPATH_NETDEV = 'netdev'
# PLATFORM_LINUX = 'linux2'
# PLATFORM_WIN32 = 'win32'
# OVS_DPDK_INTERFACE_TYPE = 'dpdk'
# DEAD_VLAN = 4095
# 
# Path: vif_plug_ovs/linux_net.py
# LOG = logging.getLogger(__name__)
# VIRTFN_RE = re.compile(r"virtfn(\d+)")
# INT_RE = re.compile(r"^(\d+)$")
# VF_RE = re.compile(r"vf(\d+)", re.IGNORECASE)
# PF_RE = re.compile(r"pf(\d+)", re.IGNORECASE)
# PF_FUNC_RE = re.compile(r"\.(\d+)", 0)
# UPLINK_PORT_RE = re.compile(r"p(\d+)", re.IGNORECASE)
# _SRIOV_TOTALVFS = "sriov_totalvfs"
# NIC_NAME_LEN = 14
# def _update_device_mtu(dev, mtu):
# def delete_net_dev(dev):
# def create_veth_pair(dev1_name, dev2_name, mtu):
# def update_veth_pair(dev1_name, dev2_name, mtu):
# def _disable_ipv6(bridge):
# def _arp_filtering(bridge):
# def ensure_bridge(bridge):
# def delete_bridge(bridge, dev):
# def add_bridge_port(bridge, dev):
# def set_device_mtu(dev, mtu):
# def set_interface_state(interface_name, port_state):
# def _parse_vf_number(phys_port_name):
# def _parse_pf_number(phys_port_name):
# def get_function_by_ifname(ifname):
# def _get_pf_func(pf_ifname):
# def get_representor_port(pf_ifname, vf_num):
# def _get_sysfs_netdev_path(pci_addr, pf_interface):
# def _is_switchdev(netdev):
# def get_ifname_by_pci_address(pci_addr, pf_interface=False, switchdev=False):
# def get_vf_num_by_pci_address(pci_addr):
# def get_dpdk_representor_port_name(port_id):
# def get_pf_pci_from_vf(vf_pci):
# def _get_phys_port_name(ifname):
# def _get_phys_switch_id(ifname):
# 
# Path: vif_plug_ovs/ovsdb/ovsdb_lib.py
# LOG = logging.getLogger(__name__)
# class BaseOVS(object):
#     def __init__(self, config):
#     def ovsdb(self):
#     def _ovs_supports_mtu_requests(self):
#     def _set_mtu_request(self, dev, mtu):
#     def update_device_mtu(self, dev, mtu, interface_type=None):
#     def ensure_ovs_bridge(self, bridge, datapath_type):
#     def delete_ovs_bridge(self, bridge):
#     def create_patch_port_pair(
#         self, port_bridge, port_bridge_port, int_bridge, int_bridge_port,
#         iface_id, mac, instance_id, tag=None
#     ):
#     def create_ovs_vif_port(
#         self, bridge, dev, iface_id, mac, instance_id,
#         mtu=None, interface_type=None, vhost_server_path=None,
#         tag=None, pf_pci=None, vf_num=None, set_ids=True
#     ):
#     def update_ovs_vif_port(self, dev, mtu=None, interface_type=None):
#     def delete_ovs_vif_port(self, bridge, dev, delete_netdev=True):


from unittest import mock
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import uuidutils
from vif_plug_ovs import constants
from vif_plug_ovs import linux_net
from vif_plug_ovs.ovsdb import ovsdb_lib
import testtools
        CONF.register_group(test_vif_plug_ovs_group)
        CONF.register_opt(cfg.IntOpt('ovs_vsctl_timeout', default=1500),
                          test_vif_plug_ovs_group)
        CONF.register_opt(cfg.StrOpt('ovsdb_interface', default='vsctl'),
                          test_vif_plug_ovs_group)
        CONF.register_opt(cfg.StrOpt('ovsdb_connection', default=None),
                          test_vif_plug_ovs_group)
        self.br = ovsdb_lib.BaseOVS(cfg.CONF.test_vif_plug_ovs)
        self.mock_db_set = mock.patch.object(self.br.ovsdb, 'db_set').start()
        self.mock_del_port = mock.patch.object(self.br.ovsdb,
                                               'del_port').start()
        self.mock_add_port = mock.patch.object(self.br.ovsdb,
                                               'add_port').start()
        self.mock_add_br = mock.patch.object(self.br.ovsdb, 'add_br').start()
        self.mock_transaction = mock.patch.object(self.br.ovsdb,
                                                  'transaction').start()

    def test__set_mtu_request(self):
        self.br._set_mtu_request('device', 1500)
        calls = [mock.call('Interface', 'device', ('mtu_request', 1500))]
        self.mock_db_set.assert_has_calls(calls)

    @mock.patch('sys.platform', constants.PLATFORM_LINUX)
    @mock.patch.object(linux_net, 'set_device_mtu')
    def test__update_device_mtu_interface_not_vhostuser_linux(self,
            mock_set_device_mtu):
        self.br.update_device_mtu('device', 1500, 'not_vhost')
        mock_set_device_mtu.assert_has_calls([mock.call('device', 1500)])

    @mock.patch('sys.platform', constants.PLATFORM_WIN32)

# Path: vif_plug_ovs/constants.py
# PLUGIN_NAME = 'ovs'
# OVS_VHOSTUSER_INTERFACE_TYPE = 'dpdkvhostuser'
# OVS_VHOSTUSER_CLIENT_INTERFACE_TYPE = 'dpdkvhostuserclient'
# OVS_VHOSTUSER_PREFIX = 'vhu'
# OVS_DATAPATH_SYSTEM = 'system'
# OVS_DATAPATH_NETDEV = 'netdev'
# PLATFORM_LINUX = 'linux2'
# PLATFORM_WIN32 = 'win32'
# OVS_DPDK_INTERFACE_TYPE = 'dpdk'
# DEAD_VLAN = 4095
# 
# Path: vif_plug_ovs/linux_net.py
# LOG = logging.getLogger(__name__)
# VIRTFN_RE = re.compile(r"virtfn(\d+)")
# INT_RE = re.compile(r"^(\d+)$")
# VF_RE = re.compile(r"vf(\d+)", re.IGNORECASE)
# PF_RE = re.compile(r"pf(\d+)", re.IGNORECASE)
# PF_FUNC_RE = re.compile(r"\.(\d+)", 0)
# UPLINK_PORT_RE = re.compile(r"p(\d+)", re.IGNORECASE)
# _SRIOV_TOTALVFS = "sriov_totalvfs"
# NIC_NAME_LEN = 14
# def _update_device_mtu(dev, mtu):
# def delete_net_dev(dev):
# def create_veth_pair(dev1_name, dev2_name, mtu):
# def update_veth_pair(dev1_name, dev2_name, mtu):
# def _disable_ipv6(bridge):
# def _arp_filtering(bridge):
# def ensure_bridge(bridge):
# def delete_bridge(bridge, dev):
# def add_bridge_port(bridge, dev):
# def set_device_mtu(dev, mtu):
# def set_interface_state(interface_name, port_state):
# def _parse_vf_number(phys_port_name):
# def _parse_pf_number(phys_port_name):
# def get_function_by_ifname(ifname):
# def _get_pf_func(pf_ifname):
# def get_representor_port(pf_ifname, vf_num):
# def _get_sysfs_netdev_path(pci_addr, pf_interface):
# def _is_switchdev(netdev):
# def get_ifname_by_pci_address(pci_addr, pf_interface=False, switchdev=False):
# def get_vf_num_by_pci_address(pci_addr):
# def get_dpdk_representor_port_name(port_id):
# def get_pf_pci_from_vf(vf_pci):
# def _get_phys_port_name(ifname):
# def _get_phys_switch_id(ifname):
# 
# Path: vif_plug_ovs/ovsdb/ovsdb_lib.py
# LOG = logging.getLogger(__name__)
# class BaseOVS(object):
#     def __init__(self, config):
#     def ovsdb(self):
#     def _ovs_supports_mtu_requests(self):
#     def _set_mtu_request(self, dev, mtu):
#     def update_device_mtu(self, dev, mtu, interface_type=None):
#     def ensure_ovs_bridge(self, bridge, datapath_type):
#     def delete_ovs_bridge(self, bridge):
#     def create_patch_port_pair(
#         self, port_bridge, port_bridge_port, int_bridge, int_bridge_port,
#         iface_id, mac, instance_id, tag=None
#     ):
#     def create_ovs_vif_port(
#         self, bridge, dev, iface_id, mac, instance_id,
#         mtu=None, interface_type=None, vhost_server_path=None,
#         tag=None, pf_pci=None, vf_num=None, set_ids=True
#     ):
#     def update_ovs_vif_port(self, dev, mtu=None, interface_type=None):
#     def delete_ovs_vif_port(self, bridge, dev, delete_netdev=True):


# Path: vif_plug_ovs/tests/unit/ovsdb/test_ovsdb_lib.py
from unittest import mock
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import uuidutils
from vif_plug_ovs import constants
from vif_plug_ovs import linux_net
from vif_plug_ovs.ovsdb import ovsdb_lib
import testtools

        CONF.register_group(test_vif_plug_ovs_group)
        CONF.register_opt(cfg.IntOpt('ovs_vsctl_timeout', default=1500),
                          test_vif_plug_ovs_group)
        CONF.register_opt(cfg.StrOpt('ovsdb_interface', default='vsctl'),
                          test_vif_plug_ovs_group)
        CONF.register_opt(cfg.StrOpt('ovsdb_connection', default=None),
                          test_vif_plug_ovs_group)
        self.br = ovsdb_lib.BaseOVS(cfg.CONF.test_vif_plug_ovs)
        self.mock_db_set = mock.patch.object(self.br.ovsdb, 'db_set').start()
        self.mock_del_port = mock.patch.object(self.br.ovsdb,
                                               'del_port').start()
        self.mock_add_port = mock.patch.object(self.br.ovsdb,
                                               'add_port').start()
        self.mock_add_br = mock.patch.object(self.br.ovsdb, 'add_br').start()
        self.mock_transaction = mock.patch.object(self.br.ovsdb,
                                                  'transaction').start()

    def test__set_mtu_request(self):
        self.br._set_mtu_request('device', 1500)
        calls = [mock.call('Interface', 'device', ('mtu_request', 1500))]
        self.mock_db_set.assert_has_calls(calls)

    @mock.patch('sys.platform', constants.PLATFORM_LINUX)
    @mock.patch.object(linux_net, 'set_device_mtu')
    def test__update_device_mtu_interface_not_vhostuser_linux(self,
            mock_set_device_mtu):
        self.br.update_device_mtu('device', 1500, 'not_vhost')
        mock_set_device_mtu.assert_has_calls([mock.call('device', 1500)])

    @mock.patch('sys.platform', constants.PLATFORM_WIN32)

    @mock.patch.object(linux_net, 'set_device_mtu')