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
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 = {}
        self.plugin_instances = []
        self.failed_plugins = {}
        self.argparser = argparser

        # TODO, put plugins in order
        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[mn] = foo
                    self.plugin_instances.append(foo.SubPlugin())

    def load_plugin(self, pgdir, pgname):
        pgname = pgname[0:-3]
        foo = importlib.import_module('{}.{}'.format(pgdir, pgname))
        self.plugins[pgname] = foo
        self.plugin_instances.append(foo.SubPlugin())
        self.plugin_instances[-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 = []
        for t in testlist:
            try:
                if 'requires' in t['plugins']:
                    if isinstance(t['plugins']['requires'], list):
                        reqs.extend(t['plugins']['requires'])
                    else:
                        reqs.append(t['plugins']['requires'])
            except KeyError:
                continue
        reqs = get_unique_item(reqs)
        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_inst in self.plugin_instances:
            try:
                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(caseinfo['id']))
                raise

    def call_post_case(self):
        for pgn_inst in reversed(self.plugin_instances):
            pgn_inst.post_case()

    def call_pre_execute(self):
        for pgn_inst in self.plugin_instances:
            pgn_inst.pre_execute()

    def call_post_execute(self):
        for pgn_inst in reversed(self.plugin_instances):
            pgn_inst.post_execute()

    def call_add_args(self, parser):
        for pgn_inst in self.plugin_instances:
            parser = pgn_inst.add_args(parser)
        return parser

    def call_check_args(self, args, remaining):
        for pgn_inst in self.plugin_instances:
            pgn_inst.check_args(args, remaining)

    def call_adjust_command(self, stage, command):
        for pgn_inst in self.plugin_instances:
            command = pgn_inst.adjust_command(stage, command)
        return command

    def set_args(self, args):
        self.args = args

    @staticmethod