commit 87446136328bddbbd0cbecf931349a4cdd029518 Author: Damian Johnson atagar@torproject.org Date: Sat Sep 26 23:08:51 2020 -0700
Move exit_used test
Seriously? I copy-pasted the tutorial example into our test? That is... embarrasing. --- docs/_static/example/exit_used.py | 2 +- test/unit/examples.py | 52 +++++++++++++++++++++++++------- test/unit/tutorial_examples.py | 63 --------------------------------------- 3 files changed, 43 insertions(+), 74 deletions(-)
diff --git a/docs/_static/example/exit_used.py b/docs/_static/example/exit_used.py index 9a16c130..fde2db84 100644 --- a/docs/_static/example/exit_used.py +++ b/docs/_static/example/exit_used.py @@ -13,7 +13,7 @@ def main(): stream_listener = functools.partial(stream_event, controller) controller.add_event_listener(stream_listener, EventType.STREAM)
- raw_input() # wait for user to press enter + input() # wait for user to press enter
def stream_event(controller, event): diff --git a/test/unit/examples.py b/test/unit/examples.py index 30375ea3..efdac292 100644 --- a/test/unit/examples.py +++ b/test/unit/examples.py @@ -4,6 +4,7 @@ Exercise the code in our examples directory.
import base64 import binascii +import functools import io import os import sys @@ -104,6 +105,17 @@ EXPECTED_COLLECTOR_READING = """\ caerSidi (4F0C867DF0EF68160568C826838F482CEA7CFE44) """
+EXPECTED_EXIT_USED = """\ +Tracking requests for tor exits. Press 'enter' to end. + +Exit relay for our connection to 64.15.112.44:80 + address: 31.172.30.2:443 + fingerprint: A59E1E7C7EAEE083D756EE1FF6EC31CA3D8651D7 + nickname: chaoscomputerclub19 + locale: unknown + +""" + EXPECTED_LIST_CIRCUITS = """\
Circuit 4 (GENERAL) @@ -123,6 +135,12 @@ Circuit 10 (GENERAL) """
+def _make_circ_event(circ_id, hop1, hop2, hop3): + path = '$%s=%s,$%s=%s,$%s=%s' % (hop1[0], hop1[1], hop2[0], hop2[1], hop3[0], hop3[1]) + content = '650 CIRC %i BUILT %s PURPOSE=GENERAL' % (circ_id, path) + return ControlMessage.from_str(content, 'EVENT', normalize = True) + + class TestExamples(unittest.TestCase): def setUp(self): self.original_path = list(sys.path) @@ -328,8 +346,27 @@ class TestExamples(unittest.TestCase): def test_event_listening(self): pass
- def test_exit_used(self): - pass + @patch('stem.control.Controller.from_port', spec = Controller) + @patch('sys.stdout', new_callable = io.StringIO) + def test_exit_used(self, stdout_mock, from_port_mock): + path_1 = ('9EA317EECA56BDF30CAEB208A253FB456EDAB1A0', 'bolobolo1') + path_2 = ('00C2C2A16AEDB51D5E5FB7D6168FC66B343D822F', 'ph3x') + path_3 = ('A59E1E7C7EAEE083D756EE1FF6EC31CA3D8651D7', 'chaoscomputerclub19') + + event = ControlMessage.from_str('650 STREAM 15 SUCCEEDED 3 64.15.112.44:80', 'EVENT', normalize = True) + r_line = '%s pZ4efH6u4IPXVu4f9uwxyj2GUdc= oQZFLYe9e4A7bOkWKR7TaNxb0JE 2012-08-06 11:19:31 31.172.30.2 443 0' + + controller = from_port_mock().__enter__() + controller.get_circuit.return_value = _make_circ_event(1, path_1, path_2, path_3) + controller.get_network_status.return_value = RouterStatusEntryV3.create({'r': r_line % path_3[1]}) + controller.get_info.return_value = 'unknown' + + import exit_used + + with patch('builtins.input', Mock(side_effect = functools.partial(exit_used.stream_event, controller, event))): + exit_used.main() + + self.assertEqual(EXPECTED_EXIT_USED, stdout_mock.getvalue())
def test_fibonacci_multiprocessing(self): pass @@ -349,11 +386,6 @@ class TestExamples(unittest.TestCase): @patch('stem.control.Controller.from_port', spec = Controller) @patch('sys.stdout', new_callable = io.StringIO) def test_list_circuits(self, stdout_mock, from_port_mock): - def _get_circ_event(circ_id, hop1, hop2, hop3): - path = '$%s=%s,$%s=%s,$%s=%s' % (hop1[0], hop1[1], hop2[0], hop2[1], hop3[0], hop3[1]) - content = '650 CIRC %i BUILT %s PURPOSE=GENERAL' % (circ_id, path) - return ControlMessage.from_str(content, 'EVENT', normalize = True) - path_1 = ('B1FA7D51B8B6F0CB585D944F450E7C06EDE7E44C', 'ByTORAndTheSnowDog') path_2 = ('0DD9935C5E939CFA1E07B8DDA6D91C1A2A9D9338', 'afo02') path_3 = ('DB3B1CFBD3E4D97B84B548ADD5B9A31451EEC4CC', 'edwardsnowden3') @@ -362,9 +394,9 @@ class TestExamples(unittest.TestCase): path_6 = ('00C2C2A16AEDB51D5E5FB7D6168FC66B343D822F', 'ph3x') path_7 = ('65242C91BFF30F165DA4D132C81A9EBA94B71D62', 'torexit16')
- circuit_4 = _get_circ_event(4, path_1, path_2, path_3) - circuit_6 = _get_circ_event(6, path_1, path_4, path_5) - circuit_10 = _get_circ_event(10, path_1, path_6, path_7) + circuit_4 = _make_circ_event(4, path_1, path_2, path_3) + circuit_6 = _make_circ_event(6, path_1, path_4, path_5) + circuit_10 = _make_circ_event(10, path_1, path_6, path_7)
controller = from_port_mock().__enter__() controller.get_circuits.return_value = [circuit_4, circuit_6, circuit_10] diff --git a/test/unit/tutorial_examples.py b/test/unit/tutorial_examples.py index d673662f..4df9f6c4 100644 --- a/test/unit/tutorial_examples.py +++ b/test/unit/tutorial_examples.py @@ -9,7 +9,6 @@ import unittest
from unittest.mock import Mock, patch
-from stem.control import Controller from stem.descriptor.networkstatus import NetworkStatusDocumentV3 from stem.descriptor.router_status_entry import RouterStatusEntryV3 from stem.descriptor.server_descriptor import RelayDescriptor @@ -26,17 +25,6 @@ PURPOSE=%s'
PATH_CONTENT = '$%s=%s,$%s=%s,$%s=%s'
-EXIT_USED_OUTPUT = """\ -Tracking requests for tor exits. Press 'enter' to end. - -Exit relay for our connection to 64.15.112.44:80 - address: 31.172.30.2:443 - fingerprint: A59E1E7C7EAEE083D756EE1FF6EC31CA3D8651D7 - nickname: chaoscomputerclub19 - locale: unknown - -""" - OUTDATED_RELAYS_OUTPUT = """\ Checking for outdated relays...
@@ -99,57 +87,6 @@ def _get_router_status(address = None, port = None, nickname = None, fingerprint
class TestTutorialExamples(unittest.TestCase): - @patch('sys.stdout', new_callable = io.StringIO) - @patch('stem.control.Controller.from_port', spec = Controller) - def test_exit_used(self, from_port_mock, stdout_mock): - def tutorial_example(mock_event): - import functools - - from stem import StreamStatus - from stem.control import EventType, Controller - - def main(): - print("Tracking requests for tor exits. Press 'enter' to end.\n") - - with Controller.from_port() as controller: - controller.authenticate() - - stream_listener = functools.partial(stream_event, controller) - controller.add_event_listener(stream_listener, EventType.STREAM) - - stream_event(controller, mock_event) # simulate an event during the raw_input() - - def stream_event(controller, event): - if event.status == StreamStatus.SUCCEEDED and event.circ_id: - circ = controller.get_circuit(event.circ_id) - - exit_fingerprint = circ.path[-1][0] - exit_relay = controller.get_network_status(exit_fingerprint) - - print('Exit relay for our connection to %s' % (event.target)) - print(' address: %s:%i' % (exit_relay.address, exit_relay.or_port)) - print(' fingerprint: %s' % exit_relay.fingerprint) - print(' nickname: %s' % exit_relay.nickname) - print(' locale: %s\n' % controller.get_info('ip-to-country/%s' % exit_relay.address, 'unknown')) - - main() - - path_1 = ('9EA317EECA56BDF30CAEB208A253FB456EDAB1A0', 'bolobolo1') - path_2 = ('00C2C2A16AEDB51D5E5FB7D6168FC66B343D822F', 'ph3x') - path_3 = ('A59E1E7C7EAEE083D756EE1FF6EC31CA3D8651D7', 'chaoscomputerclub19') - circuit = _get_circ_event(1, 'BUILT', path_1, path_2, path_3, 'GENERAL') - - event_content = '650 STREAM 15 SUCCEEDED 3 64.15.112.44:80' - event = _get_event(event_content) - - controller = from_port_mock().__enter__() - controller.get_circuit.return_value = circuit - controller.get_network_status.return_value = _get_router_status('31.172.30.2', '443', path_3[1], 'pZ4efH6u4IPXVu4f9uwxyj2GUdc=') - controller.get_info.return_value = 'unknown' - - tutorial_example(event) - self.assertCountEqual(EXIT_USED_OUTPUT.splitlines(), stdout_mock.getvalue().splitlines()) - @patch('sys.stdout', new_callable = io.StringIO) @patch('stem.descriptor.remote.DescriptorDownloader') def test_outdated_relays(self, downloader_mock, stdout_mock):