[Spambayes-checkins] spambayes/spambayes/test test_sb_server.py, 1.1, 1.2

[Tony Meyer]

Update of /cvsroot/spambayes/spambayes/spambayes/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv7591/spambayes/test

Modified Files:
	test_sb_server.py 
Log Message:
Implement [ 870524 ] Make the message-proxy timeout configurable

Also add a test for it in test_sb_server (this does vastly increase the time that
 that test script takes to run, because it has to wait for the timeout).

Use email.message_from_string(text, _class) rather than our wrapper functions, to
 avoid the Python 2.4 DeprecationWarnings about the strict argument.

Index: test_sb_server.py
===================================================================
RCS file: /cvsroot/spambayes/spambayes/spambayes/test/test_sb_server.py,v
retrieving revision 1.1
retrieving revision 1.2
diff -C2 -d -r1.1 -r1.2
*** test_sb_server.py	5 Nov 2004 02:34:28 -0000	1.1
--- test_sb_server.py	9 Nov 2004 02:37:41 -0000	1.2
***************
*** 81,84 ****
--- 81,85 ----
  import operator
  import re
+ import time
  import getopt
  import sys, os
***************
*** 113,117 ****
      UIDL.  USER, PASS, APOP, DELE and RSET simply return "+OK"
      without doing anything.  Also understands the 'KILL' command, to
!     kill it.  The mail content is the example messages above.
      """
  
--- 114,119 ----
      UIDL.  USER, PASS, APOP, DELE and RSET simply return "+OK"
      without doing anything.  Also understands the 'KILL' command, to
!     kill it, and a 'SLOW' command, to change to really slow retrieval.
!     The mail content is the example messages above.
      """
  
***************
*** 123,127 ****
          self.maildrop = [spam1, good1]
          self.set_terminator('\r\n')
!         self.okCommands = ['USER', 'PASS', 'APOP', 'NOOP',
                             'DELE', 'RSET', 'QUIT', 'KILL']
          self.handlers = {'CAPA': self.onCapa,
--- 125,129 ----
          self.maildrop = [spam1, good1]
          self.set_terminator('\r\n')
!         self.okCommands = ['USER', 'PASS', 'APOP', 'NOOP', 'SLOW',
                             'DELE', 'RSET', 'QUIT', 'KILL']
          self.handlers = {'CAPA': self.onCapa,
***************
*** 132,135 ****
--- 134,138 ----
          self.push("+OK ready\r\n")
          self.request = ''
+         self.push_delay = 0.0 # 0.02 is a useful value for testing.
  
      def collect_incoming_data(self, data):
***************
*** 148,165 ****
              if command == 'QUIT':
                  self.close_when_done()
!             if command == 'KILL':
                  self.socket.shutdown(2)
                  self.close()
                  raise SystemExit
          else:
              handler = self.handlers.get(command, self.onUnknown)
!             self.push(handler(command, args))   # Or push_slowly for testing
          self.request = ''
  
      def push_slowly(self, response):
!         """Useful for testing."""
!         for c in response:
!             self.push(c)
!             time.sleep(0.02)
  
      def onCapa(self, command, args):
--- 151,179 ----
              if command == 'QUIT':
                  self.close_when_done()
!             elif command == 'KILL':
                  self.socket.shutdown(2)
                  self.close()
                  raise SystemExit
+             elif command == 'SLOW':
+                 self.push_delay = 1.0
          else:
              handler = self.handlers.get(command, self.onUnknown)
!             self.push_slowly(handler(command, args))
          self.request = ''
  
      def push_slowly(self, response):
!         """Sometimes we push out the response slowly to try and generate
!         timeouts.  If the delay is 0, this just does a regular push."""
!         if self.push_delay:
!             for c in response.split('\n'):
!                 if c and c[-1] == '\r':
!                     self.push(c + '\n')
!                 else:
!                     # We want to trigger onServerLine, so need the '\r',
!                     # so modify the message just a wee bit.
!                     self.push(c + '\r\n')
!                 time.sleep(self.push_delay * len(c))
!         else:
!             self.push(response)
  
      def onCapa(self, command, args):
***************
*** 291,295 ****
  
      # Ask for the capabilities via the proxy, and verify that the proxy
!     # is filtering out the PIPELINING capability.
      proxy.send("capa\r\n")
      response = proxy.recv(1000)
--- 305,309 ----
  
      # Ask for the capabilities via the proxy, and verify that the proxy
!     # is filtering out the STLS capability.
      proxy.send("capa\r\n")
      response = proxy.recv(1000)
***************
*** 311,314 ****
--- 325,341 ----
          assert response.find(options["Headers", "classification_header_name"]) >= 0
  
+     # Check that the proxy times out when it should.
+     options["pop3proxy", "retrieval_timeout"] = 30
+     options["Headers", "include_evidence"] = False
+     assert spam1.find('\n\n') > options["pop3proxy", "retrieval_timeout"]
+     print "This test is rather slow..."
+     proxy.send("slow\r\n")
+     response = proxy.recv(100)
+     assert response.find("OK") != -1
+     proxy.send("retr 1\r\n")
+     response = proxy.recv(1000)
+     assert len(response) < len(spam1)
+     print "Slow test done.  Thanks for waiting!"
+ 
      # Smoke-test the HTML UI.
      httpServer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)