<commit_before># -*- coding: utf-8 -*-

import os
import json
import time
import socket
import subprocess
from pyroute2 import config
from pyroute2 import IPDB
from pyroute2 import IPRoute
from pyroute2.common import basestring
from pyroute2.common import AddrPool
from pyroute2.netlink import NetlinkError
from pyroute2.ipdb.common import CreateException
from utils import grep
from utils import create_link
from utils import remove_link
from utils import require_user
from utils import require_executable
from utils import require_8021q
from utils import require_bond
from utils import require_bridge
from utils import get_ip_addr


class _TestException(Exception):
    pass


class TestRace(object):

    def test_dummy0_unloaded(object):
        require_user('root')
        # firstly unload the dummy module
        with open(os.devnull, 'w') as fnull:
            subprocess.call(['modprobe', '-r', 'dummy'],
                            stdout=fnull,
                            stderr=fnull)
        ip = None
        try:
            # now create the dummy0 -- it will cause the
            # module autoload
            ip = IPDB()
            # that must succeed
            ip.create(ifname='dummy0', kind='dummy').commit()
            # just in case: the second attempt must fail on the
            # create() stage, even w/o any commit()
            try:
                ip.create(ifname='dummy0', kind='dummy')
            except CreateException:
                pass
        except Exception:
            raise
        finally:
            if ip is not None:
                ip.release()

    def test_dummy0_loaded(object):
        require_user('root')
        # assert the module is loaded
        create_link('dummyX', 'dummy')
        ip = None
        try:
            # try to create and fail in create()
            ip = IPDB()
            ip.create(ifname='dummy0', kind='dummy')
        except CreateException:
            pass
        except Exception:
            raise
        finally:
            if ip is not None:
                ip.release()


class TestExplicit(object):
    ip = None
    mode = 'explicit'

    def setup(self):
        self.ap = AddrPool()
        self.iftmp = 'pr2x{0}'
        self.ifaces = []
        self.ifd = self.get_ifname()
        create_link(self.ifd, kind='dummy')
        self.ip = IPDB(mode=self.mode)

    def get_ifname(self):
        naddr = self.ap.alloc()
        ifname = self.iftmp.format(naddr)
        self.ifaces.append(ifname)
        return ifname

    def teardown(self):
        for name in self.ifaces:
            try:
                with self.ip.interfaces[name] as i:
                    i.remove()
            except:
                pass
        self.ip.release()
        self.ifaces = []

    def test_simple(self):
        assert len(list(self.ip.interfaces.keys())) > 0

    def test_empty_transaction(self):
        assert 'lo' in self.ip.interfaces
        with self.ip.interfaces.lo as i:
            assert isinstance(i.mtu, int)

    def test_idx_len(self):
        assert len(self.ip.by_name.keys()) == len(self.ip.by_index.keys())

    def test_idx_set(self):
        assert set(self.ip.by_name.values()) == set(self.ip.by_index.values())

    def test_idx_types(self):
        assert all(isinstance(i, int) for i in self.ip.by_index.keys())
        assert all(isinstance(i, basestring) for i in self.ip.by_name.keys())

    def test_addr_attributes(self):
        require_user('root')

        if1 = self.get_ifname()
        if2 = self.get_ifname()

        with self.ip.create(ifname=if1, kind='dummy') as i:
            # +scope host (=> broadcast == None)
            i.add_ip('172.16.102.1/24', scope=254)

        with self.ip.create(ifname=if2, kind='dummy') as i:
            # +broadcast (default scope == 0)
            i.add_ip('172.16.103.1/24', broadcast='172.16.103.128')

        index = self.ip.interfaces[if1]['index']
        addr = self.ip.nl.get_addr(index=index)[0]
        assert addr['scope'] == 254
        assert addr.get_attr('IFA_BROADCAST') == None

        index = self.ip.interfaces[if2]['index']
        addr = self.ip.nl.get_addr(index=index)[0]
        assert addr['scope'] == 0
        assert addr.get_attr('IFA_BROADCAST') == '172.16.103.128'

    def test_addr_loaded(self):
        for name in self.ip.by_name:
            assert len(self.ip.interfaces[name]['ipaddr']) == \
                len(get_ip_addr(name))

    def test_reprs(self):
        assert isinstance(repr(self.ip.interfaces.lo.ipaddr), basestring)
        assert isinstance(repr(self.ip.interfaces.lo), basestring)

    def test_dotkeys(self):
        # self.ip.lo hint for ipython
        assert 'lo' in dir(self.ip.interfaces)
        assert 'lo' in self.ip.interfaces
        assert self.ip.interfaces.lo == self.ip.interfaces['lo']
        # create attribute
        self.ip.interfaces['newitem'] = True
        self.ip.interfaces.newattr = True
        self.ip.interfaces.newitem = None
        assert self.ip.interfaces.newitem == self.ip.interfaces['newitem']
        assert self.ip.interfaces.newitem is None
        # delete attribute
        del self.ip.interfaces.newitem
        del self.ip.interfaces.newattr
        assert 'newattr' not in dir(self.ip.interfaces)<commit_msg>tests: pep8 fix<commit_after># -*- coding: utf-8 -*-

import os
import json
import time
import socket
import subprocess
from pyroute2 import config
from pyroute2 import IPDB
from pyroute2 import IPRoute
from pyroute2.common import basestring
from pyroute2.common import AddrPool
from pyroute2.netlink import NetlinkError
from pyroute2.ipdb.common import CreateException
from utils import grep
from utils import create_link
from utils import remove_link
from utils import require_user
from utils import require_executable
from utils import require_8021q
from utils import require_bond
from utils import require_bridge
from utils import get_ip_addr


class _TestException(Exception):
    pass


class TestRace(object):

    def test_dummy0_unloaded(object):
        require_user('root')
        # firstly unload the dummy module
        with open(os.devnull, 'w') as fnull:
            subprocess.call(['modprobe', '-r', 'dummy'],
                            stdout=fnull,
                            stderr=fnull)
        ip = None
        try:
            # now create the dummy0 -- it will cause the
            # module autoload
            ip = IPDB()
            # that must succeed
            ip.create(ifname='dummy0', kind='dummy').commit()
            # just in case: the second attempt must fail on the
            # create() stage, even w/o any commit()
            try:
                ip.create(ifname='dummy0', kind='dummy')
            except CreateException:
                pass
        except Exception:
            raise
        finally:
            if ip is not None:
                ip.release()

    def test_dummy0_loaded(object):
        require_user('root')
        # assert the module is loaded
        create_link('dummyX', 'dummy')
        ip = None
        try:
            # try to create and fail in create()
            ip = IPDB()
            ip.create(ifname='dummy0', kind='dummy')
        except CreateException:
            pass
        except Exception:
            raise
        finally:
            if ip is not None:
                ip.release()


class TestExplicit(object):
    ip = None
    mode = 'explicit'

    def setup(self):
        self.ap = AddrPool()
        self.iftmp = 'pr2x{0}'
        self.ifaces = []
        self.ifd = self.get_ifname()
        create_link(self.ifd, kind='dummy')
        self.ip = IPDB(mode=self.mode)

    def get_ifname(self):
        naddr = self.ap.alloc()
        ifname = self.iftmp.format(naddr)
        self.ifaces.append(ifname)
        return ifname

    def teardown(self):
        for name in self.ifaces:
            try:
                with self.ip.interfaces[name] as i:
                    i.remove()
            except:
                pass
        self.ip.release()
        self.ifaces = []

    def test_simple(self):
        assert len(list(self.ip.interfaces.keys())) > 0

    def test_empty_transaction(self):
        assert 'lo' in self.ip.interfaces
        with self.ip.interfaces.lo as i:
            assert isinstance(i.mtu, int)

    def test_idx_len(self):
        assert len(self.ip.by_name.keys()) == len(self.ip.by_index.keys())

    def test_idx_set(self):
        assert set(self.ip.by_name.values()) == set(self.ip.by_index.values())

    def test_idx_types(self):
        assert all(isinstance(i, int) for i in self.ip.by_index.keys())
        assert all(isinstance(i, basestring) for i in self.ip.by_name.keys())

    def test_addr_attributes(self):
        require_user('root')

        if1 = self.get_ifname()
        if2 = self.get_ifname()

        with self.ip.create(ifname=if1, kind='dummy') as i:
            # +scope host (=> broadcast == None)
            i.add_ip('172.16.102.1/24', scope=254)

        with self.ip.create(ifname=if2, kind='dummy') as i:
            # +broadcast (default scope == 0)
            i.add_ip('172.16.103.1/24', broadcast='172.16.103.128')

        index = self.ip.interfaces[if1]['index']
        addr = self.ip.nl.get_addr(index=index)[0]
        assert addr['scope'] == 254
        assert addr.get_attr('IFA_BROADCAST') is None

        index = self.ip.interfaces[if2]['index']
        addr = self.ip.nl.get_addr(index=index)[0]
        assert addr['scope'] == 0
        assert addr.get_attr('IFA_BROADCAST') == '172.16.103.128'

    def test_addr_loaded(self):
        for name in self.ip.by_name:
            assert len(self.ip.interfaces[name]['ipaddr']) == \
                len(get_ip_addr(name))

    def test_reprs(self):
        assert isinstance(repr(self.ip.interfaces.lo.ipaddr), basestring)
        assert isinstance(repr(self.ip.interfaces.lo), basestring)

    def test_dotkeys(self):
        # self.ip.lo hint for ipython
        assert 'lo' in dir(self.ip.interfaces)
        assert 'lo' in self.ip.interfaces
        assert self.ip.interfaces.lo == self.ip.interfaces['lo']
        # create attribute
        self.ip.interfaces['newitem'] = True
        self.ip.interfaces.newattr = True
        self.ip.interfaces.newitem = None
        assert self.ip.interfaces.newitem == self.ip.interfaces['newitem']
        assert self.ip.interfaces.newitem is None
        # delete attribute
        del self.ip.interfaces.newitem
        del self.ip.interfaces.newattr
        assert 'newattr' not in dir(self.ip.interfaces)
{'fraction_non_alphanumeric': 0.10822306238185256, 'fraction_numerical': 0.015406427221172022, 'mean_word_length': 2.1002050981541167, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7615935', 'n_tokens_mistral': 3330, 'n_tokens_neox': 3202, 'n_words': 774}