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_auto_netns(self):
newns = str(uuid.uuid4())
assert self.ndb.interfaces.count() > 0
assert len(tuple(self
.ndb
.interfaces
.summary(match={'target': 'netns/%s' % newns}))) == 0
netns.create(newns)
self.ndb.interfaces.wait(**{'target': 'netns/%s' % newns})
netns.remove(newns)
self.ip.commit()
assert 'x' + ifA not in self.ip.interfaces
assert 'x' + ifB not in self.ip.interfaces
assert ifA in self.ip.interfaces
assert ifB in self.ip.interfaces
assert self.ip.interfaces[ifA].flags & 1
assert self.ip.interfaces[ifB].flags & 1
if self.ip.interfaces[ifA]._mode == 'explicit':
self.ip.interfaces[ifA].begin()
self.ip.interfaces[ifB].begin()
self.ip.interfaces[ifA].remove()
self.ip.interfaces[ifB].remove()
self.ip.commit()
netns.remove(ns)
def setup(self):
require_user('root')
try:
netns.create(NS_NAME)
except OSError as e:
if e.errno == errno.EEXIST:
netns.remove(NS_NAME)
netns.create(NS_NAME)
else:
raise
try:
check_output([NFT_BIN_PATH, "-f", "nftables.ruleset"])
except OSError as e:
if e.errno == errno.ENOENT:
raise SkipTest("You must install nftables for the test")
else:
raise
def in_cgroup():
try:
pid = os.getpid()
cg = Cgroup(name)
for env in env_vars:
log.info('Setting ENV %s' % env)
os.putenv(*env.split('=', 1))
# Set network namespace
netns.setns(netns_name)
# add process to cgroup
cg.add(pid)
os.chroot(layer_dir)
if working_dir != '':
log.info("Setting working directory to %s" % working_dir)
os.chdir(working_dir)
except Exception as e:
traceback.print_exc()
log.error("Failed to preexecute function")
log.error(e)
cmd = start_cmd
def create(self, path):
netnspath = netns._get_netnspath(path)
try:
netns.create(netnspath, self.libc)
except OSError as e:
raise NetlinkError(e.errno)
info = self.ipr._dump_one_ns(netnspath, set())
info['header']['type'] = RTM_NEWNETNS
info['event'] = 'RTM_NEWNETNS'
del info['value']
return info,
def start(self):
'''
Start the network namespace object, and create the network
namespace if it doesn't exist.
'''
self.set_starting()
self.logger.debug("starting network namespace '%s'" % self.name)
if not self.dry_run:
if self.name not in pyroute2.netns.listnetns():
try:
pyroute2.netns.create(self.name)
except FileExistsError:
# Network namespace already exists
pass
except Exception as exc:
# pylint: disable=no-member
raise UnableToCreateNetnsError(self, exc.message)
# pylint: disable=no-member
self.logger.debug("creating NetNS and IPDB objects for '%s'" % self.name)
self.netns = pyroute2.NetNS(self.name)
self.ipdb = pyroute2.IPDB(nl=self.netns)
# Bring up the loopback interface inside the namespace.
lo_if = interface.get(self.dry_run, self.logger, "lo", netns=self)
def remove(self, path):
netnspath = netns._get_netnspath(path)
info = None
try:
info = self.ipr._dump_one_ns(netnspath, set())
except SkipInode:
raise NetlinkError(errno.EEXIST)
info['header']['type'] = RTM_DELNETNS
info['event'] = 'RTM_DELNETNS'
del info['value']
try:
netns.remove(netnspath, self.libc)
except OSError as e:
raise NetlinkError(e.errno)
return info,