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

   1 #!/usr/bin/env python3
   2
   3 import os
   4 import signal
   5 from string import Template
   6 import subprocess
   7 import time
   8 from TdcPlugin import TdcPlugin
   9
  10 from tdc_config import *
  11
  12 try:
  13     from scapy.all import *
  14 except ImportError:
  15     print("Unable to import the scapy python module.")
  16     print("\nIf not already installed, you may do so with:")
  17     print("\t\tpip3 install scapy==2.4.2")
  18     exit(1)
  19
  20 class SubPlugin(TdcPlugin):
  21     def __init__(self):
  22         self.sub_class = 'scapy/SubPlugin'
  23         super().__init__()
  24
  25     def post_execute(self):
  26         if 'scapy' not in self.args.caseinfo:
  27             if self.args.verbose:
  28                 print('{}.post_execute: no scapy info in test case'.format(self.sub_class))
  29             return
  30
  31         # Check for required fields
  32         scapyinfo = self.args.caseinfo['scapy']
  33         scapy_keys = ['iface', 'count', 'packet']
  34         missing_keys = []
  35         keyfail = False
  36         for k in scapy_keys:
  37             if k not in scapyinfo:
  38                 keyfail = True
  39                 missing_keys.add(k)
  40         if keyfail:
  41             print('{}: Scapy block present in the test, but is missing info:'
  42                 .format(self.sub_class))
  43             print('{}'.format(missing_keys))
  44
  45         pkt = eval(scapyinfo['packet'])
  46         if '$' in scapyinfo['iface']:
  47             tpl = Template(scapyinfo['iface'])
  48             scapyinfo['iface'] = tpl.safe_substitute(NAMES)
  49         for count in range(scapyinfo['count']):
  50             sendp(pkt, iface=scapyinfo['iface'])