mirror of
https://github.com/saymrwulf/uhd.git
synced 2026-05-16 21:10:10 +00:00
21ed72b5a2
Usually, devtest is run via make (or ninja), and will use the correct Python interpreter. When running directly on the command line, it is important to pick the right Python interpreter so it will work with the Python API. Here, we change the default interpreter from Python 2 to 3, because that's the more common version, and will be the only option for upcoming UHD 4.0 anyway.
257 lines
9.1 KiB
Python
Executable file
257 lines
9.1 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
#
|
|
# Copyright 2015-2016 Ettus Research LLC
|
|
# Copyright 2018 Ettus Research, a National Instruments Company
|
|
#
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
#
|
|
"""
|
|
Devtest: Base module. Provides classes for running devtest tests.
|
|
"""
|
|
|
|
from __future__ import print_function
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import re
|
|
import time
|
|
import logging
|
|
from subprocess import Popen, PIPE
|
|
import yaml
|
|
from six import iteritems
|
|
from usrp_probe import get_usrp_list
|
|
|
|
#--------------------------------------------------------------------------
|
|
# Helpers
|
|
#--------------------------------------------------------------------------
|
|
def filter_warnings(errstr):
|
|
"""
|
|
Searches errstr for UHD warnings, removes them, and puts them into a
|
|
separate string.
|
|
Returns (errstr, warnstr), where errstr no longer has warnings. """
|
|
warn_re = re.compile("UHD Warning:\n(?: .*\n)+")
|
|
warnstr = "\n".join(warn_re.findall(errstr)).strip()
|
|
errstr = warn_re.sub('', errstr).strip()
|
|
return (errstr, warnstr)
|
|
|
|
def filter_stderr(stderr, run_results=None):
|
|
"""
|
|
Filters the output to stderr. run_results[] is a dictionary.
|
|
This function will:
|
|
- Remove warnings and put them in run_results['warnings']
|
|
- Put the filtered error string into run_results['errors'] and returns the dictionary
|
|
"""
|
|
run_results = run_results or {}
|
|
errstr, run_results['warnings'] = filter_warnings(stderr)
|
|
# Scan for underruns and sequence errors / dropped packets not detected in the counter
|
|
errstr = re.sub("\n\n+", "\n", errstr)
|
|
run_results['errors'] = errstr.strip()
|
|
return run_results
|
|
|
|
#--------------------------------------------------------------------------
|
|
# Application
|
|
#--------------------------------------------------------------------------
|
|
class shell_application(object):
|
|
"""
|
|
Wrapper for applications that are in $PATH.
|
|
Note: The CMake infrastructure makes sure all examples and utils are in $PATH.
|
|
"""
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.stdout = ''
|
|
self.stderr = ''
|
|
self.returncode = None
|
|
self.exec_time = None
|
|
|
|
def run(self, args=None):
|
|
"""Test executor."""
|
|
args = args or []
|
|
cmd_line = [self.name]
|
|
cmd_line.extend(args)
|
|
start_time = time.time()
|
|
env = os.environ
|
|
env["UHD_LOG_FASTPATH_DISABLE"] = "1"
|
|
try:
|
|
proc = Popen(
|
|
cmd_line,
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
close_fds=True,
|
|
env=env,
|
|
universal_newlines=True
|
|
)
|
|
self.stdout, self.stderr = proc.communicate()
|
|
self.returncode = proc.returncode
|
|
self.exec_time = time.time() - start_time
|
|
except OSError as ex:
|
|
raise RuntimeError("Failed to execute command: `{}'\n{}"
|
|
.format(cmd_line, str(ex)))
|
|
|
|
|
|
#--------------------------------------------------------------------------
|
|
# Test case base
|
|
#--------------------------------------------------------------------------
|
|
class uhd_test_case(unittest.TestCase):
|
|
"""
|
|
Base class for UHD test cases.
|
|
"""
|
|
test_name = '--TEST--'
|
|
|
|
def set_up(self):
|
|
"""
|
|
Override this to add own setup code per test.
|
|
"""
|
|
pass
|
|
|
|
def setUp(self):
|
|
self.name = self.__class__.__name__
|
|
self.test_id = self.id().split('.')[-1]
|
|
self.results = {}
|
|
self.results_file = os.getenv('_UHD_TEST_RESULTSFILE', "")
|
|
if self.results_file and os.path.isfile(self.results_file):
|
|
self.results = yaml.safe_load(open(self.results_file).read()) or {}
|
|
self.args_str = os.getenv('_UHD_TEST_ARGS_STR', "")
|
|
self.usrp_info = get_usrp_list(self.args_str)[0]
|
|
if self.usrp_info['serial'] not in self.results:
|
|
self.results[self.usrp_info['serial']] = {}
|
|
if self.name not in self.results[self.usrp_info['serial']]:
|
|
self.results[self.usrp_info['serial']][self.name] = {}
|
|
self.setup_logger()
|
|
self.set_up()
|
|
|
|
def setup_logger(self):
|
|
" Add logging infrastructure "
|
|
self.log = logging.getLogger("devtest.{name}".format(name=self.name))
|
|
self.log_file = os.getenv('_UHD_TEST_LOGFILE', "devtest.log")
|
|
#self.log_level = int(os.getenv('_UHD_TEST_LOG_LEVEL', logging.DEBUG))
|
|
#self.print_level = int(os.getenv('_UHD_TEST_PRINT_LEVEL', logging.WARNING))
|
|
self.log_level = logging.DEBUG
|
|
self.print_level = logging.WARNING
|
|
file_handler = logging.FileHandler(self.log_file)
|
|
file_handler.setLevel(self.log_level)
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(self.print_level)
|
|
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
file_handler.setFormatter(formatter)
|
|
console_handler.setFormatter(formatter)
|
|
self.log.setLevel(logging.DEBUG)
|
|
self.log.addHandler(file_handler)
|
|
self.log.addHandler(console_handler)
|
|
self.log.info("Starting test with device: %s", str(self.args_str))
|
|
|
|
def tear_down(self):
|
|
"""Nothing to do."""
|
|
pass
|
|
|
|
def tearDown(self):
|
|
self.tear_down()
|
|
if self.results_file:
|
|
open(self.results_file, 'w').write(
|
|
yaml.dump(self.results, default_flow_style=False))
|
|
time.sleep(15)
|
|
|
|
def report_result(self, testname, key, value):
|
|
""" Store a result as a key/value pair.
|
|
After completion, all results for one test are written to the results file.
|
|
"""
|
|
if not testname in self.results[self.usrp_info['serial']][self.name]:
|
|
self.results[self.usrp_info['serial']][self.name][testname] = {}
|
|
self.results[self.usrp_info['serial']][self.name][testname][key] = value
|
|
|
|
def create_addr_args_str(self, argname="args"):
|
|
""" Returns an args string, usually '--args "type=XXX,serial=YYY" """
|
|
if len(self.args_str) == 0:
|
|
return ''
|
|
return '--{}={}'.format(argname, self.args_str)
|
|
|
|
class uhd_example_test_case(uhd_test_case):
|
|
"""
|
|
A test case that runs an example.
|
|
"""
|
|
|
|
def setup_example(self):
|
|
"""
|
|
Override this to add specific setup code.
|
|
"""
|
|
pass
|
|
|
|
def set_up(self):
|
|
"""Called by the unit testing framework on tests. """
|
|
self.setup_example()
|
|
|
|
def run_test(self, test_name, test_args):
|
|
"""
|
|
Override this to run the actual example.
|
|
|
|
Needs to return either a boolean or a dict with key 'passed' to determine
|
|
pass/fail.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def run_example(self, example, args):
|
|
"""
|
|
Run `example' (which has to be a UHD example or utility) with `args'.
|
|
Return results and the app object.
|
|
|
|
Note: UHD_LOG_FASTPATH_DISABLE will be set to 1.
|
|
"""
|
|
self.log.info("Running example: `%s %s'", example, " ".join(args))
|
|
app = shell_application(example)
|
|
app.run(args)
|
|
run_results = {
|
|
'return_code': app.returncode,
|
|
'passed': False,
|
|
}
|
|
run_results = filter_stderr(app.stderr, run_results)
|
|
self.log.info('STDERR Output:')
|
|
self.log.info(str(app.stderr))
|
|
return (app, run_results)
|
|
|
|
|
|
def report_example_results(self, test_name, run_results):
|
|
for key in sorted(run_results):
|
|
self.log.info('%s = %s', str(key), str(run_results[key]))
|
|
self.report_result(
|
|
test_name,
|
|
key, run_results[key]
|
|
)
|
|
if 'passed' in run_results:
|
|
self.report_result(
|
|
test_name,
|
|
'status',
|
|
'Passed' if run_results['passed'] else 'Failed',
|
|
)
|
|
if 'errors' in run_results:
|
|
self.report_result(
|
|
test_name,
|
|
'errors',
|
|
'Yes' if run_results['errors'] else 'No',
|
|
)
|
|
|
|
def test_all(self):
|
|
"""
|
|
Hook for test runner. Needs to be a class method that starts with 'test'.
|
|
Calls run_test().
|
|
"""
|
|
for test_name, test_args in iteritems(self.test_params):
|
|
time.sleep(15) # Wait for X300 devices to reclaim them
|
|
if not 'products' in test_args \
|
|
or (self.usrp_info['product'] in test_args.get('products', [])):
|
|
run_results = self.run_test(test_name, test_args)
|
|
passed = bool(run_results)
|
|
if isinstance(run_results, dict):
|
|
passed = run_results['passed']
|
|
errors = run_results.pop("errors", None)
|
|
if not passed:
|
|
print("Error log:", file=sys.stderr)
|
|
print(errors)
|
|
self.assertTrue(
|
|
passed,
|
|
msg="Errors occurred during test `{t}'. "
|
|
"Check log file for details.\n"
|
|
"Run results:\n{r}".format(
|
|
t=test_name,
|
|
r=yaml.dump(run_results, default_flow_style=False)
|
|
)
|
|
)
|
|
|