[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [stem/master] Rewriting authentication integ tests



commit a73be509009942b4b2711b49a3b30e3c3f6e609b
Author: Damian Johnson <atagar@xxxxxxxxxxxxxx>
Date:   Sat Jan 28 17:06:36 2012 -0800

    Rewriting authentication integ tests
    
    Excessive testing helper functions are bad. They hurt test readability and
    makes the code a pita to trace through. This cleans up the authentication
    integration tests to collapse the helper functions into just the few that make
    life better. Much more maintainable now. \o/
---
 stem/socket.py                          |    6 +
 test/integ/connection/authentication.py |  276 ++++++++++++-------------------
 2 files changed, 116 insertions(+), 166 deletions(-)

diff --git a/stem/socket.py b/stem/socket.py
index f3aeaf8..889713d 100644
--- a/stem/socket.py
+++ b/stem/socket.py
@@ -217,6 +217,12 @@ class ControlSocket:
     self._send_cond.release()
     self._recv_cond.release()
   
+  def __enter__(self):
+    return self
+  
+  def __exit__(self, type, value, traceback):
+    self.close()
+  
   def _make_socket(self):
     """
     Constructs and connects new socket. This is implemented by subclasses.
diff --git a/test/integ/connection/authentication.py b/test/integ/connection/authentication.py
index bc94ead..cc36c66 100644
--- a/test/integ/connection/authentication.py
+++ b/test/integ/connection/authentication.py
@@ -5,7 +5,6 @@ stem.connection.authenticate* functions.
 
 import os
 import unittest
-import functools
 
 import test.runner
 import stem.connection
@@ -23,6 +22,60 @@ MULTIPLE_AUTH_FAIL = "Authentication failed: Password did not match HashedContro
 INCORRECT_COOKIE_FAIL = "Authentication failed: Authentication cookie did not match expected value."
 INCORRECT_PASSWORD_FAIL = "Authentication failed: Password did not match HashedControlPassword value from configuration"
 
+def _can_authenticate(auth_type):
+  """
+  Checks if a given authentication method can authenticate to our control
+  socket.
+  
+  Arguments:
+    auth_type (stem.connection.AuthMethod) - authentication method to check
+  
+  Returns:
+    bool that's True if we should be able to authenticate and False otherwise
+  """
+  
+  tor_options = test.runner.get_runner().get_options()
+  password_auth = test.runner.Torrc.PASSWORD in tor_options
+  cookie_auth = test.runner.Torrc.COOKIE in tor_options
+  
+  if not password_auth and not cookie_auth: return True # open socket
+  elif auth_type == stem.connection.AuthMethod.PASSWORD: return password_auth
+  elif auth_type == stem.connection.AuthMethod.COOKIE: return cookie_auth
+  else: return False
+
+def _get_auth_failure_message(auth_type):
+  """
+  Provides the message that tor will respond with if our current method of
+  authentication fails. Note that this test will need to be updated if tor
+  changes its rejection reponse.
+  
+  Arguments:
+    auth_type (stem.connection.AuthMethod) - authentication method to check
+  
+  Returns:
+    string with the rejection message that tor would provide
+  """
+  
+  tor_options = test.runner.get_runner().get_options()
+  password_auth = test.runner.Torrc.PASSWORD in tor_options
+  cookie_auth = test.runner.Torrc.COOKIE in tor_options
+  
+  if cookie_auth and password_auth:
+    return MULTIPLE_AUTH_FAIL
+  elif cookie_auth:
+    if auth_type == stem.connection.AuthMethod.COOKIE:
+      return INCORRECT_COOKIE_FAIL
+    else:
+      return COOKIE_AUTH_FAIL
+  elif password_auth:
+    if auth_type == stem.connection.AuthMethod.PASSWORD:
+      return INCORRECT_PASSWORD_FAIL
+    else:
+      return PASSWORD_AUTH_FAIL
+  else:
+    # shouldn't happen, if so then the test has a bug
+    raise ValueError("No methods of authentication. If this is an open socket then auth shoulnd't fail.")
+
 class TestAuthenticate(unittest.TestCase):
   def setUp(self):
     # none of these tests apply if there's no control connection
@@ -34,10 +87,9 @@ class TestAuthenticate(unittest.TestCase):
     Tests that the authenticate function can authenticate to our socket.
     """
     
-    control_socket = test.runner.get_runner().get_tor_socket(False)
-    stem.connection.authenticate(control_socket, test.runner.CONTROL_PASSWORD)
-    test.runner.exercise_socket(self, control_socket)
-    control_socket.close()
+    with test.runner.get_runner().get_tor_socket(False) as control_socket:
+      stem.connection.authenticate(control_socket, test.runner.CONTROL_PASSWORD)
+      test.runner.exercise_socket(self, control_socket)
   
   def test_authenticate_general_example(self):
     """
@@ -87,34 +139,25 @@ class TestAuthenticate(unittest.TestCase):
     is_password_only = test.runner.Torrc.PASSWORD in tor_options and not test.runner.Torrc.COOKIE in tor_options
     
     # tests without a password
-    control_socket = runner.get_tor_socket(False)
-    auth_function = functools.partial(stem.connection.authenticate, control_socket)
-    
-    if is_password_only:
-      self.assertRaises(stem.connection.MissingPassword, auth_function)
-    else:
-      auth_function()
-      test.runner.exercise_socket(self, control_socket)
-    
-    control_socket.close()
+    with runner.get_tor_socket(False) as control_socket:
+      if is_password_only:
+        self.assertRaises(stem.connection.MissingPassword, stem.connection.authenticate, control_socket)
+      else:
+        stem.connection.authenticate(control_socket)
+        test.runner.exercise_socket(self, control_socket)
     
     # tests with the incorrect password
-    control_socket = runner.get_tor_socket(False)
-    auth_function = functools.partial(stem.connection.authenticate, control_socket, "blarg")
-    
-    if is_password_only:
-      self.assertRaises(stem.connection.IncorrectPassword, auth_function)
-    else:
-      auth_function()
-      test.runner.exercise_socket(self, control_socket)
-    
-    control_socket.close()
+    with runner.get_tor_socket(False) as control_socket:
+      if is_password_only:
+        self.assertRaises(stem.connection.IncorrectPassword, stem.connection.authenticate, control_socket, "blarg")
+      else:
+        stem.connection.authenticate(control_socket, "blarg")
+        test.runner.exercise_socket(self, control_socket)
     
     # tests with the right password
-    control_socket = runner.get_tor_socket(False)
-    stem.connection.authenticate(control_socket, test.runner.CONTROL_PASSWORD)
-    test.runner.exercise_socket(self, control_socket)
-    control_socket.close()
+    with runner.get_tor_socket(False) as control_socket:
+      stem.connection.authenticate(control_socket, test.runner.CONTROL_PASSWORD)
+      test.runner.exercise_socket(self, control_socket)
   
   def test_authenticate_none(self):
     """
@@ -122,11 +165,11 @@ class TestAuthenticate(unittest.TestCase):
     """
     
     auth_type = stem.connection.AuthMethod.NONE
-    if self._can_authenticate(auth_type):
+    
+    if _can_authenticate(auth_type):
       self._check_auth(auth_type)
     else:
       self.assertRaises(stem.connection.OpenAuthRejected, self._check_auth, auth_type)
-      self._assert_auth_rejected_msg(auth_type)
   
   def test_authenticate_password(self):
     """
@@ -136,26 +179,24 @@ class TestAuthenticate(unittest.TestCase):
     auth_type = stem.connection.AuthMethod.PASSWORD
     auth_value = test.runner.CONTROL_PASSWORD
     
-    if self._can_authenticate(auth_type):
+    if _can_authenticate(auth_type):
       self._check_auth(auth_type, auth_value)
     else:
       self.assertRaises(stem.connection.PasswordAuthRejected, self._check_auth, auth_type, auth_value)
-      self._assert_auth_rejected_msg(auth_type, auth_value)
     
     # Check with an empty, invalid, and quoted password. These should work if
     # we have no authentication, and fail otherwise.
     
     for auth_value in ("", "blarg", "this has a \" in it"):
-      if self._can_authenticate(stem.connection.AuthMethod.NONE):
+      if _can_authenticate(stem.connection.AuthMethod.NONE):
         self._check_auth(auth_type, auth_value)
       else:
-        if self._can_authenticate(stem.connection.AuthMethod.PASSWORD):
+        if _can_authenticate(stem.connection.AuthMethod.PASSWORD):
           exc_type = stem.connection.IncorrectPassword
         else:
           exc_type = stem.connection.PasswordAuthRejected
         
         self.assertRaises(exc_type, self._check_auth, auth_type, auth_value)
-        self._assert_auth_rejected_msg(auth_type, auth_value)
   
   def test_authenticate_cookie(self):
     """
@@ -173,11 +214,10 @@ class TestAuthenticate(unittest.TestCase):
       # missing file.
       
       self.assertRaises(stem.connection.UnreadableCookieFile, self._check_auth, auth_type, auth_value)
-    elif self._can_authenticate(auth_type):
+    elif _can_authenticate(auth_type):
       self._check_auth(auth_type, auth_value)
     else:
       self.assertRaises(stem.connection.CookieAuthRejected, self._check_auth, auth_type, auth_value)
-      self._assert_auth_rejected_msg(auth_type, auth_value)
   
   def test_authenticate_cookie_invalid(self):
     """
@@ -193,17 +233,16 @@ class TestAuthenticate(unittest.TestCase):
     fake_cookie.write("0" * 32)
     fake_cookie.close()
     
-    if self._can_authenticate(stem.connection.AuthMethod.NONE):
+    if _can_authenticate(stem.connection.AuthMethod.NONE):
       # authentication will work anyway
       self._check_auth(auth_type, auth_value)
     else:
-      if self._can_authenticate(auth_type):
+      if _can_authenticate(auth_type):
         exc_type = stem.connection.IncorrectCookieValue
       else:
         exc_type = stem.connection.CookieAuthRejected
       
       self.assertRaises(exc_type, self._check_auth, auth_type, auth_value)
-      self._assert_auth_rejected_msg(auth_type, auth_value)
     
     os.remove(auth_value)
   
@@ -215,7 +254,7 @@ class TestAuthenticate(unittest.TestCase):
     
     auth_type = stem.connection.AuthMethod.COOKIE
     auth_value = "/if/this/exists/then/they're/asking/for/a/failure"
-    self.assertRaises(stem.connection.UnreadableCookieFile, self._check_auth, auth_type, auth_value)
+    self.assertRaises(stem.connection.UnreadableCookieFile, self._check_auth, auth_type, auth_value, False)
   
   def test_authenticate_cookie_wrong_size(self):
     """
@@ -231,139 +270,44 @@ class TestAuthenticate(unittest.TestCase):
       # Weird coincidence? Fail so we can pick another file to check against.
       self.fail("Our torrc is 32 bytes, preventing the test_authenticate_cookie_wrong_size test from running.")
     else:
-      self.assertRaises(stem.connection.IncorrectCookieSize, self._check_auth, auth_type, auth_value)
-  
-  def _get_socket_auth(self):
-    """
-    Provides the types of authentication that our current test socket accepts.
-    
-    Returns:
-      bool tuple of the form (password_auth, cookie_auth)
-    """
-    
-    tor_options = test.runner.get_runner().get_options()
-    password_auth = test.runner.Torrc.PASSWORD in tor_options
-    cookie_auth = test.runner.Torrc.COOKIE in tor_options
-    
-    return password_auth, cookie_auth
-  
-  def _can_authenticate(self, auth_type):
-    """
-    Checks if the given authentication type should be able to authenticate to
-    our current socket.
-    
-    Arguments:
-      auth_type (stem.connection.AuthMethod) - authentication method to check
-    
-    Returns:
-      bool that's True if we should be able to authenticate and False otherwise
-    """
-    
-    password_auth, cookie_auth = self._get_socket_auth()
-    
-    # If the control socket is open then all authentication methods will be
-    # accepted. Otherwise check if our auth type matches what the socket
-    # accepts.
-    
-    if not password_auth and not cookie_auth: return True
-    elif auth_type == stem.connection.AuthMethod.PASSWORD: return password_auth
-    elif auth_type == stem.connection.AuthMethod.COOKIE: return cookie_auth
-    else: return False
-  
-  def _get_auth_function(self, control_socket, auth_type, *auth_args):
-    """
-    Constructs a functor that performs the given authentication without
-    additional arguments.
-    
-    Arguments:
-      control_socket (stem.socket.ControlSocket) - socket for the function to
-          authenticate to
-      auth_type (stem.connection.AuthMethod) - method by which we should
-          authentiate to the control socket
-      auth_args (str) - arguments to be passed to the authentication function
-    """
-    
-    if auth_type == stem.connection.AuthMethod.NONE:
-      auth_function = stem.connection.authenticate_none
-    elif auth_type == stem.connection.AuthMethod.PASSWORD:
-      auth_function = stem.connection.authenticate_password
-    elif auth_type == stem.connection.AuthMethod.COOKIE:
-      auth_function = stem.connection.authenticate_cookie
-    else:
-      raise ValueError("unexpected auth type: %s" % auth_type)
-    
-    if auth_args:
-      return functools.partial(auth_function, control_socket, *auth_args)
-    else:
-      return functools.partial(auth_function, control_socket)
-  
-  def _assert_auth_rejected_msg(self, auth_type, *auth_args):
-    """
-    This asserts that authentication will fail with the rejection message given
-    by tor. Note that this test will need to be updated if tor changes its
-    rejection reponse.
-    
-    Arguments:
-      auth_type (stem.connection.AuthMethod) - method by which we should
-          authentiate to the control socket
-      auth_args (str) - arguments to be passed to the authentication function
-    """
-    
-    control_socket = test.runner.get_runner().get_tor_socket(False)
-    auth_function = self._get_auth_function(control_socket, auth_type, *auth_args)
-    password_auth, cookie_auth = self._get_socket_auth()
-    
-    if cookie_auth and password_auth:
-      failure_msg = MULTIPLE_AUTH_FAIL
-    elif cookie_auth:
-      if auth_type == stem.connection.AuthMethod.COOKIE:
-        failure_msg = INCORRECT_COOKIE_FAIL
-      else:
-        failure_msg = COOKIE_AUTH_FAIL
-    elif password_auth:
-      if auth_type == stem.connection.AuthMethod.PASSWORD:
-        failure_msg = INCORRECT_PASSWORD_FAIL
-      else:
-        failure_msg = PASSWORD_AUTH_FAIL
-    else:
-      # shouldn't happen, if so then the test has a bug
-      raise ValueError("No methods of authentication. If this is an open socket then auth shoulnd't fail.")
-    
-    try:
-      auth_function()
-      control_socket.close()
-      self.fail()
-    except stem.connection.AuthenticationFailure, exc:
-      self.assertTrue(control_socket.is_alive())
-      self.assertEqual(failure_msg, str(exc))
-      control_socket.close()
+      self.assertRaises(stem.connection.IncorrectCookieSize, self._check_auth, auth_type, auth_value, False)
   
-  def _check_auth(self, auth_type, *auth_args):
+  def _check_auth(self, auth_type, auth_arg = None, check_message = True):
     """
-    Attempts to use the given authentication function against our connection.
-    If this works then checks that we can use the connection. If not then this
-    raises the exception.
+    Attempts to use the given type of authentication against tor's control
+    socket. If it succeeds then we check that the socket can then be used. If
+    not then we check that this gives a message that we'd expect then raises
+    the exception.
     
     Arguments:
       auth_type (stem.connection.AuthMethod) - method by which we should
           authentiate to the control socket
-      auth_args (str) - arguments to be passed to the authentication function
+      auth_arg (str) - argument to be passed to the authentication function
+      check_message (bool) - checks that failure messages are what we'd expect
     
     Raises:
       stem.connection.AuthenticationFailure if the authentication fails
     """
     
-    control_socket = test.runner.get_runner().get_tor_socket(False)
-    auth_function = self._get_auth_function(control_socket, auth_type, *auth_args)
-    
-    # run the authentication, re-raising if there's a problem
-    try:
-      auth_function()
-    except stem.connection.AuthenticationFailure, exc:
-      self.assertTrue(control_socket.is_alive())
-      control_socket.close()
-      raise exc
-    
-    test.runner.exercise_socket(self, control_socket)
-    control_socket.close()
+    with test.runner.get_runner().get_tor_socket(False) as control_socket:
+      # run the authentication, re-raising if there's a problem
+      try:
+        if auth_type == stem.connection.AuthMethod.NONE:
+          stem.connection.authenticate_none(control_socket)
+        elif auth_type == stem.connection.AuthMethod.PASSWORD:
+          stem.connection.authenticate_password(control_socket, auth_arg)
+        elif auth_type == stem.connection.AuthMethod.COOKIE:
+          stem.connection.authenticate_cookie(control_socket, auth_arg)
+        
+        test.runner.exercise_socket(self, control_socket)
+      except stem.connection.AuthenticationFailure, exc:
+        # authentication functions should re-attach on failure
+        self.assertTrue(control_socket.is_alive())
+        
+        # check that we got the failure message that we'd expect
+        if check_message:
+          failure_msg = _get_auth_failure_message(auth_type)
+          self.assertEqual(failure_msg, str(exc))
+        
+        raise exc
 



_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits