openkylin
/
linux
mirror of https://gitee.com/openkylin/linux.git
9
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'tc-testing-Add-plugin-for-simple-traffic-generation' 

Lucas Bates says:

====================
tc-testing: Add plugin for simple traffic generation

This series supersedes the previous submission that included a patch for test
case verification using JSON output.  It adds a new tdc plugin, scapyPlugin, as
a way to send traffic to test tc filters and actions.

The first patch makes a change to the TdcPlugin module that will allow tdc
plugins to examine the test case currently being executed, so plugins can
play a more active role in testing by accepting information or commands from
the test case.  This is required for scapyPlugin to work.

The second patch adds scapyPlugin itself, and an example test case file to
demonstrate how the scapy block works in the test cases.
====================

Signed-off-by: David S. Miller <davem@davemloft.net>
This commit is contained in:
David S. Miller 2019-07-09 14:07:20 -07:00
parent a1cd4e4324 14e5175e9e
commit 6ec3d4d225
4 changed files with 155 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tools/testing/selftests/tc-testing/TdcPlugin.py
View File

@ -18,12 +18,11 @@ class TdcPlugin:
if self.args.verbose > 1:
print(' -- {}.post_suite'.format(self.sub_class))
def pre_case(self, testid, test_name, test_skip):
def pre_case(self, caseinfo, test_skip):
'''run commands before test_runner does one test'''
if self.args.verbose > 1:
print(' -- {}.pre_case'.format(self.sub_class))
self.args.testid = testid
self.args.test_name = test_name
self.args.caseinfo = caseinfo
self.args.test_skip = test_skip
def post_case(self):

98
tools/testing/selftests/tc-testing/creating-testcases/scapy-example.json Normal file
View File

@ -0,0 +1,98 @@
[
{
"id": "b1e9",
"name": "Test matching of source IP",
"category": [
"actions",
"scapy"
],
"plugins": {
"requires": [
"nsPlugin",
"scapyPlugin"
]
},
"setup": [
[
"$TC qdisc del dev $DEV1 ingress",
0,
1,
2,
255
],
"$TC qdisc add dev $DEV1 ingress"
],
"cmdUnderTest": "$TC filter add dev $DEV1 parent ffff: prio 3 protocol ip flower src_ip 16.61.16.61 flowid 1:1 action ok",
"scapy": {
"iface": "$DEV0",
"count": 1,
"packet": "Ether(type=0x800)/IP(src='16.61.16.61')/ICMP()"
},
"expExitCode": "0",
"verifyCmd": "$TC -s -j filter ls dev $DEV1 ingress prio 3",
"matchJSON": [
{
"path": [
1,
"options",
"actions",
0,
"stats",
"packets"
],
"value": 1
}
],
"teardown": [
"$TC qdisc del dev $DEV1 ingress"
]
},
{
"id": "e9c4",
"name": "Test matching of source IP with wrong count",
"category": [
"actions",
"scapy"
],
"plugins": {
"requires": [
"nsPlugin",
"scapyPlugin"
]
},
"setup": [
[
"$TC qdisc del dev $DEV1 ingress",
0,
1,
2,
255
],
"$TC qdisc add dev $DEV1 ingress"
],
"cmdUnderTest": "$TC filter add dev $DEV1 parent ffff: prio 3 protocol ip flower src_ip 16.61.16.61 flowid 1:1 action ok",
"scapy": {
"iface": "$DEV0",
"count": 3,
"packet": "Ether(type=0x800)/IP(src='16.61.16.61')/ICMP()"
},
"expExitCode": "0",
"verifyCmd": "$TC -s -j filter ls dev $DEV1 parent ffff:",
"matchJSON": [
{
"path": [
1,
"options",
"actions",
0,
"stats",
"packets"
],
"value": 1
}
],
"teardown": [
"$TC qdisc del dev $DEV1 ingress"
]
}
]

50
tools/testing/selftests/tc-testing/plugin-lib/scapyPlugin.py Normal file
View File

@ -0,0 +1,50 @@
#!/usr/bin/env python3
import os
import signal
from string import Template
import subprocess
import time
from TdcPlugin import TdcPlugin
from tdc_config import *
try:
from scapy.all import *
except ImportError:
print("Unable to import the scapy python module.")
print("\nIf not already installed, you may do so with:")
print("\t\tpip3 install scapy==2.4.2")
exit(1)
class SubPlugin(TdcPlugin):
def __init__(self):
self.sub_class = 'scapy/SubPlugin'
super().__init__()
def post_execute(self):
if 'scapy' not in self.args.caseinfo:
if self.args.verbose:
print('{}.post_execute: no scapy info in test case'.format(self.sub_class))
return
# Check for required fields
scapyinfo = self.args.caseinfo['scapy']
scapy_keys = ['iface', 'count', 'packet']
missing_keys = []
keyfail = False
for k in scapy_keys:
if k not in scapyinfo:
keyfail = True
missing_keys.add(k)
if keyfail:
print('{}: Scapy block present in the test, but is missing info:'
.format(self.sub_class))
print('{}'.format(missing_keys))
pkt = eval(scapyinfo['packet'])
if '$' in scapyinfo['iface']:
tpl = Template(scapyinfo['iface'])
scapyinfo['iface'] = tpl.safe_substitute(NAMES)
for count in range(scapyinfo['count']):
sendp(pkt, iface=scapyinfo['iface'])

10
tools/testing/selftests/tc-testing/tdc.py
View File

@ -122,15 +122,15 @@ class PluginMgr:
for pgn_inst in reversed(self.plugin_instances):
pgn_inst.post_suite(index)
def call_pre_case(self, testid, test_name, *, test_skip=False):
def call_pre_case(self, caseinfo, *, test_skip=False):
for pgn_inst in self.plugin_instances:
try:
pgn_inst.pre_case(testid, test_name, test_skip)
pgn_inst.pre_case(caseinfo, test_skip)
except Exception as ee:
print('exception {} in call to pre_case for {} plugin'.
format(ee, pgn_inst.__class__))
print('test_ordinal is {}'.format(test_ordinal))
print('testid is {}'.format(testid))
print('testid is {}'.format(caseinfo['id']))
raise
def call_post_case(self):
@ -261,14 +261,14 @@ def run_one_test(pm, args, index, tidx):
res = TestResult(tidx['id'], tidx['name'])
res.set_result(ResultState.skip)
res.set_errormsg('Test case designated as skipped.')
pm.call_pre_case(tidx['id'], tidx['name'], test_skip=True)
pm.call_pre_case(tidx, test_skip=True)
pm.call_post_execute()
return res
# populate NAMES with TESTID for this test
NAMES['TESTID'] = tidx['id']
pm.call_pre_case(tidx['id'], tidx['name'])
pm.call_pre_case(tidx)
prepare_env(args, pm, 'setup', "-----> prepare stage", tidx["setup"])
if (args.verbose > 0):