1 from test
import test_support
as support
2 # If we end up with a significant number of tests that don't require
3 # threading, this test module should be split. Right now we skip
4 # them all if we don't have threading.
5 threading
= support
.import_module('threading')
7 from contextlib
import contextmanager
13 from test_support
import reap_threads
, verbose
24 class TestImaplib(unittest
.TestCase
):
26 def test_that_Time2Internaldate_returns_a_result(self
):
27 # We can check only that it successfully produces a result,
28 # not the correctness of the result itself, since the result
29 # depends on the timezone the machine is in.
30 timevalues
= [2000000000, 2000000000.0, time
.localtime(2000000000),
31 '"18-May-2033 05:33:20 +0200"']
34 imaplib
.Time2Internaldate(t
)
39 class SecureTCPServer(SocketServer
.TCPServer
):
41 def get_request(self
):
42 newsocket
, fromaddr
= self
.socket
.accept()
43 connstream
= ssl
.wrap_socket(newsocket
,
46 return connstream
, fromaddr
48 IMAP4_SSL
= imaplib
.IMAP4_SSL
52 class SecureTCPServer
:
58 class SimpleIMAPHandler(SocketServer
.StreamRequestHandler
):
62 def _send(self
, message
):
63 if verbose
: print "SENT:", message
.strip()
64 self
.wfile
.write(message
)
67 # Send a welcome message.
68 self
._send
('* OK IMAP4rev1\r\n')
70 # Gather up input until we receive a line terminator or we timeout.
71 # Accumulate read(1) because it's simpler to handle the differences
72 # between naked sockets and SSL sockets.
76 part
= self
.rfile
.read(1)
78 # Naked sockets return empty strings..
82 # ..but SSLSockets throw exceptions.
84 if line
.endswith('\r\n'):
87 if verbose
: print 'GOT:', line
.strip()
88 splitline
= line
.split()
93 if hasattr(self
, 'cmd_%s' % (cmd
,)):
94 getattr(self
, 'cmd_%s' % (cmd
,))(tag
, args
)
96 self
._send
('%s BAD %s unknown\r\n' % (tag
, cmd
))
98 def cmd_CAPABILITY(self
, tag
, args
):
99 self
._send
('* CAPABILITY IMAP4rev1\r\n')
100 self
._send
('%s OK CAPABILITY completed\r\n' % (tag
,))
103 class BaseThreadedNetworkedTests(unittest
.TestCase
):
105 def make_server(self
, addr
, hdlr
):
107 class MyServer(self
.server_class
):
108 def handle_error(self
, request
, client_address
):
109 self
.close_request(request
)
113 if verbose
: print "creating server"
114 server
= MyServer(addr
, hdlr
)
115 self
.assertEquals(server
.server_address
, server
.socket
.getsockname())
118 print "server created"
120 print "CLASS =", self
.server_class
121 print "HDLR =", server
.RequestHandlerClass
123 t
= threading
.Thread(
124 name
='%s serving' % self
.server_class
,
125 target
=server
.serve_forever
,
126 # Short poll interval to make the test finish quickly.
127 # Time between requests is short enough that we won't wake
128 # up spuriously too many times.
129 kwargs
={'poll_interval':0.01})
130 t
.daemon
= True # In case this function raises.
132 if verbose
: print "server running"
135 def reap_server(self
, server
, thread
):
136 if verbose
: print "waiting for server"
139 if verbose
: print "done"
142 def reaped_server(self
, hdlr
):
143 server
, thread
= self
.make_server((support
.HOST
, 0), hdlr
)
147 self
.reap_server(server
, thread
)
150 def test_connect(self
):
151 with self
.reaped_server(SimpleIMAPHandler
) as server
:
152 client
= self
.imap_class(*server
.server_address
)
156 def test_issue5949(self
):
158 class EOFHandler(SocketServer
.StreamRequestHandler
):
160 # EOF without sending a complete welcome message.
161 self
.wfile
.write('* OK')
163 with self
.reaped_server(EOFHandler
) as server
:
164 self
.assertRaises(imaplib
.IMAP4
.abort
,
165 self
.imap_class
, *server
.server_address
)
168 class ThreadedNetworkedTests(BaseThreadedNetworkedTests
):
170 server_class
= SocketServer
.TCPServer
171 imap_class
= imaplib
.IMAP4
174 @unittest.skipUnless(ssl
, "SSL not available")
175 class ThreadedNetworkedTestsSSL(BaseThreadedNetworkedTests
):
177 server_class
= SecureTCPServer
178 imap_class
= IMAP4_SSL
183 tests
= [TestImaplib
]
185 if support
.is_resource_enabled('network'):
188 CERTFILE
= os
.path
.join(os
.path
.dirname(__file__
) or os
.curdir
,
190 if not os
.path
.exists(CERTFILE
):
191 raise support
.TestFailed("Can't read certificate files!")
192 tests
.extend([ThreadedNetworkedTests
, ThreadedNetworkedTestsSSL
])
194 support
.run_unittest(*tests
)
197 if __name__
== "__main__":
198 support
.use_resources
= ['network']