devtest: Clean up & refactor

- Move filter_* functions out of uhd_test_case
- Reduced some line lengths

2 files changed, 55 insertions, 35 deletions

diff --git a/host/tests/devtest/benchmark_rate_test.py b/host/tests/devtest/benchmark_rate_test.py
index 3d8ba3938..39d8bd483 100755
--- a/host/tests/devtest/benchmark_rate_test.py
+++ b/host/tests/devtest/benchmark_rate_test.py
@@ -54,7 +54,11 @@ class uhd_benchmark_rate_test(uhd_example_test_case):
         match = re.search(r'(Num received samples):\s*(.*)', app.stdout)
         run_results['num_rx_samples'] = int(match.group(2)) if match else -1
         if run_results['num_rx_samples'] != -1:
-            run_results['rel_rx_samples_error'] = 1.0 * abs(run_results['num_rx_samples'] - test_args.get('rx_buffer',0) - expected_samples) / expected_samples
+            run_results['rel_rx_samples_error'] = 1.0 * abs(
+                run_results['num_rx_samples']
+                - test_args.get('rx_buffer', 0)
+                - expected_samples
+            ) / expected_samples
         else:
             run_results['rel_rx_samples_error'] = 100
         match = re.search(r'(Num dropped samples):\s*(.*)', app.stdout)
@@ -64,7 +68,11 @@ class uhd_benchmark_rate_test(uhd_example_test_case):
         match = re.search(r'(Num transmitted samples):\s*(.*)', app.stdout)
         run_results['num_tx_samples'] = int(match.group(2)) if match else -1
         if run_results['num_tx_samples'] != -1:
-            run_results['rel_tx_samples_error'] = 1.0 * abs(run_results['num_tx_samples'] - test_args.get('tx_buffer',0) - expected_samples) / expected_samples
+            run_results['rel_tx_samples_error'] = 1.0 * abs(
+                run_results['num_tx_samples']
+                - test_args.get('tx_buffer', 0)
+                - expected_samples
+            ) / expected_samples
         else:
             run_results['rel_tx_samples_error'] = 100
         match = re.search(r'(Num sequence errors \(Tx\)):\s*(.*)', app.stdout)
@@ -86,3 +94,4 @@ class uhd_benchmark_rate_test(uhd_example_test_case):
         ])
         self.report_example_results(test_name, run_results)
         return run_results
+
diff --git a/host/tests/devtest/uhd_test_base.py b/host/tests/devtest/uhd_test_base.py
index 0a8f7f2af..2bee86941 100755
--- a/host/tests/devtest/uhd_test_base.py
+++ b/host/tests/devtest/uhd_test_base.py
@@ -5,6 +5,9 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 #
+"""
+Devtest: Base module. Provides classes for running devtest tests.
+"""
 
 from __future__ import print_function
 import os
@@ -19,6 +22,43 @@ 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 UUUUU... strings, since they are generally not a problem.
+    - Remove all DDDD and SSSS strings, and add run_results['has_S'] = True
+      and run_results['has_D'] = True.
+    - 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('UU+', '', errstr)
+    (errstr, n_subs) = re.subn('SS+', '', errstr)
+    if n_subs:
+        run_results['has_S'] = True
+    (errstr, n_subs) = re.subn('DD+', '', errstr)
+    if n_subs:
+        run_results['has_D'] = True
+    errstr = re.sub("\n\n+", "\n", errstr)
+    run_results['errors'] = errstr.strip()
+    return run_results
+
+#--------------------------------------------------------------------------
 # Application
 #--------------------------------------------------------------------------
 class shell_application(object):
@@ -102,8 +142,10 @@ class uhd_test_case(unittest.TestCase):
     def tearDown(self):
         self.tear_down()
         if self.results_file:
-            open(self.results_file, 'w').write(yaml.dump(self.results, default_flow_style=False))
+            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.
@@ -118,37 +160,6 @@ class uhd_test_case(unittest.TestCase):
             return ''
         return '--{}={}'.format(argname, self.args_str)
 
-    def filter_warnings(self, errstr):
-        """ Searches errstr for UHD warnings, removes them, and puts them into a separate string.
-        Returns (errstr, warnstr), where errstr no longer has warning. """
-        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(self, stderr, run_results=None):
-        """ Filters the output to stderr. run_results[] is a dictionary.
-        This function will:
-        - Remove UUUUU... strings, since they are generally not a problem.
-        - Remove all DDDD and SSSS strings, and add run_results['has_S'] = True
-          and run_results['has_D'] = True.
-        - 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'] = self.filter_warnings(stderr)
-        # Scan for underruns and sequence errors / dropped packets  not detected in the counter
-        errstr = re.sub('UU+', '', errstr)
-        (errstr, n_subs) = re.subn('SS+', '', errstr)
-        if n_subs:
-            run_results['has_S'] = True
-        (errstr, n_subs) = re.subn('DD+', '', errstr)
-        if n_subs:
-            run_results['has_D'] = True
-        errstr = re.sub("\n\n+", "\n", errstr)
-        run_results['errors'] = errstr.strip()
-        return run_results
-
 class uhd_example_test_case(uhd_test_case):
     """
     A test case that runs an example.
@@ -187,7 +198,7 @@ class uhd_example_test_case(uhd_test_case):
             'has_D': False,
             'has_S': False,
         }
-        run_results = self.filter_stderr(app.stderr, run_results)
+        run_results = filter_stderr(app.stderr, run_results)
         self.log.info('STDERR Output:')
         self.log.info(str(app.stderr))
         return (app, run_results)