October 2020 - tor-commits - lists.torproject.org

[stem/master] Test ephemeral_hidden_services example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 31ec07e1bfd259063b5408683b67e2dba8aa674d Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 13:43:48 2020 -0700 Test ephemeral_hidden_services example --- test/unit/examples.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/test/unit/examples.py b/test/unit/examples.py index eb9bdb2a..35aa80bb 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -127,6 +127,12 @@ Downloading server descriptor from 1.2.3.4:443... router caerSidi 71.35.133.197 9001 0 0 """ +EXPECTED_EPHEMEREAL_HIDDEN_SERVICES = """\ + * Connecting to tor + * Our service is available at my-service.onion, press ctrl+c to quit + * Shutting down our hidden service +""" + EXPECTED_EXIT_USED = """\ Tracking requests for tor exits. Press 'enter' to end. @@ -515,8 +521,23 @@ class TestExamples(unittest.TestCase): download_descriptor.main(['--dirport', '1.2.3.4:443']) self.assertTrue(stdout_mock.getvalue().startswith(EXPECTED_DOWNLOAD_DESCRIPTOR_PREFIX)) - def test_ephemeral_hidden_services(self): - pass + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_ephemeral_hidden_services(self, stdout_mock, from_port_mock): + original_modules = dict(sys.modules) + + try: + sys.modules['flask'] = Mock() + + controller = from_port_mock().__enter__() + hidden_service = controller.create_ephemeral_hidden_service() + hidden_service.service_id = 'my-service' + + import ephemeral_hidden_services + + self.assertEqual(EXPECTED_EPHEMEREAL_HIDDEN_SERVICES, stdout_mock.getvalue()) + finally: + sys.modules = original_modules def test_event_listening(self): pass

1 0

[stem/master] Test fibonacci examples
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 68ab78c21c15461936050bfaf1908abf033de85d Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 14:58:44 2020 -0700 Test fibonacci examples --- docs/_static/example/fibonacci_multiprocessing.py | 18 ++++++++++------- docs/_static/example/fibonacci_threaded.py | 24 +++++++++++++---------- test/unit/examples.py | 21 ++++++++++++++++---- 3 files changed, 42 insertions(+), 21 deletions(-) diff --git a/docs/_static/example/fibonacci_multiprocessing.py b/docs/_static/example/fibonacci_multiprocessing.py index e909e169..4ad3fe48 100644 --- a/docs/_static/example/fibonacci_multiprocessing.py +++ b/docs/_static/example/fibonacci_multiprocessing.py @@ -7,14 +7,18 @@ def fibonacci(n): else: return fibonacci(n-2) + fibonacci(n-1) -# calculate fibonacci sequences four times in parallel +def main(): + # calculate fibonacci sequences four times in parallel -start_time, threads = time.time(), [] + start_time, threads = time.time(), [] -for i in range(4): - threads.append(stem.util.system.DaemonTask(fibonacci, (35,), start = True)) + for i in range(4): + threads.append(stem.util.system.DaemonTask(fibonacci, (35,), start = True)) -for t in threads: - t.join() + for t in threads: + t.join() -print('took %0.1f seconds' % (time.time() - start_time)) + print('took %0.1f seconds' % (time.time() - start_time)) + +if __name__ == '__main__': + main() diff --git a/docs/_static/example/fibonacci_threaded.py b/docs/_static/example/fibonacci_threaded.py index f40b7401..0e449122 100644 --- a/docs/_static/example/fibonacci_threaded.py +++ b/docs/_static/example/fibonacci_threaded.py @@ -7,18 +7,22 @@ def fibonacci(n): else: return fibonacci(n-2) + fibonacci(n-1) -# calculate fibonacci sequences four times in parallel +def main(): + # calculate fibonacci sequences four times in parallel -start_time, threads = time.time(), [] + start_time, threads = time.time(), [] -for i in range(4): - t = threading.Thread(target = fibonacci, args = (35,)) - t.setDaemon(True) - t.start() + for i in range(4): + t = threading.Thread(target = fibonacci, args = (35,)) + t.setDaemon(True) + t.start() - threads.append(t) + threads.append(t) -for t in threads: - t.join() + for t in threads: + t.join() -print('took %0.1f seconds' % (time.time() - start_time)) + print('took %0.1f seconds' % (time.time() - start_time)) + +if __name__ == '__main__': + main() diff --git a/test/unit/examples.py b/test/unit/examples.py index d9a3f6fa..07b6ec00 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -578,11 +578,24 @@ class TestExamples(unittest.TestCase): self.assertEqual(EXPECTED_EXIT_USED, stdout_mock.getvalue()) - def test_fibonacci_multiprocessing(self): - pass + @patch('sys.stdout', new_callable = io.StringIO) + def test_fibonacci_multiprocessing(self, stdout_mock): + # This example intentionally takes a long time (~11 seconds), so replacing + # the work it does with a no-op. - def test_fibonacci_threaded(self): - pass + with patch('fibonacci_multiprocessing.fibonacci', Mock(return_value = 5)): + import fibonacci_multiprocessing + + fibonacci_multiprocessing.main() + self.assertEqual('took 0.0 seconds\n', stdout_mock.getvalue()) + + @patch('sys.stdout', new_callable = io.StringIO) + def test_fibonacci_threaded(self, stdout_mock): + with patch('fibonacci_threaded.fibonacci', Mock(return_value = 5)): + import fibonacci_threaded + + fibonacci_threaded.main() + self.assertEqual('took 0.0 seconds\n', stdout_mock.getvalue()) def test_get_hidden_service_descriptor(self): pass

1 0

[stem/master] Test get_hidden_service_descriptor example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit ae6a045d7808800451e1372f10e90e8e5b063bd8 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 15:02:59 2020 -0700 Test get_hidden_service_descriptor example --- test/unit/examples.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/test/unit/examples.py b/test/unit/examples.py index 07b6ec00..df46a30c 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -18,6 +18,7 @@ import test.require from stem.control import Controller from stem.descriptor.bandwidth_file import BandwidthFile from stem.descriptor.extrainfo_descriptor import RelayExtraInfoDescriptor +from stem.descriptor.hidden_service import HiddenServiceDescriptorV2 from stem.descriptor.networkstatus import NetworkStatusDocumentV3 from stem.descriptor.router_status_entry import RouterStatusEntryV2, RouterStatusEntryV3 from stem.descriptor.server_descriptor import RelayDescriptor @@ -597,8 +598,15 @@ class TestExamples(unittest.TestCase): fibonacci_threaded.main() self.assertEqual('took 0.0 seconds\n', stdout_mock.getvalue()) - def test_get_hidden_service_descriptor(self): - pass + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_get_hidden_service_descriptor(self, stdout_mock, from_port_mock): + controller = from_port_mock().__enter__() + controller.get_hidden_service_descriptor.return_value = HiddenServiceDescriptorV2.create() + + import get_hidden_service_descriptor + + self.assertTrue(stdout_mock.getvalue().startswith('rendezvous-service-descriptor ')) @patch('stem.control.Controller.from_port', spec = Controller) @patch('sys.stdout', new_callable = io.StringIO)

1 0

[stem/master] Drop unused load_test example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 26d6f781bf97872645067096d272949dcfd4c3d9 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 15:12:52 2020 -0700 Drop unused load_test example Odd, commit 843facd added this example but didn't link to it within our website. Rather than fix the trivial print() bug and test it simply removing. --- docs/_static/example/load_test.py | 25 ------------------------- test/unit/examples.py | 3 --- 2 files changed, 28 deletions(-) diff --git a/docs/_static/example/load_test.py b/docs/_static/example/load_test.py deleted file mode 100644 index e98e4c00..00000000 --- a/docs/_static/example/load_test.py +++ /dev/null @@ -1,25 +0,0 @@ -import os -import time - -import stem.control -import stem.util.proc -import stem.util.str_tools - -start_time = time.time() -samplings = [] -last_sample = None - -with stem.control.Controller.from_port() as controller: - controller.authenticate() - controller.add_event_listener(lambda *args: None, 'DEBUG') - - while True: - utime, stime = stem.util.proc.stats(os.getpid(), stem.util.proc.Stat.CPU_UTIME, stem.util.proc.Stat.CPU_STIME) - total_cpu_time = float(utime) + float(stime) - - if last_sample: - samplings.append(total_cpu_time - last_sample) - print '%0.1f%% (%s)' % (sum(samplings) / len(samplings) * 100, stem.util.str_tools.time_label(time.time() - start_time)) - - last_sample = total_cpu_time - time.sleep(1) diff --git a/test/unit/examples.py b/test/unit/examples.py index b6eb8016..094322e6 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -673,9 +673,6 @@ class TestExamples(unittest.TestCase): self.assertEqual(EXPECTED_LIST_CIRCUITS, stdout_mock.getvalue()) - def test_load_test(self): - pass - def test_manual_config_options(self): pass

1 0

[stem/master] Test introduction_points example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 9f12b49dba84bc4232b0ea03842a4e8e2142a880 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 15:09:38 2020 -0700 Test introduction_points example --- test/unit/examples.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/test/unit/examples.py b/test/unit/examples.py index df46a30c..b6eb8016 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -145,6 +145,14 @@ Exit relay for our connection to 64.15.112.44:80 """ +EXPECTED_INTRODUCTION_POINTS = """\ +DuckDuckGo's introduction points are... + + 178.62.222.129:443 => iwki77xtbvp6qvedfrwdzncxs3ckayeu + 46.4.174.52:443 => em4gjk6eiiualhmlyiifrzc7lbtrsbip + 62.210.82.169:443 => jqhfl364x3upe6lqnxizolewlfrsw2zy +""" + EXPECTED_LIST_CIRCUITS = """\ Circuit 4 (GENERAL) @@ -621,8 +629,15 @@ class TestExamples(unittest.TestCase): self.assertEqual('My Tor relay has read 33406 bytes and written 29649.\n', stdout_mock.getvalue()) - def test_introduction_points(self): - pass + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_introduction_points(self, stdout_mock, from_port_mock): + controller = from_port_mock().__enter__() + controller.get_hidden_service_descriptor.return_value = next(stem.descriptor.parse_file(os.path.join(DESC_DIR, 'hidden_service_duckduckgo'))) + + import introduction_points + + self.assertEqual(EXPECTED_INTRODUCTION_POINTS, stdout_mock.getvalue()) @patch('stem.control.Controller.from_port', spec = Controller) @patch('sys.stdout', new_callable = io.StringIO)

1 0

[stem/master] Test resuming_ephemeral_hidden_service example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 30e95d4feb4a72cdb62dfef963236b7368e7c7e9 Author: Damian Johnson <atagar(a)torproject.org> Date: Fri Oct 2 12:31:38 2020 -0700 Test resuming_ephemeral_hidden_service example --- .../example/resuming_ephemeral_hidden_service.py | 2 +- test/unit/examples.py | 22 ++++++++++++++++++++-- 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/docs/_static/example/resuming_ephemeral_hidden_service.py b/docs/_static/example/resuming_ephemeral_hidden_service.py index d2073f03..3277bce2 100644 --- a/docs/_static/example/resuming_ephemeral_hidden_service.py +++ b/docs/_static/example/resuming_ephemeral_hidden_service.py @@ -19,5 +19,5 @@ with Controller.from_port() as controller: service = controller.create_ephemeral_hidden_service({80: 5000}, key_type = key_type, key_content = key_content, await_publication = True) print("Resumed %s.onion" % service.service_id) - raw_input('press any key to shut the service down...') + input('press any key to shut the service down...') controller.remove_ephemeral_hidden_service(service.service_id) diff --git a/test/unit/examples.py b/test/unit/examples.py index e1c223bf..0b66c124 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -868,8 +868,26 @@ class TestExamples(unittest.TestCase): relay_connections.main([]) self.assertEqual(EXPECTED_RELAY_CONNECTIONS, stdout_mock.getvalue()) - def test_resuming_ephemeral_hidden_service(self): - pass + @patch('builtins.input', Mock()) + @patch('os.path.expanduser', Mock(return_value = '/tmp/stem_hs_test')) + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_resuming_ephemeral_hidden_service(self, stdout_mock, from_port_mock): + hs_response = '250-ServiceID=gfzprpioee3hoppz\n250-PrivateKey=RSA1024:MIICXgIB\n250 OK' + + controller = from_port_mock().__enter__() + controller.create_ephemeral_hidden_service.return_value = ControlMessage.from_str(hs_response, 'ADD_ONION', normalize = True) + + try: + import resuming_ephemeral_hidden_service + + with open('/tmp/stem_hs_test') as key_file: + self.assertEqual('RSA1024:MIICXgIB', key_file.read()) + + self.assertEqual('Started a new hidden service with the address of gfzprpioee3hoppz.onion\n', stdout_mock.getvalue()) + finally: + if os.path.exists('/tmp/stem_hs_test'): + os.remove('/tmp/stem_hs_test') @patch('stem.control.Controller.from_port', spec = Controller) @patch('shutil.rmtree')

1 0

[stem/master] Test that all examples are referenced
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit ef7416255976b2885733c170a6bde0263a239254 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 16:48:37 2020 -0700 Test that all examples are referenced Our unused load_test.py makes it clear that we should check that all our examples are cited by the website. I can see myself making that mistake again. --- docs/tutorials/to_russia_with_love.rst | 2 +- stem/manual.py | 2 +- test/unit/examples.py | 35 ++++++++++++++++++++++++++++++++-- 3 files changed, 35 insertions(+), 4 deletions(-) diff --git a/docs/tutorials/to_russia_with_love.rst b/docs/tutorials/to_russia_with_love.rst index fc1b4d2c..af94185a 100644 --- a/docs/tutorials/to_russia_with_love.rst +++ b/docs/tutorials/to_russia_with_love.rst @@ -68,7 +68,7 @@ connections through Tor, so this'll break our ability to connect to Tor's control port. To use this approach simply replace the query() function above with... -.. literalinclude:: /_static/example/client_usage_using_socksipy.py +.. literalinclude:: /_static/example/client_usage_using_socksipy.py :caption: `[Download] <../_static/example/client_usage_using_socksipy.py>`__ :language: python diff --git a/stem/manual.py b/stem/manual.py index dbde9827..e7658317 100644 --- a/stem/manual.py +++ b/stem/manual.py @@ -21,7 +21,7 @@ Manual information includes arguments, signals, and probably most usefully the torrc configuration options. For example, say we want a little script that told us what our torrc options do... -.. literalinclude:: /_static/example/manual_config_options.py +.. literalinclude:: /_static/example/manual_config_options.py :language: python | diff --git a/test/unit/examples.py b/test/unit/examples.py index 094322e6..12ffb60c 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -7,6 +7,7 @@ import binascii import functools import io import os +import re import sys import unittest @@ -231,13 +232,13 @@ class TestExamples(unittest.TestCase): stem.descriptor.remote.SINGLETON_DOWNLOADER = None - def test_runs_everything(self): + def test_everything_is_tested(self): """ Ensure we have tests for all our examples. """ all_examples = set([os.path.basename(path)[:-3] for path in stem.util.system.files_with_suffix(EXAMPLE_DIR, '.py')]) - tested_examples = set([method[5:] for method in dir(self) if method.startswith('test_') and method != 'test_runs_everything']) + tested_examples = set([method[5:] for method in dir(self) if method.startswith('test_') and not method.startswith('test_everything_')]) extra = sorted(tested_examples.difference(all_examples)) missing = sorted(all_examples.difference(tested_examples).difference(UNTESTED)) @@ -248,6 +249,36 @@ class TestExamples(unittest.TestCase): if missing: self.fail("Changed our examples directory? The following are untested: %s" % ', '.join(missing)) + def test_everything_is_referenced(self): + """ + Ensure that all our examples are referenced our website. Otherwise they're + dead code. + """ + + all_examples = set([os.path.basename(path)[:-3] for path in stem.util.system.files_with_suffix(EXAMPLE_DIR, '.py')]) + + include_regex = re.compile('.. literalinclude:: /_static/example/(\\S*).py') + referenced_examples = set() + + for path in stem.util.system.files_with_suffix(os.path.join(test.STEM_BASE, 'docs'), '.rst'): + with open(path) as example_file: + referenced_examples.update(include_regex.findall(example_file.read())) + + for path in stem.util.system.files_with_suffix(os.path.join(test.STEM_BASE, 'stem'), '.py'): + with open(path) as source_file: + referenced_examples.update(include_regex.findall(source_file.read())) + + extra = sorted(referenced_examples.difference(all_examples)) + missing = sorted(all_examples.difference(referenced_examples)) + + missing.remove('benchmark_stem') # expanded copy of benchmark_server_descriptor_stem.py + + if extra: + self.fail("Changed our documentation? We reference the following examples which are not present: %s" % ', '.join(extra)) + + if missing: + self.fail("Changed our documntation? The following examples are unreferenced: %s" % ', '.join(missing)) + @patch('stem.descriptor.remote.get_bandwidth_file') @patch('sys.stdout', new_callable = io.StringIO) def test_bandwidth_stats(self, stdout_mock, get_bandwidth_file_mock):

1 0

[stem/master] Test manual_config_options example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 51512eb6627a8b945340d5208fab68eb29885f57 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 18:10:28 2020 -0700 Test manual_config_options example --- docs/_static/example/manual_config_options.py | 2 +- test/unit/examples.py | 25 +++++++++++++++++++++++-- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/docs/_static/example/manual_config_options.py b/docs/_static/example/manual_config_options.py index 4a503579..336634a3 100644 --- a/docs/_static/example/manual_config_options.py +++ b/docs/_static/example/manual_config_options.py @@ -13,7 +13,7 @@ print('Which tor configuration would you like to learn about? (press ctrl+c to try: while True: - requested_option = raw_input('> ').strip() + requested_option = input('> ').strip() if requested_option: if requested_option in manual.config_options: diff --git a/test/unit/examples.py b/test/unit/examples.py index 12ffb60c..0b4d7082 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -172,6 +172,21 @@ Circuit 10 (GENERAL) +- 65242C91BFF30F165DA4D132C81A9EBA94B71D62 (torexit16, 176.67.169.171) """ +EXPECTED_MANUAL_CONFIG_OPTIONS = """\ +Downloading tor's manual information, please wait... + done + +Which tor configuration would you like to learn about? (press ctrl+c to quit) + +IPv6Exit 0|1 +Allow clients to use us for IPv6 traffic + +Full Description: + +If set, and we are an exit node, allow clients to use us for IPv6 traffic. When this option is set and ExitRelay is auto, we act as if ExitRelay is 1. (Default: 0) + +""" + EXPECTED_OUTDATED_RELAYS = """\ Checking for outdated relays... @@ -704,8 +719,14 @@ class TestExamples(unittest.TestCase): self.assertEqual(EXPECTED_LIST_CIRCUITS, stdout_mock.getvalue()) - def test_manual_config_options(self): - pass + @patch('stem.manual.Manual.from_remote', Mock(return_value = stem.manual.Manual.from_cache())) + @patch('stem.util.term.format', Mock(side_effect = lambda msg, *args: msg)) + @patch('sys.stdout', new_callable = io.StringIO) + def test_manual_config_options(self, stdout_mock): + with patch('builtins.input', Mock(side_effect = ['IPv6Exit', KeyboardInterrupt()])): + import manual_config_options + + self.assertEqual(EXPECTED_MANUAL_CONFIG_OPTIONS, stdout_mock.getvalue()) @patch('stem.descriptor.remote.DescriptorDownloader') @patch('sys.stdout', new_callable = io.StringIO)

1 0

[stem/master] Test slow_listener example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 2880be78fd34fd48939ddb417d2c4b7d725e57f6 Author: Damian Johnson <atagar(a)torproject.org> Date: Fri Oct 2 13:23:46 2020 -0700 Test slow_listener example --- test/unit/examples.py | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/test/unit/examples.py b/test/unit/examples.py index 73bae36c..5e2bbc78 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -930,8 +930,25 @@ class TestExamples(unittest.TestCase): if os.path.exists('/tmp/descriptor_dump'): os.remove('/tmp/descriptor_dump') - def test_slow_listener(self): - pass + @patch('time.sleep') + @patch('time.time', Mock(return_value = 123)) + @patch('stem.control.Controller.authenticate', Mock()) + @patch('stem.control.Controller.is_alive', Mock(return_value = True)) + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_slow_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() at the end, but *not* + # within the listener + + bw_event = ControlMessage.from_str('650 BW 15 25', 'EVENT', normalize = True) + sleep_mock.side_effect = lambda duration: controller._handle_event(bw_event) if duration == 10 else None + + import slow_listener + + self.assertEqual("processing a BW event that's 0.0 seconds old (0 more events are waiting)\n", stdout_mock.getvalue()) @patch('stem.descriptor.remote.DescriptorDownloader') @patch('sys.stdout', new_callable = io.StringIO)

1 0

[stem/master] Test queue_listener example
by atagar＠torproject.org 02 Oct '20

02 Oct '20

commit 437ea486834ae9e320675ec0c80aa4ecf293fcc6 Author: Damian Johnson <atagar(a)torproject.org> Date: Tue Sep 29 18:39:46 2020 -0700 Test queue_listener example --- test/unit/examples.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/test/unit/examples.py b/test/unit/examples.py index 0b4d7082..6e32a0f2 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -772,8 +772,18 @@ class TestExamples(unittest.TestCase): self.assertEqual(EXPECTED_PERSISTING_A_CONSENSUS, stdout_mock.getvalue()) - def test_queue_listener(self): - pass + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_queue_listener(self, stdout_mock, from_port_mock): + bw_event = ControlMessage.from_str('650 BW 15 25', 'EVENT', normalize = True) + + controller = from_port_mock().__enter__() + controller.add_event_listener.side_effect = lambda handler, event_type: handler(bw_event) + + with patch('time.time', Mock(side_effect = [1, 1, 10])): + import queue_listener + + self.assertEqual('I got a BW event for 15 bytes downloaded and 25 bytes uploaded\n', stdout_mock.getvalue()) def test_read_with_parse_file(self): pass

1 0