def test_callback_positive(self):
        require_user('root')
        assert 'dummyX' in self.ip.interfaces

        # test callback, that adds an address by itself --
        # just to check the possibility
        def cb(snapshot, transaction):
            self.ip.nl.addr('add',
                            self.ip.interfaces.dummyX.index,
                            address='172.16.22.1',
                            mask=24)

        # register callback and check CB chain length
        self.ip.interfaces.dummyX.register_callback(cb)
        assert len(self.ip.interfaces.dummyX._callbacks) == 1

        # create a transaction and commit it
        if self.ip.interfaces.dummyX._mode == 'explicit':
            self.ip.interfaces.dummyX.begin()
        self.ip.interfaces.dummyX.add_ip('172.16.21.1/24')
        self.ip.interfaces.dummyX.commit()

        # the second address added w/o watchdogs,
        # so we have to wait
        time.sleep(1)

        # added address should be there
        assert ('172.16.21.1', 24) in self.ip.interfaces.dummyX.ipaddr
        # and the one, added by the callback, too
        assert ('172.16.22.1', 24) in self.ip.interfaces.dummyX.ipaddr

        # unregister callback
        self.ip.interfaces.dummyX.unregister_callback(cb)
        assert len(self.ip.interfaces.dummyX._callbacks) == 0

    def test_mass_ipv6(self):
        #
        # Achtung! This test is time consuming.
        # It is really time consuming, I'm not not
        # kidding you. Beware.
        #
        require_user('root')
        base = 'fdb3:84e5:4ff4:55e4::{0}'
        limit = int(os.environ.get('PYROUTE2_SLIMIT', '0x800'), 16)

        # add addresses
        for idx in range(limit):
            self.ip.addr('add', self.ifaces[0],
                         base.format(hex(idx)[2:]), 48)

        # assert addresses in two steps, to ease debug
        addrs = self.ip.get_addr(10)
        assert len(addrs) >= limit

        # clean up addresses
        #
        # it is not required, but if you don't do that,
        # you'll get this on the interface removal:
        #
        # >> kernel:BUG: soft lockup - CPU#0 stuck for ...
        #
        # so, not to scare people, remove addresses gracefully
        # one by one
        #
        # it also verifies all the addresses are in place
        for idx in reversed(range(limit)):
            self.ip.addr('delete', self.ifaces[0],
                         base.format(hex(idx)[2:]), 48)

    def test_flush_routes(self):
        require_user('root')
        self.ip.link('set', index=self.ifaces[0], state='up')
        self.ip.addr('add', self.ifaces[0], address='172.16.0.2', mask=24)
        self.ip.route('add',
                      prefix='172.16.1.0',
                      mask=24,
                      gateway='172.16.0.1',
                      table=100)
        self.ip.route('add',
                      prefix='172.16.2.0',
                      mask=24,
                      gateway='172.16.0.1',
                      table=100)

        assert grep('ip route show table 100',
                    pattern='172.16.1.0/24.*172.16.0.1')
        assert grep('ip route show table 100',
                    pattern='172.16.2.0/24.*172.16.0.1')

        self.ip.flush_routes(table=100)

        assert not grep('ip route show table 100',
                        pattern='172.16.1.0/24.*172.16.0.1')
        assert not grep('ip route show table 100',
                        pattern='172.16.2.0/24.*172.16.0.1')

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

        # test exception to differentiate
        class CBException(Exception):
            pass

        # test callback, that always fail
        def cb(interface, snapshot, transaction):
            raise CBException()

        # register callback and check CB chain length
        self.ip.interfaces[self.ifd].register_commit_hook(cb)
        assert len(self.ip.interfaces[self.ifd]._commit_hooks) == 1

        # create a transaction and commit it; should fail
        # 'cause of the callback
        if self.ip.interfaces[self.ifd]._mode == 'explicit':
            self.ip.interfaces[self.ifd].begin()
        self.ip.interfaces[self.ifd].add_ip('172.16.21.1/24')
        try:
            self.ip.interfaces[self.ifd].commit()
        except CBException:
            pass

        # added address should be removed
        assert ('172.16.21.1', 24) not in \
            self.ip.interfaces[self.ifd].ipaddr

        # unregister callback
        self.ip.interfaces[self.ifd].unregister_commit_hook(cb)
        assert len(self.ip.interfaces[self.ifd]._commit_hooks) == 0

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

        ifL = self.get_ifname()
        ifV = self.get_ifname()
        ifdb = self.ip.interfaces

        self.ip.create(kind='dummy',
                       ifname=ifL).commit()
        self.ip.create(kind='vxlan',
                       ifname=ifV,
                       vxlan_link=ifdb[ifL],
                       vxlan_id=101,
                       vxlan_group='239.1.1.1').commit()

        ip2 = IPDB()
        ifdb = ip2.interfaces

        try:
            assert ifdb[ifV].vxlan_link == ifdb[ifL].index
            assert ifdb[ifV].vxlan_group == '239.1.1.1'
            assert ifdb[ifV].vxlan_id == 101
        except Exception:
            raise
        finally:
            ip2.release()

    def test_slave_data(self):
        require_user("root")

        ifBR = self.get_ifname()
        ifP = self.get_ifname()
        self.ip.debug = True

        bridge = self.ip.create(ifname=ifBR, kind="bridge").commit()
        port = self.ip.create(ifname=ifP, kind="dummy").commit()

        if self.ip.mode == "explicit":
            bridge.begin()
        bridge.add_port(port)
        bridge.up()
        bridge.commit()

        li = port.nlmsg.get_attr("IFLA_LINKINFO")
        skind = li.get_attr("IFLA_INFO_SLAVE_KIND")
        sdata = li.get_attr("IFLA_INFO_SLAVE_DATA")
        self.ip.debug = False
        if skind is None or sdata is None:
            raise SkipTest("slave data not provided")

        assert sdata.get_attr("IFLA_BRPORT_STATE") is not None
        assert sdata.get_attr("IFLA_BRPORT_MODE") is not None

 def test_lwtunnel_multipath_mpls(self):
     require_kernel(4, 4)
     require_user('root')
     require_kernel(4, 5)
     naddr = str(self.ipnets[1].network)
     self.ip.route('add',
                   dst='%s/24' % naddr,
                   multipath=[{'encap': {'type': 'mpls',
                                         'labels': 500},
                               'oif': 1},
                              {'encap': {'type': 'mpls',
                                         'labels': '600/700'},
                               'gateway': '127.0.0.4'}])
     routes = self.ip.route('dump', dst='%s/24' % naddr)
     assert len(routes) == 1
     mp = routes[0].get_attr('RTA_MULTIPATH')
     assert len(mp) == 2
     assert mp[0]['oif'] == 1
     assert mp[0].get_attr('RTA_ENCAP_TYPE') == 1
     labels = mp[0].get_attr('RTA_ENCAP').get_attr('MPLS_IPTUNNEL_DST')
     assert len(labels) == 1
     assert labels[0]['bos'] == 1
     assert labels[0]['label'] == 500
     assert mp[1].get_attr('RTA_ENCAP_TYPE') == 1
     labels = mp[1].get_attr('RTA_ENCAP').get_attr('MPLS_IPTUNNEL_DST')
     assert len(labels) == 2
     assert labels[0]['bos'] == 0
     assert labels[0]['label'] == 600
     assert labels[1]['bos'] == 1
     assert labels[1]['label'] == 700
     self.ip.route('del', dst='%s/24' % naddr)

 def setup(self):
     require_user('root')
     self.if_simple = None
     self.interfaces = self.create_interfaces()
     self.ndb = NDB(db_provider=self.db_provider,
                    db_spec=self.db_spec,
                    sources=self.nl_class(**self.nl_kwarg))

    def test_routes_multipath_gateway(self):
        require_user('root')
        ifR = self.get_ifname()

        with self.ip.create(ifname=ifR, kind='dummy') as i:
            i.add_ip('172.16.231.1/24')
            i.up()

        r = self.ip.routes.add({'dst': '172.16.232.0/24',
                                'multipath': [{'gateway': '172.16.231.2',
                                               'hops': 20},
                                              {'gateway': '172.16.231.3',
                                               'hops': 30},
                                              {'gateway': '172.16.231.4'}]})
        r.commit()
        assert grep('ip ro', pattern='172.16.232.0/24')
        assert grep('ip ro', pattern='nexthop.*172.16.231.2.*weight.*21')
        assert grep('ip ro', pattern='nexthop.*172.16.231.3.*weight.*31')
        assert grep('ip ro', pattern='nexthop.*172.16.231.4.*weight.*1')

        with self.ip.routes['172.16.232.0/24'] as r:
            r.add_nh({'gateway': '172.16.231.5', 'hops': 50})
            r.del_nh({'gateway': '172.16.231.2'})
        assert grep('ip ro', pattern='172.16.232.0/24')
        assert grep('ip ro', pattern='nexthop.*172.16.231.5.*weight.*51')
        assert grep('ip ro', pattern='nexthop.*172.16.231.3.*weight.*31')
        assert grep('ip ro', pattern='nexthop.*172.16.231.4.*weight.*1')

 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_addr_broadcast_default(self):
     require_user('root')
     self.ip.addr('add', self.ifaces[0],
                  address='172.16.0.1',
                  mask=24,
                  broadcast=True)
     assert '172.16.0.255' in get_ip_brd()

    def setup(self):
        require_user('root')
        require_executable('busybox')
        self.ip = IPDB()
        # create internal network
        self.if1 = uifname()
        self.if2 = uifname()
        self.ip.create(kind='veth', ifname=self.if1, peer=self.if2).commit()
        # set interfaces up
        with self.ip.interfaces[self.if1] as i:
            i.add_ip('172.16.101.1/24')
            i.up()

        with self.ip.interfaces[self.if2] as i:
            i.up()
        # prepare configuration for udhcpd
        with open('udhcpd.conf.in', 'r') as conf_in:
            with open('udhcpd.conf', 'w') as conf_out:
                conf_out.write('interface %s\n' % self.if1)
                conf_out.write(conf_in.read())
        # run busybox dhcp server on $if1
        with open(os.devnull, 'w') as fnull:
            subprocess.check_call(['busybox', 'udhcpd', 'udhcpd.conf'],
                                  stdout=fnull,
                                  stderr=fnull)

    def test_generic_post_callback(self):
        require_user('root')
        require_bridge()

        ifP1 = self.get_ifname()
        ifP2 = self.get_ifname()
        ifM = self.get_ifname()

        def cb(ipdb, msg, action):
            if action == 'RTM_NEWLINK' and \
                    msg.get_attr('IFLA_IFNAME', '') in (ifP1, ifP2):
                obj = ipdb.interfaces[msg['index']]
                ipdb.interfaces[ifM].add_port(obj)
                ipdb.interfaces[ifM].commit()

        wd0 = self.ip.watchdog(ifname=ifM)
        wd1 = self.ip.watchdog(ifname=ifP1)
        wd2 = self.ip.watchdog(ifname=ifP2)
        # create bridge
        self.ip.create(kind='bridge', ifname=ifM).commit()
        wd0.wait()
        # register callback
        self.ip.register_callback(cb)
        # create ports
        self.ip.create(kind='dummy', ifname=ifP1).commit()
        self.ip.create(kind='dummy', ifname=ifP2).commit()
        wd1.wait()
        wd2.wait()
        # check that ports are attached
        assert self.ip.interfaces[ifP1].index in \
            self.ip.interfaces[ifM].ports
        assert self.ip.interfaces[ifP2].index in \
            self.ip.interfaces[ifM].ports
        # unregister callback
        self.ip.unregister_callback(cb)

    def test_commit_barrier(self):
        require_user("root")

        ifname = uifname()

        # barrier 0
        try:
            ip = IPDB()
            config.commit_barrier = 0
            ts1 = time.time()
            ip.create(ifname=ifname, kind="dummy").commit()
            ts2 = time.time()
            assert 0 < (ts2 - ts1) < 1
        except:
            raise
        finally:
            config.commit_barrier = 0.2
            ip.interfaces[ifname].remove().commit()
            ip.release()

        # barrier 5
        try:
            ip = IPDB()
            config.commit_barrier = 5
            ts1 = time.time()
            ip.create(ifname=ifname, kind="dummy").commit()
            ts2 = time.time()
            assert 5 < (ts2 - ts1) < 6
        except:
            raise
        finally:
            config.commit_barrier = 0.2
            ip.interfaces[ifname].remove().commit()
            ip.release()

 def setup(self):
     require_user('root')
     self._nofile = resource.getrlimit(resource.RLIMIT_NOFILE)
     soft, hard = self._nofile
     new_limit = (min(soft, RESPAWNS / 2), min(hard, RESPAWNS / 2))
     resource.setrlimit(resource.RLIMIT_NOFILE, new_limit)
     self._socket_fd_count = count_socket_fds()

    def test_routes_multipath_gateway(self):
        require_user("root")
        ifR = self.get_ifname()

        with self.ip.create(ifname=ifR, kind="dummy") as i:
            i.add_ip("172.16.231.1/24")
            i.up()

        r = self.ip.routes.add(
            {
                "dst": "172.16.232.0/24",
                "multipath": [
                    {"gateway": "172.16.231.2", "hops": 20},
                    {"gateway": "172.16.231.3", "hops": 30},
                    {"gateway": "172.16.231.4"},
                ],
            }
        )
        r.commit()
        assert grep("ip ro", pattern="172.16.232.0/24")
        assert grep("ip ro", pattern="nexthop.*172.16.231.2.*weight.*21")
        assert grep("ip ro", pattern="nexthop.*172.16.231.3.*weight.*31")
        assert grep("ip ro", pattern="nexthop.*172.16.231.4.*weight.*1")

        with self.ip.routes["172.16.232.0/24"] as r:
            r.add_nh({"gateway": "172.16.231.5", "hops": 50})
            r.del_nh({"gateway": "172.16.231.2"})
        assert grep("ip ro", pattern="172.16.232.0/24")
        assert grep("ip ro", pattern="nexthop.*172.16.231.5.*weight.*51")
        assert grep("ip ro", pattern="nexthop.*172.16.231.3.*weight.*31")
        assert grep("ip ro", pattern="nexthop.*172.16.231.4.*weight.*1")

 def setup(self):
     require_user('root')
     self.rlink = Rlink()
     try:
         self.rlink.bind('EXMPL_GENL', rcmd)
     except NetlinkError:
         raise SkipTest('module not loaded')

 def test_route_replace_existing(self):
     # route('replace', ...) should succeed, if route exists
     require_user('root')
     naddr = str(self.ipnets[1].network)
     ifaddr1 = self.ifaddr()
     ifaddr2 = self.ifaddr()
     ifaddr3 = self.ifaddr()
     self.ip.link('set', index=self.ifaces[0], state='up')
     self.ip.addr('add', self.ifaces[0], address=ifaddr1, mask=24)
     self.ip.route('replace',
                   dst=naddr,
                   mask=24,
                   gateway=ifaddr2,
                   table=100)
     assert grep('ip route show table 100',
                 pattern='%s/24.*%s' % (naddr, ifaddr2))
     self.ip.route('replace',
                   dst=naddr,
                   mask=24,
                   gateway=ifaddr3,
                   table=100)
     assert not grep('ip route show table 100',
                     pattern='%s/24.*%s' % (naddr, ifaddr2))
     assert grep('ip route show table 100',
                 pattern='%s/24.*%s' % (naddr, ifaddr3))
     self.ip.flush_routes(table=100)
     assert not grep('ip route show table 100',
                     pattern='%s/24.*%s' % (naddr, ifaddr3))

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

        # create ports
        with self.ip.create(kind='dummy', ifname='bala_port0'):
            pass
        with self.ip.create(kind='dummy', ifname='bala_port1'):
            pass
        assert 'bala_port0' in self.ip.interfaces
        assert 'bala_port1' in self.ip.interfaces

        set_ancient(True)

        # create bridge
        try:
            with self.ip.create(kind='bridge', ifname='bala') as i:
                i.add_ip('172.16.0.1/24')
                i.add_ip('172.16.0.2/24')
                i.add_port(self.ip.interfaces.bala_port0)
                i.add_port(self.ip.interfaces.bala_port1)
        finally:
            set_ancient(False)

        assert 'bala' in self.ip.interfaces
        assert 'bala_port0' in self.ip.interfaces
        assert 'bala_port1' in self.ip.interfaces
        assert ('172.16.0.1', 24) in self.ip.interfaces.bala.ipaddr
        assert ('172.16.0.2', 24) in self.ip.interfaces.bala.ipaddr
        assert self.ip.interfaces.bala_port0.index in \
            self.ip.interfaces.bala.ports
        assert self.ip.interfaces.bala_port1.index in \
            self.ip.interfaces.bala.ports

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

        # test callback, that adds an address by itself --
        # just to check the possibility
        def cb(interface, snapshot, transaction):
            self.ip.nl.addr('add',
                            self.ip.interfaces[self.ifd].index,
                            address='172.16.22.1',
                            mask=24)

        # register callback and check CB chain length
        self.ip.interfaces[self.ifd].register_commit_hook(cb)
        assert len(self.ip.interfaces[self.ifd]._commit_hooks) == 1

        # create a transaction and commit it
        if self.ip.interfaces[self.ifd]._mode == 'explicit':
            self.ip.interfaces[self.ifd].begin()
        self.ip.interfaces[self.ifd].add_ip('172.16.21.1/24')
        self.ip.interfaces[self.ifd].commit()

        # added address should be there
        assert ('172.16.21.1', 24) in \
            self.ip.interfaces[self.ifd].ipaddr
        # and the one, added by the callback, too
        assert ('172.16.22.1', 24) in \
            self.ip.interfaces[self.ifd].ipaddr

        # unregister callback
        self.ip.interfaces[self.ifd].unregister_commit_hook(cb)
        assert len(self.ip.interfaces[self.ifd]._commit_hooks) == 0