<commit_before>import errno
import socket
from utils import require_user
from utils import get_simple_bpf_program
from utils import skip_if_not_supported
from pyroute2 import IPRoute
from pyroute2 import protocols
from pyroute2.common import uifname
from pyroute2.netlink import NetlinkError
from pyroute2.netlink.rtnl import RTM_NEWQDISC
from pyroute2.netlink.rtnl import RTM_NEWTFILTER
from pyroute2.netlink.rtnl import RTM_NEWTCLASS
from pyroute2.netlink.rtnl import TC_H_INGRESS
from nose.plugins.skip import SkipTest


def try_qd(qd, call, *argv, **kwarg):
    try:
        call(*argv, **kwarg)
    except NetlinkError as e:
        # code 2 'no such file or directory)
        if e.code == errno.ENOENT:
            raise SkipTest('missing traffic control <%s>' % (qd))
        raise


class BasicTest(object):

    def setup(self):
        require_user('root')
        self.ip = IPRoute()
        self.ifname = uifname()
        self.ip.link_create(ifname=self.ifname, kind='dummy')
        self.interface = self.ip.link_lookup(ifname=self.ifname)[0]

    def teardown(self):
        self.ip.link('delete', index=self.interface)
        self.ip.close()

    def get_qdiscs(self):
        return [x for x in self.ip.get_qdiscs() if
                x['index'] == self.interface]

    def get_qdisc(self):
        # get qdiscs list and filter out our interface
        qds = self.get_qdiscs()
        if qds:
            return qds[0]
        else:
            return None


class TestIngress(BasicTest):

    def test_simple(self):
        self.ip.tc(RTM_NEWQDISC, 'ingress', self.interface, 0xffff0000)
        qds = self.get_qdisc()
        # assert the list is not empty
        assert qds
        # assert there is the ingress queue
        assert qds.get_attr('TCA_KIND') == 'ingress'
        # assert it has proper handle and parent
        assert qds['handle'] == 0xffff0000
        assert qds['parent'] == TC_H_INGRESS

    def test_filter(self):
        self.test_simple()
        self.ip.tc(RTM_NEWTFILTER, 'u32', self.interface, 0,
                   protocol=socket.AF_INET,
                   parent=0xffff0000,
                   action='drop',
                   target=0x1,
                   rate='10kbit',
                   burst=10240,
                   limit=0,
                   prio=50,
                   keys=['<KEY>'])
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        # assert there are filters
        assert fls
        # assert there is one police rule:
        prs = [x for x in fls
               if x.get_attr('TCA_OPTIONS') is not None and
               (x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_POLICE')
                is not None or
                x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_ACT')
                is not None)][0]
        # assert the police rule has specified parameters
        options = prs.get_attr('TCA_OPTIONS')
        police_u32 = options.get_attr('TCA_U32_POLICE')
        # on modern kernels there is no TCA_U32_POLICE under
        # TCA_OPTIONS, but there is TCA_U32_ACT
        if police_u32 is None:
            police_u32 = options.get_attr('TCA_U32_ACT').\
                get_attr('TCA_ACT_PRIO_0').\
                get_attr('TCA_ACT_OPTIONS')
        police_tbf = police_u32.get_attr('TCA_POLICE_TBF')
        assert police_tbf['rate'] == 1250
        assert police_tbf['mtu'] == 2040

    @skip_if_not_supported
    def test_bpf_filter(self):
        self.test_simple()
        fd1 = get_simple_bpf_program('sched_cls')
        fd2 = get_simple_bpf_program('sched_act')
        if fd1 == -1 or fd2 == -1:
            # to get bpf filter working, one should have:
            # kernel >= 4.1
            # CONFIG_EXPERT=y
            # CONFIG_BPF_SYSCALL=y
            # CONFIG_NET_CLS_BPF=m/y
            #
            # see `grep -rn BPF_PROG_TYPE_SCHED_CLS kernel_sources`
            raise SkipTest('bpf syscall error')
        self.ip.tc(RTM_NEWTFILTER, 'bpf', self.interface, 0,
                   fd=fd1, name='my_func', parent=0xffff0000,
                   action='ok', classid=1)
        action = {'kind': 'bpf', 'fd': fd2, 'name': 'my_func', 'action': 'ok'}
        self.ip.tc(RTM_NEWTFILTER, 'u32', self.interface, 1,
                   protocol=protocols.ETH_P_ALL, parent=0xffff0000,
                   target=0x10002, keys=['<KEY>'], action=action)
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        assert fls
        bpf_filter = [x for x in fls
                      if x.get_attr('TCA_OPTIONS') is not None and
                      (x.get_attr('TCA_OPTIONS').get_attr('TCA_BPF_ACT')
                       is not None)][0]
        bpf_options = bpf_filter.get_attr('TCA_OPTIONS')
        gact_parms = bpf_options.get_attr('TCA_BPF_ACT').\
            get_attr('TCA_ACT_PRIO_1').\
            get_attr('TCA_ACT_OPTIONS').\
            get_attr('TCA_GACT_PARMS')
        assert gact_parms['action'] == 0

        u32_filter = [x for x in fls
                      if x.get_attr('TCA_OPTIONS') is not None and
                      (x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_ACT')
                       is not None)][0]
        u32_options = u32_filter.get_attr('TCA_OPTIONS')
        bpf_fd = u32_options.get_attr('TCA_U32_ACT').\
            get_attr('TCA_ACT_PRIO_1').\
            get_attr('TCA_ACT_OPTIONS').\
            get_attr('TCA_ACT_BPF_FD')
        assert bpf_fd == fd2

    @skip_if_not_supported
    def test_bpf_filter_policer(self):
        self.test_simple()
        fd = get_simple_bpf_program('sched_cls')
        if fd == -1:
            # see comment above about kernel requirements
            raise SkipTest('bpf syscall error')
        self.ip.tc(RTM_NEWTFILTER, 'bpf', self.interface, 0,
                   fd=fd, name='my_func', parent=0xffff0000,
                   action='ok', classid=1, rate='10kbit',
                   burst=10240, mtu=2040)
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        # assert the supplied policer is returned to us intact
        plcs = [x for x in fls
                if x.get_attr('TCA_OPTIONS') is not None and
                (x.get_attr('TCA_OPTIONS').get_attr('TCA_BPF_POLICE')
                 is not None)][0]
        options = plcs.get_attr('TCA_OPTIONS')
        police = options.get_attr('TCA_BPF_POLICE').\
            get_attr('TCA_POLICE_TBF')
        assert police['rate'] == 1250
        assert police['mtu'] == 2040


class TestPfifo(BasicTest):

    def test_pfifo(self):
        try_qd('pfifo_fast', self.ip.tc,
               RTM_NEWQDISC, 'pfifo_fast', self.interface, 0)
        qds = self.get_qdisc()
        assert qds
        assert qds.get_attr('TCA_KIND') == 'pfifo_fast'
        assert isinstance(qds.get_attr('TCA_OPTIONS')['priomap'], tuple)


class TestPlug(BasicTest):

    def test_plug(self):
        try_qd('plug', self.ip.tc,
               RTM_NEWQDISC, 'plug', self.interface, limit=13107)<commit_msg>tests: codel qdisc<commit_after>import errno
import socket
from utils import require_user
from utils import get_simple_bpf_program
from utils import skip_if_not_supported
from pyroute2 import IPRoute
from pyroute2 import protocols
from pyroute2.common import uifname
from pyroute2.netlink import NetlinkError
from pyroute2.netlink.rtnl import RTM_NEWQDISC
from pyroute2.netlink.rtnl import RTM_NEWTFILTER
from pyroute2.netlink.rtnl import RTM_NEWTCLASS
from pyroute2.netlink.rtnl import TC_H_INGRESS
from nose.plugins.skip import SkipTest


def try_qd(qd, call, *argv, **kwarg):
    try:
        call(*argv, **kwarg)
    except NetlinkError as e:
        # code 2 'no such file or directory)
        if e.code == errno.ENOENT:
            raise SkipTest('missing traffic control <%s>' % (qd))
        raise


class BasicTest(object):

    def setup(self):
        require_user('root')
        self.ip = IPRoute()
        self.ifname = uifname()
        self.ip.link_create(ifname=self.ifname, kind='dummy')
        self.interface = self.ip.link_lookup(ifname=self.ifname)[0]

    def teardown(self):
        self.ip.link('delete', index=self.interface)
        self.ip.close()

    def get_qdiscs(self):
        return [x for x in self.ip.get_qdiscs() if
                x['index'] == self.interface]

    def get_qdisc(self):
        # get qdiscs list and filter out our interface
        qds = self.get_qdiscs()
        if qds:
            return qds[0]
        else:
            return None


class TestIngress(BasicTest):

    def test_simple(self):
        self.ip.tc(RTM_NEWQDISC, 'ingress', self.interface, 0xffff0000)
        qds = self.get_qdisc()
        # assert the list is not empty
        assert qds
        # assert there is the ingress queue
        assert qds.get_attr('TCA_KIND') == 'ingress'
        # assert it has proper handle and parent
        assert qds['handle'] == 0xffff0000
        assert qds['parent'] == TC_H_INGRESS

    def test_filter(self):
        self.test_simple()
        self.ip.tc(RTM_NEWTFILTER, 'u32', self.interface, 0,
                   protocol=socket.AF_INET,
                   parent=0xffff0000,
                   action='drop',
                   target=0x1,
                   rate='10kbit',
                   burst=10240,
                   limit=0,
                   prio=50,
                   keys=['<KEY>'])
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        # assert there are filters
        assert fls
        # assert there is one police rule:
        prs = [x for x in fls
               if x.get_attr('TCA_OPTIONS') is not None and
               (x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_POLICE')
                is not None or
                x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_ACT')
                is not None)][0]
        # assert the police rule has specified parameters
        options = prs.get_attr('TCA_OPTIONS')
        police_u32 = options.get_attr('TCA_U32_POLICE')
        # on modern kernels there is no TCA_U32_POLICE under
        # TCA_OPTIONS, but there is TCA_U32_ACT
        if police_u32 is None:
            police_u32 = options.get_attr('TCA_U32_ACT').\
                get_attr('TCA_ACT_PRIO_0').\
                get_attr('TCA_ACT_OPTIONS')
        police_tbf = police_u32.get_attr('TCA_POLICE_TBF')
        assert police_tbf['rate'] == 1250
        assert police_tbf['mtu'] == 2040

    @skip_if_not_supported
    def test_bpf_filter(self):
        self.test_simple()
        fd1 = get_simple_bpf_program('sched_cls')
        fd2 = get_simple_bpf_program('sched_act')
        if fd1 == -1 or fd2 == -1:
            # to get bpf filter working, one should have:
            # kernel >= 4.1
            # CONFIG_EXPERT=y
            # CONFIG_BPF_SYSCALL=y
            # CONFIG_NET_CLS_BPF=m/y
            #
            # see `grep -rn BPF_PROG_TYPE_SCHED_CLS kernel_sources`
            raise SkipTest('bpf syscall error')
        self.ip.tc(RTM_NEWTFILTER, 'bpf', self.interface, 0,
                   fd=fd1, name='my_func', parent=0xffff0000,
                   action='ok', classid=1)
        action = {'kind': 'bpf', 'fd': fd2, 'name': 'my_func', 'action': 'ok'}
        self.ip.tc(RTM_NEWTFILTER, 'u32', self.interface, 1,
                   protocol=protocols.ETH_P_ALL, parent=0xffff0000,
                   target=0x10002, keys=['<KEY>'], action=action)
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        assert fls
        bpf_filter = [x for x in fls
                      if x.get_attr('TCA_OPTIONS') is not None and
                      (x.get_attr('TCA_OPTIONS').get_attr('TCA_BPF_ACT')
                       is not None)][0]
        bpf_options = bpf_filter.get_attr('TCA_OPTIONS')
        gact_parms = bpf_options.get_attr('TCA_BPF_ACT').\
            get_attr('TCA_ACT_PRIO_1').\
            get_attr('TCA_ACT_OPTIONS').\
            get_attr('TCA_GACT_PARMS')
        assert gact_parms['action'] == 0

        u32_filter = [x for x in fls
                      if x.get_attr('TCA_OPTIONS') is not None and
                      (x.get_attr('TCA_OPTIONS').get_attr('TCA_U32_ACT')
                       is not None)][0]
        u32_options = u32_filter.get_attr('TCA_OPTIONS')
        bpf_fd = u32_options.get_attr('TCA_U32_ACT').\
            get_attr('TCA_ACT_PRIO_1').\
            get_attr('TCA_ACT_OPTIONS').\
            get_attr('TCA_ACT_BPF_FD')
        assert bpf_fd == fd2

    @skip_if_not_supported
    def test_bpf_filter_policer(self):
        self.test_simple()
        fd = get_simple_bpf_program('sched_cls')
        if fd == -1:
            # see comment above about kernel requirements
            raise SkipTest('bpf syscall error')
        self.ip.tc(RTM_NEWTFILTER, 'bpf', self.interface, 0,
                   fd=fd, name='my_func', parent=0xffff0000,
                   action='ok', classid=1, rate='10kbit',
                   burst=10240, mtu=2040)
        fls = self.ip.get_filters(index=self.interface, parent=0xffff0000)
        # assert the supplied policer is returned to us intact
        plcs = [x for x in fls
                if x.get_attr('TCA_OPTIONS') is not None and
                (x.get_attr('TCA_OPTIONS').get_attr('TCA_BPF_POLICE')
                 is not None)][0]
        options = plcs.get_attr('TCA_OPTIONS')
        police = options.get_attr('TCA_BPF_POLICE').\
            get_attr('TCA_POLICE_TBF')
        assert police['rate'] == 1250
        assert police['mtu'] == 2040


class TestPfifo(BasicTest):

    def test_pfifo(self):
        try_qd('pfifo_fast', self.ip.tc,
               RTM_NEWQDISC, 'pfifo_fast', self.interface, 0)
        qds = self.get_qdisc()
        assert qds
        assert qds.get_attr('TCA_KIND') == 'pfifo_fast'
        assert isinstance(qds.get_attr('TCA_OPTIONS')['priomap'], tuple)


class TestCodel(BasicTest):

    def test_codel(self):
        try_qd('codel', self.ip.tc,
               RTM_NEWQDISC, 'codel', self.interface,
               handle='1:0',
               cdl_interval='40ms',
               cdl_target='2ms',
               cdl_limit=5000,
               cdl_ecn=1)
        qds = self.get_qdisc()
        assert qds
        assert qds.get_attr('TCA_KIND') == 'codel'
        opts = qds.get_attr('TCA_OPTIONS')
        assert opts.get_attr('TCA_CODEL_ECN') == 1
        assert opts.get_attr('TCA_CODEL_LIMIT') == 5000


class TestPlug(BasicTest):

    def test_plug(self):
        try_qd('plug', self.ip.tc,
               RTM_NEWQDISC, 'plug', self.interface, limit=13107)
{'fraction_non_alphanumeric': 0.10674543206484408, 'fraction_numerical': 0.024247836241241928, 'mean_word_length': 2.1188946015424164, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '3795211', 'n_tokens_mistral': 5139, 'n_tokens_neox': 4814, 'n_words': 992}