path: root/tools/testing/selftests/tc-testing/tdc.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0

"""
tdc.py - Linux tc (Traffic Control) unit test driver

Copyright (C) 2017 Lucas Bates <lucasb@mojatatu.com>
"""

import re
import os
import sys
import argparse
import importlib
import json
import subprocess
import time
import traceback
import random
from multiprocessing import Pool
from collections import OrderedDict
from string import Template

from tdc_config import *
from tdc_helper import *

import TdcPlugin
from TdcResults import *

class PluginDependencyException(Exception):
    def __init__(self, missing_pg):
        self.missing_pg = missing_pg

class PluginMgrTestFail(Exception):
    def __init__(self, stage, output, message):
        self.stage = stage
        self.output = output
        self.message = message

class PluginMgr:
    def __init__(self, argparser):
        super().__init__()
        self.plugins = set()
        self.plugin_instances = []
        self.failed_plugins = {}
        self.argparser = argparser

        plugindir = os.getenv('TDC_PLUGIN_DIR', './plugins')
        for dirpath, dirnames, filenames in os.walk(plugindir):
            for fn in filenames:
                if (fn.endswith('.py') and
                    not fn == '__init__.py' and
                    not fn.startswith('#') and
                    not fn.startswith('.#')):
                    mn = fn[0:-3]
                    foo = importlib.import_module('plugins.' + mn)
                    self.plugins.add(mn)
                    self.plugin_instances[mn] = foo.SubPlugin()

    def load_plugin(self, pgdir, pgname):
        pgname = pgname[0:-3]
        self.plugins.add(pgname)

        foo = importlib.import_module('{}.{}'.format(pgdir, pgname))

        # nsPlugin must always be the first one
        if pgname == "nsPlugin":
            self.plugin_instances.insert(0, (pgname, foo.SubPlugin()))
            self.plugin_instances[0][1].check_args(self.args, None)
        else:
            self.plugin_instances.append((pgname, foo.SubPlugin()))
            self.plugin_instances[-1][1].check_args(self.args, None)

    def get_required_plugins(self, testlist):
        '''
        Get all required plugins from the list of test cases and return
        all unique items.
        '''
        reqs = set()
        for t in testlist:
            try:
                if 'requires' in t['plugins']:
                    if isinstance(t['plugins']['requires'], list):
                        reqs.update(set(t['plugins']['requires']))
                    else:
                        reqs.add(t['plugins']['requires'])
                    t['plugins'] = t['plugins']['requires']
                else:
                    t['plugins'] = []
            except KeyError:
                t['plugins'] = []
                continue

        return reqs

    def load_required_plugins(self, reqs, parser, args, remaining):
        '''
        Get all required plugins from the list of test cases and load any plugin
        that is not already enabled.
        '''
        pgd = ['plugin-lib', 'plugin-lib-custom']
        pnf = []

        for r in reqs:
            if r not in self.plugins:
                fname = '{}.py'.format(r)
                source_path = []
                for d in pgd:
                    pgpath = '{}/{}'.format(d, fname)
                    if os.path.isfile(pgpath):
                        source_path.append(pgpath)
                if len(source_path) == 0:
                    print('ERROR: unable to find required plugin {}'.format(r))
                    pnf.append(fname)
                    continue
                elif len(source_path) > 1:
                    print('WARNING: multiple copies of plugin {} found, using version found')
                    print('at {}'.format(source_path[0]))
                pgdir = source_path[0]
                pgdir = pgdir.split('/')[0]
                self.load_plugin(pgdir, fname)
        if len(pnf) > 0:
            raise PluginDependencyException(pnf)

        parser = self.call_add_args(parser)
        (args, remaining) = parser.parse_known_args(args=remaining, namespace=args)
        return args

    def call_pre_suite(self, testcount, testidlist):
        for (_, pgn_inst) in self.plugin_instances:
            pgn_inst.pre_suite(testcount, testidlist)

    def call_post_suite(self, index):
        for (_, pgn_inst) in reversed(self.plugin_instances):
            pgn_inst.post_suite(index)

    def call_pre_case(self, caseinfo, *, test_skip=False):
        for (pgn, pgn_inst) in self.plugin_instances:
            if pgn not in caseinfo[