Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'event': event})
assert event.wait(5)
with self.ndb.readonly:
s = len(list(self.ndb.interfaces.summary()))
assert self.count_interfaces(nsname) > 0
assert self.count_interfaces('localhost') < s
# disconnect the source
self.ndb.sources[nsname].close()
with self.ndb.readonly:
s = len(list(self.ndb.interfaces.summary()))
assert self.count_interfaces(nsname) == 0
assert self.count_interfaces('localhost') <= s
netns.remove(nsname)
foo = str(uuid4())
#
# please notice, that IPRoute / IPDB, started in a netns, will continue
# to work in a given netns even if the process changes to another netns
#
with IPRoute() as ip:
links_main1 = set([x.get('index', None) for x in ip.get_links()])
netnsmod.setns(foo)
with IPRoute() as ip:
links_foo = set([x.get('index', None) for x in ip.get_links()])
netnsmod.setns(fd)
with IPRoute() as ip:
links_main2 = set([x.get('index', None) for x in ip.get_links()])
assert links_main1 == links_main2
assert links_main1 != links_foo
netnsmod.remove(foo)
fd.close()
def teardown(self):
for net in self.ipnets:
free_network(net)
self.ndb.close()
netns.remove(self.netns)
iface.net_ns_pid = ns_bar.nl.child
assert ifA in ns_bar.interfaces.keys()
assert ifA not in ns_foo.interfaces.keys()
with ns_bar.interfaces[ifA] as iface:
iface.net_ns_pid = ns_foo.nl.child
assert ifA not in ns_bar.interfaces.keys()
assert ifA in ns_foo.interfaces.keys()
finally:
ns_foo.release()
ns_bar.release()
netnsmod.remove(foo)
netnsmod.remove(bar)
def teardown(self):
self.ip.close()
for ns in self.names:
netnsmod.remove(ns)
for ns_name in ns_names:
NetNS(ns_name)
for _t in range(test_count):
threads = [
Thread(target=_ns_worker,
args=(netnsmod._get_netnspath(ns_name), i, success))
for i, ns_name in enumerate(ns_names)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
for ns_name in ns_names:
netnsmod.remove(ns_name)
assert success[0]
foo = str(uuid4())
netnsmod.create(foo)
netnsmod.pushns(foo)
foo_pid, foo_fd = os.forkpty()
if not foo_pid:
waiting_child(foo_fd)
netnsmod.popns()
pids = netnsmod.ns_pids()
ns_name = netnsmod.pid_to_ns(foo_pid)
try:
assert pids[foo] == [foo_pid]
assert ns_name == foo
finally:
os.close(foo_fd)
netnsmod.remove(foo)
def remove_netns(name, **kwargs):
"""Remove a network namespace.
:param name: The name of the namespace to remove
"""
try:
netns.remove(name, libc=_get_cdll())
except OSError as e:
if e.errno != errno.ENOENT:
raise
def del_netns_by_name(ns):
netns.remove(ns)
def release(self):
super().release()
name = self.context.make_network_namespace_name()
try:
pyroute2.netns.remove(name)
except OSError as e:
raise exc.SystemResourceError(type(self), name) from e