tools: tc-testing: nsPlugin (a13fedbe) · Commits · e / devices / android_kernel_fairphone_FP4

tools/testing/selftests/tc-testing/plugin-lib/nsPlugin.py

0 → 100644

+141 −0

Original line number	Diff line number	Diff line
		import os
		import signal
		from string import Template
		import subprocess
		import time
		from TdcPlugin import TdcPlugin

		from tdc_config import *

		class SubPlugin(TdcPlugin):
		def __init__(self):
		self.sub_class = 'ns/SubPlugin'
		super().__init__()

		def pre_suite(self, testcount, testidlist):
		'''run commands before test_runner goes into a test loop'''
		super().pre_suite(testcount, testidlist)

		if self.args.namespace:
		self._ns_create()

		def post_suite(self, index):
		'''run commands after test_runner goes into a test loop'''
		super().post_suite(index)
		if self.args.verbose:
		print('{}.post_suite'.format(self.sub_class))

		if self.args.namespace:
		self._ns_destroy()

		def add_args(self, parser):
		super().add_args(parser)
		self.argparser_group = self.argparser.add_argument_group(
		'netns',
		'options for nsPlugin(run commands in net namespace)')
		self.argparser_group.add_argument(
		'-n', '--namespace', action='store_true',
		help='Run commands in namespace')
		return self.argparser

		def adjust_command(self, stage, command):
		super().adjust_command(stage, command)
		cmdform = 'list'
		cmdlist = list()

		if not self.args.namespace:
		return command

		if self.args.verbose:
		print('{}.adjust_command'.format(self.sub_class))

		if not isinstance(command, list):
		cmdform = 'str'
		cmdlist = command.split()
		else:
		cmdlist = command
		if stage == 'setup' or stage == 'execute' or stage == 'verify' or stage == 'teardown':
		if self.args.verbose:
		print('adjust_command: stage is {}; inserting netns stuff in command [{}] list [{}]'.format(stage, command, cmdlist))
		cmdlist.insert(0, self.args.NAMES['NS'])
		cmdlist.insert(0, 'exec')
		cmdlist.insert(0, 'netns')
		cmdlist.insert(0, 'ip')
		else:
		pass

		if cmdform == 'str':
		command = ' '.join(cmdlist)
		else:
		command = cmdlist

		if self.args.verbose:
		print('adjust_command: return command [{}]'.format(command))
		return command

		def _ns_create(self):
		'''
		Create the network namespace in which the tests will be run and set up
		the required network devices for it.
		'''
		if self.args.namespace:
		cmd = 'ip netns add {}'.format(self.args.NAMES['NS'])
		self._exec_cmd('pre', cmd)
		cmd = 'ip link add $DEV0 type veth peer name $DEV1'
		self._exec_cmd('pre', cmd)
		cmd = 'ip link set $DEV1 netns {}'.format(self.args.NAMES['NS'])
		self._exec_cmd('pre', cmd)
		cmd = 'ip link set $DEV0 up'
		self._exec_cmd('pre', cmd)
		cmd = 'ip -n {} link set $DEV1 up'.format(self.args.NAMES['NS'])
		self._exec_cmd('pre', cmd)
		if self.args.device:
		cmd = 'ip link set $DEV2 netns {}'.format(self.args.NAMES['NS'])
		self._exec_cmd('pre', cmd)
		cmd = 'ip -n {} link set $DEV2 up'.format(self.args.NAMES['NS'])
		self._exec_cmd('pre', cmd)

		def _ns_destroy(self):
		'''
		Destroy the network namespace for testing (and any associated network
		devices as well)
		'''
		if self.args.namespace:
		cmd = 'ip netns delete {}'.format(self.args.NAMES['NS'])
		self._exec_cmd('post', cmd)

		def _exec_cmd(self, stage, command):
		'''
		Perform any required modifications on an executable command, then run
		it in a subprocess and return the results.
		'''
		if '$' in command:
		command = self._replace_keywords(command)

		self.adjust_command(stage, command)
		if self.args.verbose:
		print('_exec_cmd: command "{}"'.format(command))
		proc = subprocess.Popen(command,
		shell=True,
		stdout=subprocess.PIPE,
		stderr=subprocess.PIPE,
		env=ENVIR)
		(rawout, serr) = proc.communicate()

		if proc.returncode != 0 and len(serr) > 0:
		foutput = serr.decode("utf-8")
		else:
		foutput = rawout.decode("utf-8")

		proc.stdout.close()
		proc.stderr.close()
		return proc, foutput

		def _replace_keywords(self, cmd):
		"""
		For a given executable command, substitute any known
		variables contained within NAMES with the correct values
		"""
		tcmd = Template(cmd)
		subcmd = tcmd.safe_substitute(self.args.NAMES)
		return subcmd

tools/testing/selftests/tc-testing/tdc.py

+1 −44

Original line number	Diff line number	Diff line
		@@ -23,8 +23,6 @@ from tdc_helper import *

		import TdcPlugin

		USE_NS = True

		class PluginMgr:
		def __init__(self, argparser):
		super().__init__()
		@@ -107,16 +105,13 @@ def replace_keywords(cmd):
		return subcmd


		def exec_cmd(args, pm, stage, command, nsonly=True):
		def exec_cmd(args, pm, stage, command):
		"""
		Perform any required modifications on an executable command, then run
		it in a subprocess and return the results.
		"""
		if len(command.strip()) == 0:
		return None, None
		if (USE_NS and nsonly):
		command = 'ip netns exec $NS ' + command

		if '$' in command:
		command = replace_keywords(command)

		@@ -265,39 +260,6 @@ def test_runner(pm, args, filtered_tests):

		return tap


		def ns_create(args, pm):
		"""
		Create the network namespace in which the tests will be run and set up
		the required network devices for it.
		"""
		if (USE_NS):
		cmd = 'ip netns add $NS'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip link add $DEV0 type veth peer name $DEV1'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip link set $DEV1 netns $NS'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip link set $DEV0 up'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip -n $NS link set $DEV1 up'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip link set $DEV2 netns $NS'
		exec_cmd(args, pm, 'pre', cmd, False)
		cmd = 'ip -n $NS link set $DEV2 up'
		exec_cmd(args, pm, 'pre', cmd, False)


		def ns_destroy(args, pm):
		"""
		Destroy the network namespace for testing (and any associated network
		devices as well)
		"""
		if (USE_NS):
		cmd = 'ip netns delete $NS'
		exec_cmd(args, pm, 'post', cmd, False)


		def has_blank_ids(idlist):
		"""
		Search the list for empty ID fields and return true/false accordingly.
		@@ -579,17 +541,12 @@ def set_operation_mode(pm, args):
		list_test_cases(alltests)
		exit(0)

		ns_create(args, pm)

		if len(alltests):
		catresults = test_runner(pm, args, alltests)
		else:
		catresults = 'No tests found\n'
		print('All test results: \n\n{}'.format(catresults))

		ns_destroy(args, pm)


		def main():
		"""
		Start of execution; set up argument parser and get the arguments,