Merge pull request #824 from tlsfuzzer/analysis-fixes

Analysis fixes

docs/source/tlsfuzzer.utils.progress_report.rst

@@ -0,0 +1,7 @@
+tlsfuzzer.utils.progress\_report module
+=======================================
+
+.. automodule:: tlsfuzzer.utils.progress_report
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/source/tlsfuzzer.utils.rst

@@ -13,4 +13,5 @@ Submodules
 
    tlsfuzzer.utils.lists
    tlsfuzzer.utils.ordered_dict
+   tlsfuzzer.utils.progress_report
 

scripts/test-bleichenbacher-timing-pregenerate.py

@@ -1388,15 +1388,15 @@ def main():
             if not timing_runner.extract():
                 ret_val = 2
             else:
-                timing_runner.analyse()
+                ret_val = timing_runner.analyse()
 
             if ret_val == 0:
                 print("No statistically significant difference detected")
             elif ret_val == 1:
-                print("Statisticaly significant difference detected at alpha="
-                      "0.05")
+                print("Statisticaly significant difference detected")
             else:
-                print("Statistical analysis exited with {0}".format(ret_val))
+                print("Error: Statistical analysis exited with {0}".format(ret_val))
+            sys.exit(ret_val)
         else:
             print("Could not run timing tests because tcpdump is not present!")
             sys.exit(1)

tests/test_tlsfuzzer_analysis.py

@@ -610,6 +610,8 @@ def test_conf_interval_plot(self, print_fun):
                                bbox_inches='tight'),
                      mock.call('/tmp/conf_interval_plot_trim_mean_25.png',
                                bbox_inches='tight'),
+                     mock.call('/tmp/conf_interval_plot_trim_mean_45.png',
+                               bbox_inches='tight'),
                      mock.call('/tmp/conf_interval_plot_trimean.png',
                                bbox_inches='tight')])
 
@@ -708,7 +710,20 @@ def test_command_line(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, True, True, True, False, False, None, None)
+                        output, True, True, True, False, False, None, None, None)
+
+    def test_call_with_workers(self):
+        output = "/tmp"
+        args = ["analysis.py", "-o", output, '--workers', '200']
+        mock_init = mock.Mock()
+        mock_init.return_value = None
+        with mock.patch('tlsfuzzer.analysis.Analysis.generate_report') as mock_report:
+            with mock.patch('tlsfuzzer.analysis.Analysis.__init__', mock_init):
+                with mock.patch("sys.argv", args):
+                    main()
+                    mock_report.assert_called_once()
+                    mock_init.assert_called_once_with(
+                        output, True, True, True, False, False, None, None, 200)
 
     def test_call_with_verbose(self):
         output = "/tmp"
@@ -721,7 +736,7 @@ def test_call_with_verbose(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, True, True, True, False, True, None, None)
+                        output, True, True, True, False, True, None, None, None)
 
     def test_call_with_multithreaded_plots(self):
         output = "/tmp"
@@ -734,7 +749,7 @@ def test_call_with_multithreaded_plots(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, True, True, True, True, False, None, None)
+                        output, True, True, True, True, False, None, None, None)
 
     def test_call_with_no_plots(self):
         output = "/tmp"
@@ -748,7 +763,7 @@ def test_call_with_no_plots(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, False, False, False, False, False, None, None)
+                        output, False, False, False, False, False, None, None, None)
 
     def test_call_with_frequency(self):
         output = "/tmp"
@@ -761,7 +776,7 @@ def test_call_with_frequency(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, True, True, True, False, False, 10*1e6, None)
+                        output, True, True, True, False, False, 10*1e6, None, None)
 
     def test_call_with_alpha(self):
         output = "/tmp"
@@ -774,7 +789,7 @@ def test_call_with_alpha(self):
                     main()
                     mock_report.assert_called_once()
                     mock_init.assert_called_once_with(
-                        output, True, True, True, False, False, None, 1e-3)
+                        output, True, True, True, False, False, None, 1e-3, None)
 
     def test_help(self):
         args = ["analysis.py", "--help"]

tests/test_tlsfuzzer_utils_progress_report.py

@@ -0,0 +1,112 @@
+# Author: Hubert Kario, (c) 2023
+# Released under Gnu GPL v2.0, see LICENSE file for details
+
+from __future__ import print_function
+from threading import Event, Thread
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+try:
+    import mock
+except ImportError:
+    import unittest.mock as mock
+import sys
+import time
+
+from tlsfuzzer.utils.progress_report import _format_seconds, _binary_prefix, \
+        progress_report
+
+if sys.version_info < (3, 0):
+    BUILTIN_PRINT = "__builtin__.print"
+else:
+    BUILTIN_PRINT = "builtins.print"
+
+
+class TestFormatSeconds(unittest.TestCase):
+    def test_days(self):
+        self.assertEqual(_format_seconds(60 * 60 * 24 * 3),
+                          " 3d  0h  0m  0.00s")
+
+    def test_hours(self):
+        self.assertEqual(_format_seconds(60 * 60 * 4),
+                          " 4h  0m  0.00s")
+
+    def test_minutes(self):
+        self.assertEqual(_format_seconds(60 * 15), "15m  0.00s")
+
+    def test_all(self):
+        self.assertEqual(_format_seconds(
+            60 * 60 * 24 * 4 +
+            60 * 60 * 5 +
+            60 * 14 +
+            7), " 4d  5h 14m  7.00s")
+
+
+class TestBinaryPrefix(unittest.TestCase):
+    def test_bytes(self):
+        self.assertEqual(_binary_prefix(12), "12.00")
+
+    def test_kilobytes(self):
+        self.assertEqual(_binary_prefix(1024*12), "12.00ki")
+
+
+class TestInvalidInputs(unittest.TestCase):
+    def test_wrong_status(self):
+        with self.assertRaises(ValueError) as e:
+            progress_report([0, 1, 2, 3])
+
+        self.assertIn("status is not a 3 element", str(e.exception))
+
+    def test_wrong_prefix(self):
+        with self.assertRaises(AssertionError):
+            progress_report([0, 1, 2], prefix="none")
+
+
+class TestOperation(unittest.TestCase):
+    @mock.patch(BUILTIN_PRINT)
+    def test_with_binary_prefix(self, mock_print):
+        status = [10000, 100000, Event()]
+        params = {'prefix': 'binary', 'delay': 0.001}
+        progress = Thread(target=progress_report, args=(status,),
+                          kwargs=params)
+        progress.start()
+        while not mock_print.mock_calls:
+            pass
+        status[2].set()
+        progress.join()
+        self.assertIn('Done:  10.00%', str(mock_print.mock_calls[0]))
+
+    @mock.patch(BUILTIN_PRINT)
+    def test_with_bool(self, mock_print):
+        status = [200, 1000, True]
+        params = {'delay': 0.001}
+        progress = Thread(target=progress_report, args=(status,),
+                          kwargs=params)
+        progress.start()
+        while not mock_print.mock_calls:
+            pass
+        status[0] = 1000
+        while len(mock_print.mock_calls) < 2:
+            pass
+        status[2] = False
+        progress.join()
+        self.assertIn('Done:  20.00%', str(mock_print.mock_calls[0]))
+        self.assertIn('Done: 100.00%', str(mock_print.mock_calls[-1]))
+
+    @mock.patch(BUILTIN_PRINT)
+    def test_with_zero_start(self, mock_print):
+        status = [0, 1000, True]
+        params = {'delay': 0.001}
+        progress = Thread(target=progress_report, args=(status,),
+                          kwargs=params)
+        progress.start()
+        while not mock_print.mock_calls:
+            pass
+        status[0] = 1000
+        while len(mock_print.mock_calls) < 2:
+            pass
+        status[2] = False
+        progress.join()
+        self.assertIn('Done:   0.00%', str(mock_print.mock_calls[0]))
+        self.assertIn('Done: 100.00%', str(mock_print.mock_calls[-1]))