[tor-commits] [stem/master] Added test for exit_used

[atagar at torproject.org]

commit 7deeb8520e4302d7c3df4137704baed59716025b
Author: Sambuddha Basu <sambuddhabasu1 at gmail.com>
Date:   Tue Dec 16 17:17:21 2014 +0400

    Added test for exit_used
---
 test/unit/tutorial_examples.py |   92 ++++++++++++++++++++++++++++++++++++----
 1 file changed, 84 insertions(+), 8 deletions(-)

diff --git a/test/unit/tutorial_examples.py b/test/unit/tutorial_examples.py
index 59be150..9184957 100644
--- a/test/unit/tutorial_examples.py
+++ b/test/unit/tutorial_examples.py
@@ -42,17 +42,40 @@ Circuit 10 (GENERAL)
  +- 65242C91BFF30F165DA4D132C81A9EBA94B71D62 (torexit16, 176.67.169.171)
 """
 
+EXIT_USED_OUTPUT = """\
+Tracking requests for tor exits. Press 'enter' to end.
 
-def _get_event(id, status, hop1, hop2, hop3, purpose):
-  path = PATH_CONTENT % (hop1[0], hop1[1], hop2[0], hop2[1], hop3[0], hop3[1])
-  content = CIRC_CONTENT % (id, status, path, purpose)
+Exit relay for our connection to 64.15.112.44:80
+  address: 31.172.30.2:443
+  fingerprint: A59E1E7C7EAEE083D756EE1FF6EC31CA3D8651D7
+  nickname: chaoscomputerclub19
+  locale: unknown
+
+"""
+
+
+def _get_event(content):
   controller_event = mocking.get_message(content)
   stem.response.convert('EVENT', controller_event)
   return controller_event
 
 
-def _get_router_status(address):
-  r_line = ROUTER_STATUS_ENTRY_V3_HEADER[0][1].replace('71.35.150.29', address)
+def _get_circ_event(id, status, hop1, hop2, hop3, purpose):
+  path = PATH_CONTENT % (hop1[0], hop1[1], hop2[0], hop2[1], hop3[0], hop3[1])
+  content = CIRC_CONTENT % (id, status, path, purpose)
+  return _get_event(content)
+
+
+def _get_router_status(address = None, port = None, nickname = None, fingerprint_base64 = None):
+  r_line = ROUTER_STATUS_ENTRY_V3_HEADER[0][1]
+  if address:
+    r_line = r_line.replace('71.35.150.29', address)
+  if port:
+    r_line = r_line.replace('9001', port)
+  if nickname:
+    r_line = r_line.replace('caerSidi', nickname)
+  if fingerprint_base64:
+    r_line = r_line.replace('p1aag7VwarGxqctS7/fS0y5FU+s', fingerprint_base64)
   content = get_router_status_entry_v3({'r': r_line})
   return content
 
@@ -92,9 +115,9 @@ class TestTutorialExamples(unittest.TestCase):
     path_6 = ('00C2C2A16AEDB51D5E5FB7D6168FC66B343D822F', 'ph3x')
     path_7 = ('65242C91BFF30F165DA4D132C81A9EBA94B71D62', 'torexit16')
 
-    circuit_4 = _get_event(4, 'BUILT', path_1, path_2, path_3, 'GENERAL')
-    circuit_6 = _get_event(6, 'BUILT', path_1, path_4, path_5, 'GENERAL')
-    circuit_10 = _get_event(10, 'BUILT', path_1, path_6, path_7, 'GENERAL')
+    circuit_4 = _get_circ_event(4, 'BUILT', path_1, path_2, path_3, 'GENERAL')
+    circuit_6 = _get_circ_event(6, 'BUILT', path_1, path_4, path_5, 'GENERAL')
+    circuit_10 = _get_circ_event(10, 'BUILT', path_1, path_6, path_7, 'GENERAL')
 
     controller = from_port_mock().__enter__()
     controller.get_circuits.return_value = [circuit_4, circuit_6, circuit_10]
@@ -109,3 +132,56 @@ class TestTutorialExamples(unittest.TestCase):
     }[fingerprint]
     tutorial_example()
     self.assertEqual(LIST_CIRCUITS_OUTPUT, stdout_mock.getvalue())
+
+  @patch('sys.stdout', new_callable = StringIO.StringIO)
+  @patch('stem.control.Controller.from_port', spec = Controller)
+  def test_exit_used(self, from_port_mock, stdout_mock):
+    import functools
+
+    from stem import StreamStatus
+    from stem.control import EventType, Controller
+
+    def main():
+      print "Tracking requests for tor exits. Press 'enter' to end."
+      print
+
+      with Controller.from_port() as controller:
+        controller.authenticate()
+
+        stream_listener = functools.partial(stream_event, controller)
+        controller.add_event_listener(stream_listener, EventType.STREAM)
+
+        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" % controller.get_info("ip-to-country/%s" % exit_relay.address, 'unknown')
+        print
+
+    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'
+    origin_raw_input = __builtins__['raw_input']
+    __builtins__['raw_input'] = lambda: ""
+    main()
+    __builtins__['raw_input'] = origin_raw_input
+    stream_event(controller, event)
+    self.assertEqual(EXIT_USED_OUTPUT, stdout_mock.getvalue())