Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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) < 2
except:
raise
finally:
config.commit_barrier = 0.2
ip.interfaces[ifname].remove().commit()
ip.release()
def test_rename_plus_ipv6(self):
require_user('root')
mtu = 1280 # mtu must be >= 1280 if you plan to use IPv6
txqlen = 2000
nsid = str(uuid4())
ipdb_main = IPDB()
ipdb_test = IPDB(nl=NetNS(nsid))
if1 = uifname()
if2 = uifname()
if3 = uifname()
# create
ipdb_main.create(kind='veth',
ifname=if1,
peer=if2,
mtu=mtu,
txqlen=txqlen).commit()
# move
with ipdb_main.interfaces[if2] as veth:
veth.net_ns_fd = nsid
# set it up
with ipdb_test.interfaces[if2] as veth:
veth.add_ip('fdb3:84e5:4ff4:55e4::1/64')
def test_bridge_deps(self):
self.if_br0 = uifname()
self.if_br0p0 = uifname()
self.if_br0p1 = uifname()
ifaddr1 = self.ifaddr()
ifaddr2 = self.ifaddr()
router = self.ifaddr()
dst = str(self.ipnets[1].network)
(self
.interfaces
.append(self
.ndb
.interfaces
.create(ifname=self.if_br0,
kind='bridge',
state='up')
.commit()['index']))
(self
.interfaces
def test_view_constraints_pipeline(self):
ifname = uifname()
ifaddr = self.ifaddr()
(self
.ndb
.interfaces
.constraint('target', self.netns)
.create(ifname=ifname, kind='dummy')
.set('address', '00:11:22:33:44:55')
.set('state', 'up')
.ipaddr
.create(address=ifaddr, prefixlen=24)
.commit())
self._assert_test_view(ifname, ifaddr)
def test_pushns(self):
require_user('root')
foo = str(uuid4())
ifA = uifname()
with IPRoute() as ipr:
ipr.link('add', ifname=ifA, kind='dummy')
netnsmod.pushns(foo)
with IPRoute() as ipr:
assert ifA not in [x.get_attr('IFLA_IFNAME') for x
in ipr.link('dump')]
netnsmod.popns()
with IPRoute() as ipr:
assert ifA in [x.get_attr('IFLA_IFNAME') for x
in ipr.link('dump')]
ipr.link('del', index=ipr.link_lookup(ifname=ifA)[0])
netnsmod.remove(foo)
def test_create_fail(self):
ifname = uifname()
kind = uifname()
interface = self.ip.create(ifname=ifname, kind=kind)
try:
with self.ip.interfaces[ifname] as i:
pass
except:
pass
assert ifname in self.ip.interfaces
assert self.ip.interfaces[ifname]['ipdb_scope'] == 'create'
assert self.ip.interfaces[ifname]['kind'] == kind
with self.ip.interfaces[ifname] as i:
i.remove()
assert ifname not in self.ip.interfaces
assert interface['ipdb_scope'] == 'invalid'
def ifname(self):
'''
Register and return a new unique interface name to
be used in a test.
'''
ifname = uifname()
self.ifnames.append(ifname)
self.rtm_events[ifname] = threading.Event()
self.rtm_newlink[ifname] = []
self.rtm_dellink[ifname] = []
return ifname
def get_ifname(self):
ifname = uifname()
self.ifaces.append(ifname)
return ifname
from pyroute2 import IPDB
from pyroute2.common import uifname
p0 = uifname()
ip = IPDB()
# create dummy interface to host routes on
ip.create(kind='dummy', ifname=p0).\
add_ip('172.16.1.1/24').\
up().\
commit()
# create a route
with ip.routes.add({'dst': '172.16.0.0/24',
'gateway': '172.16.1.2'}) as r:
pass
# modify it
with ip.routes['172.16.0.0/24'] as r:
r.gateway = '172.16.1.3'