# Path: os_vif/internal/ip/api.py
# LOG = logging.getLogger(__name__)
# 
# Path: os_vif/tests/functional/base.py
# CONF = cfg.CONF
# LOG = logging.getLogger(__name__)
# DEFAULT_LOG_DIR = os.path.join(_get_test_log_path(), 'osvif-functional-logs')
#     COMPONENT_NAME = 'os_vif'
#     PRIVILEGED_GROUP = 'os_vif_privileged'
# def _get_test_log_path():
# def wait_until_true(predicate, timeout=15, sleep=1):
#     def __init__(cls, name, bases, dct):
#     def _catch_timeout(f):
#         def func(self, *args, **kwargs):
# def setup_logging(component_name):
# def sanitize_log_path(path):
#     def setUp(self):
#     def flags(self, **kw):
# class _CatchTimeoutMetaclass(abc.ABCMeta):
# class BaseFunctionalTestCase(base.BaseTestCase,
#                              metaclass=_CatchTimeoutMetaclass):
# 
# Path: os_vif/tests/functional/privsep.py



import os
import re
from oslo_concurrency import processutils
from oslo_utils import excutils
from os_vif.internal.ip.api import ip as ip_lib
from os_vif.tests.functional import base
from os_vif.tests.functional import privsep
    def test_iproute_object_closes_correctly(self):
        # NOTE(ralonsoh): check https://bugs.launchpad.net/os-vif/+bug/1807949
        device = "test_dev_9"
        link = "test_devlink_2"
        self.add_device(link, 'dummy')
        self.addCleanup(self.del_device, device)
        self.addCleanup(self.del_device, link)
        for _ in range(300):
            _ip_cmd_add(device, 'vlan', link=link, vlan_id=100)
            _ip_cmd_delete(device)

    def test_exists(self):
        device = "test_dev_10"
        self.addCleanup(self.del_device, device)
        self.add_device(device, 'dummy')
        self.assertTrue(_ip_cmd_exists(device))
        self.del_device(device)
        self.assertFalse(_ip_cmd_exists(device))

    def test_add_bridge(self):
        device = "test_dev_11"
        self.addCleanup(self.del_device, device)
        _ip_cmd_add(device, 'bridge')
        self.assertTrue(self.exist_device(device))
        base_path = "/sys/class/net/test_dev_11/bridge/%s"
        with open(base_path % "forward_delay", "r") as f:
            self.assertEqual("0", f.readline().rstrip('\n'))
        with open(base_path % "stp_state", "r") as f:
            self.assertEqual("0", f.readline().rstrip('\n'))
        with open(base_path % "multicast_snooping", "r") as f:

# Path: os_vif/internal/ip/api.py
# LOG = logging.getLogger(__name__)
# 
# Path: os_vif/tests/functional/base.py
# CONF = cfg.CONF
# LOG = logging.getLogger(__name__)
# DEFAULT_LOG_DIR = os.path.join(_get_test_log_path(), 'osvif-functional-logs')
#     COMPONENT_NAME = 'os_vif'
#     PRIVILEGED_GROUP = 'os_vif_privileged'
# def _get_test_log_path():
# def wait_until_true(predicate, timeout=15, sleep=1):
#     def __init__(cls, name, bases, dct):
#     def _catch_timeout(f):
#         def func(self, *args, **kwargs):
# def setup_logging(component_name):
# def sanitize_log_path(path):
#     def setUp(self):
#     def flags(self, **kw):
# class _CatchTimeoutMetaclass(abc.ABCMeta):
# class BaseFunctionalTestCase(base.BaseTestCase,
#                              metaclass=_CatchTimeoutMetaclass):
# 
# Path: os_vif/tests/functional/privsep.py



# Path: os_vif/tests/functional/internal/command/ip/test_impl_pyroute2.py
import os
import re
from oslo_concurrency import processutils
from oslo_utils import excutils
from os_vif.internal.ip.api import ip as ip_lib
from os_vif.tests.functional import base
from os_vif.tests.functional import privsep

    def test_iproute_object_closes_correctly(self):
        # NOTE(ralonsoh): check https://bugs.launchpad.net/os-vif/+bug/1807949
        device = "test_dev_9"
        link = "test_devlink_2"
        self.add_device(link, 'dummy')
        self.addCleanup(self.del_device, device)
        self.addCleanup(self.del_device, link)
        for _ in range(300):
            _ip_cmd_add(device, 'vlan', link=link, vlan_id=100)
            _ip_cmd_delete(device)

    def test_exists(self):
        device = "test_dev_10"
        self.addCleanup(self.del_device, device)
        self.add_device(device, 'dummy')
        self.assertTrue(_ip_cmd_exists(device))
        self.del_device(device)
        self.assertFalse(_ip_cmd_exists(device))

    def test_add_bridge(self):
        device = "test_dev_11"
        self.addCleanup(self.del_device, device)
        _ip_cmd_add(device, 'bridge')
        self.assertTrue(self.exist_device(device))
        base_path = "/sys/class/net/test_dev_11/bridge/%s"
        with open(base_path % "forward_delay", "r") as f:
            self.assertEqual("0", f.readline().rstrip('\n'))
        with open(base_path % "stp_state", "r") as f:
            self.assertEqual("0", f.readline().rstrip('\n'))
        with open(base_path % "multicast_snooping", "r") as f:

            self.assertEqual("0", f.readline().rstrip('\n'))