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

[tor-commits] [bridgedb/master] PEP8 style fixes for bridgedb.test.test_smtp.



commit fc127ad1e59601d0507a2ae802d333c220291a17
Author: trygve <tor-dev@xxxxxxxxxxxxxxxxxxxx>
Date:   Fri Aug 8 03:30:05 2014 +0000

    PEP8 style fixes for bridgedb.test.test_smtp.
---
 lib/bridgedb/test/test_smtp.py |   76 +++++++++++++++++++++++++++-------------
 1 file changed, 52 insertions(+), 24 deletions(-)

diff --git a/lib/bridgedb/test/test_smtp.py b/lib/bridgedb/test/test_smtp.py
index 4e59002..4f30030 100644
--- a/lib/bridgedb/test/test_smtp.py
+++ b/lib/bridgedb/test/test_smtp.py
@@ -16,9 +16,13 @@ import random
 SMTP_DEBUG_LEVEL = 0  # set to 1 to see SMTP message exchange
 BRIDGEDB_SMTP_SERVER_ADDRESS = "localhost"
 BRIDGEDB_SMTP_SERVER_PORT = 6725
-FROM_ADDRESS_TEMPLATE = "test%d@xxxxxxxxx" # %d is parameterised with a random integer to make the sender unique
-MIN_FROM_ADDRESS = 1 # minimum value used to parameterise FROM_ADDRESS_TEMPLATE
-MAX_FROM_ADDRESS = 10**8  # max value used to parameterise FROM_ADDRESS_TEMPLATE. Needs to be pretty big to reduce the chance of collisions
+# %d is parameterised with a random integer to make the sender unique
+FROM_ADDRESS_TEMPLATE = "test%d@xxxxxxxxx"
+# Minimum value used to parameterise FROM_ADDRESS_TEMPLATE
+MIN_FROM_ADDRESS = 1
+# Max value used to parameterise FROM_ADDRESS_TEMPLATE. Needs to be pretty big
+# to reduce the chance of collisions
+MAX_FROM_ADDRESS = 10**8
 TO_ADDRESS = "bridges@xxxxxxxxxxxxxx"
 MESSAGE_TEMPLATE = """From: %s
 To: %s
@@ -32,6 +36,7 @@ get bridges"""
 LOCAL_SMTP_SERVER_ADDRESS = 'localhost'
 LOCAL_SMTP_SERVER_PORT = 2525 # Must be the same as bridgedb's EMAIL_SMTP_PORT
 
+
 class EmailServer(SMTPServer):
     def process_message(self, peer, mailfrom, rcpttos, data):
         ''' Overridden from SMTP server, called whenever a message is received'''
@@ -42,34 +47,42 @@ class EmailServer(SMTPServer):
         until the _stop Event object is set by the stop() function'''
         while self._stop.is_set() == False:
             asyncore.loop(timeout=0.0, count=1)
-        # must close, or asyncore will hold on to the socket and subsequent tests will fail with 'Address not in use'
+        # Must close, or asyncore will hold on to the socket and subsequent
+        # tests will fail with 'Address not in use'.
         self.close()
 
     def start(self):
         self.message_queue = Queue.Queue()
         self._stop = threading.Event()
         self._thread = threading.Thread(target=self.thread_proc)
-        self._thread.setDaemon(True) # ensures that if any tests do fail, then threads will exit when the parent exits
+        # Ensures that if any tests do fail, then threads will exit when the
+        # parent exits.
+        self._thread.setDaemon(True)
         self._thread.start()
 
     @classmethod
     def startServer(cls):
-        #print("Starting SMTP server on %s:%s" % (LOCAL_SMTP_SERVER_ADDRESS, LOCAL_SMTP_SERVER_PORT))
-        server = EmailServer((LOCAL_SMTP_SERVER_ADDRESS, LOCAL_SMTP_SERVER_PORT), None)
+        #print("Starting SMTP server on %s:%s"
+        #      % (LOCAL_SMTP_SERVER_ADDRESS, LOCAL_SMTP_SERVER_PORT))
+        server = EmailServer((LOCAL_SMTP_SERVER_ADDRESS,
+                              LOCAL_SMTP_SERVER_PORT),
+                             None)
         server.start()
         return server
 
     def stop(self):
-        # signal thread_proc to stop
+        # Signal thread_proc to stop:
         self._stop.set()
-        # wait for thread_proc to return (shouldn't take long)
+        # Wait for thread_proc to return (shouldn't take long)
         self._thread.join()
         assert self._thread.is_alive() == False, "Thread is alive and kicking"
 
     def getAndCheckMessageContains(self, text, timeoutInSecs=2.0):
         #print("Checking for reponse")
         message = self.message_queue.get(block=True, timeout=timeoutInSecs)
-        assert message.find(text) != -1, "Message did not contain text \"%s\". Full message is:\n %s" % (text, message)
+        assert message.find(text) != -1, ("Message did not contain text '%s'."
+                                          "Full message is:\n %s"
+                                          % (text, message))
 
     def checkNoMessageReceived(self, timeoutInSecs=2.0):
         try:
@@ -79,27 +92,36 @@ class EmailServer(SMTPServer):
         assert False, "Found a message in the queue, but expected none"
 
 def sendMail(fromAddress):
-    #print("Connecting to %s:%d" % (BRIDGEDB_SMTP_SERVER_ADDRESS, BRIDGEDB_SMTP_SERVER_PORT))
-    client = smtplib.SMTP(BRIDGEDB_SMTP_SERVER_ADDRESS, BRIDGEDB_SMTP_SERVER_PORT)
+    #print("Connecting to %s:%d"
+    #      % (BRIDGEDB_SMTP_SERVER_ADDRESS, BRIDGEDB_SMTP_SERVER_PORT))
+    client = smtplib.SMTP(BRIDGEDB_SMTP_SERVER_ADDRESS,
+                          BRIDGEDB_SMTP_SERVER_PORT)
     client.set_debuglevel(SMTP_DEBUG_LEVEL)
 
-    #print("Sending mail TO:%s, FROM:%s" % (TO_ADDRESS, fromAddress))
-    result = client.sendmail(fromAddress, TO_ADDRESS, MESSAGE_TEMPLATE % (fromAddress, TO_ADDRESS))
+    #print("Sending mail TO:%s, FROM:%s"
+    #      % (TO_ADDRESS, fromAddress))
+    result = client.sendmail(fromAddress, TO_ADDRESS,
+                             MESSAGE_TEMPLATE % (fromAddress, TO_ADDRESS))
     assert result == {}, "Failed to send mail"
     client.quit()
 
+
 class SMTPTests(unittest.TestCase):
     def setUp(self):
-        ''' Called at the start of each test, ensures that the SMTP server is running'''
+        '''Called at the start of each test, ensures that the SMTP server is
+        running.
+        '''
         self.server = EmailServer.startServer()
 
     def tearDown(self):
-        ''' Called after each test, ensures that the SMTP server is cleaned up'''
+        '''Called after each test, ensures that the SMTP server is cleaned up.
+        '''
         self.server.stop()
 
     def test_getBridges(self):
         # send the mail to bridgedb, choosing a random email address
-        sendMail(fromAddress=FROM_ADDRESS_TEMPLATE % random.randint(MIN_FROM_ADDRESS, MAX_FROM_ADDRESS))
+        sendMail(fromAddress=FROM_ADDRESS_TEMPLATE
+                 % random.randint(MIN_FROM_ADDRESS, MAX_FROM_ADDRESS))
 
         # then check that our local SMTP server received a response
         # and that response contained some bridges
@@ -107,7 +129,8 @@ class SMTPTests(unittest.TestCase):
 
     def test_getBridges_rateLimitExceeded(self):
         # send the mail to bridgedb, choosing a random email address
-        FROM_ADDRESS = FROM_ADDRESS_TEMPLATE % random.randint(MIN_FROM_ADDRESS, MAX_FROM_ADDRESS)
+        FROM_ADDRESS = FROM_ADDRESS_TEMPLATE % random.randint(
+            MIN_FROM_ADDRESS, MAX_FROM_ADDRESS)
         sendMail(FROM_ADDRESS)
 
         # then check that our local SMTP server received a response
@@ -118,7 +141,8 @@ class SMTPTests(unittest.TestCase):
         sendMail(FROM_ADDRESS)
 
         # this time, the email response should not contain any bridges
-        self.server.getAndCheckMessageContains("You have exceeded the rate limit. Please slow down!")
+        self.server.getAndCheckMessageContains(
+            "You have exceeded the rate limit. Please slow down!")
 
         # then we send another request from the same email address
         sendMail(FROM_ADDRESS)
@@ -127,13 +151,17 @@ class SMTPTests(unittest.TestCase):
         self.server.checkNoMessageReceived(timeoutInSecs=1.0)
 
     def test_getBridges_stressTest(self):
-        ''' Sends a large number of emails in a short period of time, and checks that
-            a response is received for each message '''
+        '''Sends a large number of emails in a short period of time, and checks
+        that a response is received for each message.
+        '''
         NUM_MAILS = 100
         for i in range(NUM_MAILS):
-            # Note: if by chance two emails with the same FROM_ADDRESS are generated, this test will fail
-            # Setting 'MAX_FROM_ADDRESS' to be a high value reduces the probability of this occuring, but does not rule it out
-            sendMail(fromAddress=FROM_ADDRESS_TEMPLATE % random.randint(MIN_FROM_ADDRESS, MAX_FROM_ADDRESS))
+            # Note: if by chance two emails with the same FROM_ADDRESS are
+            # generated, this test will fail Setting 'MAX_FROM_ADDRESS' to be
+            # a high value reduces the probability of this occuring, but does
+            # not rule it out
+            sendMail(fromAddress=FROM_ADDRESS_TEMPLATE
+                     % random.randint(MIN_FROM_ADDRESS, MAX_FROM_ADDRESS))
 
         for i in range(NUM_MAILS):
             self.server.getAndCheckMessageContains("Here are your bridges")



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