本文整理汇总了Python中utils.require_user函数的典型用法代码示例。如果您正苦于以下问题:Python require_user函数的具体用法?Python require_user怎么用?Python require_user使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了require_user函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_callback_positive
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
开发者ID:hegusung,项目名称:pyroute2,代码行数:34,代码来源:test_ipdb.py
示例2: test_mass_ipv6
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)
开发者ID:jazzmes,项目名称:pyroute2,代码行数:33,代码来源:test_ipr.py
示例3: test_flush_routes
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')
开发者ID:jazzmes,项目名称:pyroute2,代码行数:26,代码来源:test_ipr.py
示例4: _test_commit_hook_negative
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
开发者ID:0x90,项目名称:pyroute2,代码行数:32,代码来源:test_ipdb.py
示例5: test_create_vxlan
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()
开发者ID:0x90,项目名称:pyroute2,代码行数:26,代码来源:test_ipdb.py
示例6: test_slave_data
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
开发者ID:abn,项目名称:pyroute2,代码行数:25,代码来源:test_ipdb.py
示例7: test_lwtunnel_multipath_mpls
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)
开发者ID:svinota,项目名称:pyroute2,代码行数:31,代码来源:test_ipr.py
示例8: setup
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))
开发者ID:celebdor,项目名称:pyroute2,代码行数:7,代码来源:test_ndb.py
示例9: test_routes_multipath_gateway
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')
开发者ID:thezeep,项目名称:pyroute2,代码行数:27,代码来源:test_ipdb.py
示例10: test_dummy0_unloaded
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()
开发者ID:thezeep,项目名称:pyroute2,代码行数:25,代码来源:test_ipdb.py
示例11: test_addr_broadcast_default
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()
开发者ID:celebdor,项目名称:pyroute2,代码行数:7,代码来源:test_ipr.py
示例12: setup
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)
开发者ID:celebdor,项目名称:pyroute2,代码行数:25,代码来源:test_dhcp.py
示例13: test_generic_post_callback
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)
开发者ID:wavezhang,项目名称:pyroute2,代码行数:35,代码来源:test_ipdb.py
示例14: test_commit_barrier
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()
开发者ID:abn,项目名称:pyroute2,代码行数:34,代码来源:test_ipdb.py
示例15: setup
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()
开发者ID:svinota,项目名称:pyroute2,代码行数:7,代码来源:test_stress.py
示例16: test_routes_multipath_gateway
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")
开发者ID:abn,项目名称:pyroute2,代码行数:31,代码来源:test_ipdb.py
示例17: setup
def setup(self):
require_user('root')
self.rlink = Rlink()
try:
self.rlink.bind('EXMPL_GENL', rcmd)
except NetlinkError:
raise SkipTest('module not loaded')
开发者ID:celebdor,项目名称:pyroute2,代码行数:7,代码来源:test_generic.py
示例18: test_route_replace_existing
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))
开发者ID:svinota,项目名称:pyroute2,代码行数:28,代码来源:test_ipr.py
示例19: test_ancient_bridge
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
开发者ID:skamithi,项目名称:pyroute2,代码行数:32,代码来源:test_ipdb.py
示例20: _test_commit_hook_positive
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
开发者ID:0x90,项目名称:pyroute2,代码行数:31,代码来源:test_ipdb.py
注:本文中的utils.require_user函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论