commit f8e075cdf7782ad5bee036274d51b43d99cf2f34 Author: Damian Johnson atagar@torproject.org Date: Tue Sep 22 16:46:47 2020 -0700
Test broken_listener example
Ooph! Despite being a tiny test for a tiny example this took me hours to get the mocks right. On reflection though this is pretty straight forward. is_alive() must be mocked so the 'with' clause doesn't attempt to connect to something. --- test/unit/examples.py | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-)
diff --git a/test/unit/examples.py b/test/unit/examples.py index 124ca3c7..2f664fd3 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -8,11 +8,14 @@ import os import sys import unittest
+import stem.socket import stem.util.system import test
+from stem.control import Controller from stem.descriptor.bandwidth_file import BandwidthFile -from unittest.mock import patch +from stem.response import ControlMessage +from unittest.mock import Mock, patch
EXAMPLE_DIR = os.path.join(test.STEM_BASE, 'docs', '_static', 'example') DESC_DIR = os.path.join(test.STEM_BASE, 'test', 'unit', 'descriptor', 'data') @@ -153,8 +156,23 @@ class TestExamples(unittest.TestCase): module.measure_fraction_relays_exit_80_microdescriptors(path) self.assertTrue(stdout_mock.getvalue().startswith(expected_prefix))
- def test_broken_listener(self): - pass + @patch('time.sleep') + @patch('stem.control.Controller.authenticate', Mock()) + @patch('stem.control.Controller.is_alive', Mock(return_value = True)) + @patch('stem.control.Controller.from_port') + @patch('sys.stdout', new_callable = io.StringIO) + def test_broken_listener(self, stdout_mock, from_port_mock, sleep_mock): + controller = Controller(stem.socket.ControlSocket()) + from_port_mock.return_value = controller + + # emits a BW event when the example runs time.sleep() + + bw_event = ControlMessage.from_str('650 BW 15 25', 'EVENT', normalize = True) + sleep_mock.side_effect = lambda duration: controller._handle_event(bw_event) + + import_example('broken_listener') + + self.assertEqual('start of broken_handler\n', stdout_mock.getvalue())
def test_check_digests(self): pass