EmbeddedPkg: Extend NvVarStoreFormattedLib LIBRARY_CLASS

[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.2 / Lib / test / test_imaplib.py

from test import test_support as support\r
# If we end up with a significant number of tests that don't require\r
# threading, this test module should be split.  Right now we skip\r
# them all if we don't have threading.\r
threading = support.import_module('threading')\r
\r
from contextlib import contextmanager\r
import imaplib\r
import os.path\r
import SocketServer\r
import time\r
\r
from test_support import reap_threads, verbose, transient_internet\r
import unittest\r
\r
try:\r
    import ssl\r
except ImportError:\r
    ssl = None\r
\r
CERTFILE = None\r
\r
\r
class TestImaplib(unittest.TestCase):\r
\r
    def test_that_Time2Internaldate_returns_a_result(self):\r
        # We can check only that it successfully produces a result,\r
        # not the correctness of the result itself, since the result\r
        # depends on the timezone the machine is in.\r
        timevalues = [2000000000, 2000000000.0, time.localtime(2000000000),\r
                      '"18-May-2033 05:33:20 +0200"']\r
\r
        for t in timevalues:\r
            imaplib.Time2Internaldate(t)\r
\r
\r
if ssl:\r
\r
    class SecureTCPServer(SocketServer.TCPServer):\r
\r
        def get_request(self):\r
            newsocket, fromaddr = self.socket.accept()\r
            connstream = ssl.wrap_socket(newsocket,\r
                                         server_side=True,\r
                                         certfile=CERTFILE)\r
            return connstream, fromaddr\r
\r
    IMAP4_SSL = imaplib.IMAP4_SSL\r
\r
else:\r
\r
    class SecureTCPServer:\r
        pass\r
\r
    IMAP4_SSL = None\r
\r
\r
class SimpleIMAPHandler(SocketServer.StreamRequestHandler):\r
\r
    timeout = 1\r
\r
    def _send(self, message):\r
        if verbose: print "SENT:", message.strip()\r
        self.wfile.write(message)\r
\r
    def handle(self):\r
        # Send a welcome message.\r
        self._send('* OK IMAP4rev1\r\n')\r
        while 1:\r
            # Gather up input until we receive a line terminator or we timeout.\r
            # Accumulate read(1) because it's simpler to handle the differences\r
            # between naked sockets and SSL sockets.\r
            line = ''\r
            while 1:\r
                try:\r
                    part = self.rfile.read(1)\r
                    if part == '':\r
                        # Naked sockets return empty strings..\r
                        return\r
                    line += part\r
                except IOError:\r
                    # ..but SSLSockets throw exceptions.\r
                    return\r
                if line.endswith('\r\n'):\r
                    break\r
\r
            if verbose: print 'GOT:', line.strip()\r
            splitline = line.split()\r
            tag = splitline[0]\r
            cmd = splitline[1]\r
            args = splitline[2:]\r
\r
            if hasattr(self, 'cmd_%s' % (cmd,)):\r
                getattr(self, 'cmd_%s' % (cmd,))(tag, args)\r
            else:\r
                self._send('%s BAD %s unknown\r\n' % (tag, cmd))\r
\r
    def cmd_CAPABILITY(self, tag, args):\r
        self._send('* CAPABILITY IMAP4rev1\r\n')\r
        self._send('%s OK CAPABILITY completed\r\n' % (tag,))\r
\r
\r
class BaseThreadedNetworkedTests(unittest.TestCase):\r
\r
    def make_server(self, addr, hdlr):\r
\r
        class MyServer(self.server_class):\r
            def handle_error(self, request, client_address):\r
                self.close_request(request)\r
                self.server_close()\r
                raise\r
\r
        if verbose: print "creating server"\r
        server = MyServer(addr, hdlr)\r
        self.assertEqual(server.server_address, server.socket.getsockname())\r
\r
        if verbose:\r
            print "server created"\r
            print "ADDR =", addr\r
            print "CLASS =", self.server_class\r
            print "HDLR =", server.RequestHandlerClass\r
\r
        t = threading.Thread(\r
            name='%s serving' % self.server_class,\r
            target=server.serve_forever,\r
            # Short poll interval to make the test finish quickly.\r
            # Time between requests is short enough that we won't wake\r
            # up spuriously too many times.\r
            kwargs={'poll_interval':0.01})\r
        t.daemon = True  # In case this function raises.\r
        t.start()\r
        if verbose: print "server running"\r
        return server, t\r
\r
    def reap_server(self, server, thread):\r
        if verbose: print "waiting for server"\r
        server.shutdown()\r
        thread.join()\r
        if verbose: print "done"\r
\r
    @contextmanager\r
    def reaped_server(self, hdlr):\r
        server, thread = self.make_server((support.HOST, 0), hdlr)\r
        try:\r
            yield server\r
        finally:\r
            self.reap_server(server, thread)\r
\r
    @reap_threads\r
    def test_connect(self):\r
        with self.reaped_server(SimpleIMAPHandler) as server:\r
            client = self.imap_class(*server.server_address)\r
            client.shutdown()\r
\r
    @reap_threads\r
    def test_issue5949(self):\r
\r
        class EOFHandler(SocketServer.StreamRequestHandler):\r
            def handle(self):\r
                # EOF without sending a complete welcome message.\r
                self.wfile.write('* OK')\r
\r
        with self.reaped_server(EOFHandler) as server:\r
            self.assertRaises(imaplib.IMAP4.abort,\r
                              self.imap_class, *server.server_address)\r
\r
\r
class ThreadedNetworkedTests(BaseThreadedNetworkedTests):\r
\r
    server_class = SocketServer.TCPServer\r
    imap_class = imaplib.IMAP4\r
\r
\r
@unittest.skipUnless(ssl, "SSL not available")\r
class ThreadedNetworkedTestsSSL(BaseThreadedNetworkedTests):\r
\r
    server_class = SecureTCPServer\r
    imap_class = IMAP4_SSL\r
\r
\r
class RemoteIMAPTest(unittest.TestCase):\r
    host = 'cyrus.andrew.cmu.edu'\r
    port = 143\r
    username = 'anonymous'\r
    password = 'pass'\r
    imap_class = imaplib.IMAP4\r
\r
    def setUp(self):\r
        with transient_internet(self.host):\r
            self.server = self.imap_class(self.host, self.port)\r
\r
    def tearDown(self):\r
        if self.server is not None:\r
            self.server.logout()\r
\r
    def test_logincapa(self):\r
        self.assertTrue('LOGINDISABLED' in self.server.capabilities)\r
\r
    def test_anonlogin(self):\r
        self.assertTrue('AUTH=ANONYMOUS' in self.server.capabilities)\r
        rs = self.server.login(self.username, self.password)\r
        self.assertEqual(rs[0], 'OK')\r
\r
    def test_logout(self):\r
        rs = self.server.logout()\r
        self.server = None\r
        self.assertEqual(rs[0], 'BYE')\r
\r
\r
@unittest.skipUnless(ssl, "SSL not available")\r
class RemoteIMAP_SSLTest(RemoteIMAPTest):\r
    port = 993\r
    imap_class = IMAP4_SSL\r
\r
    def test_logincapa(self):\r
        self.assertFalse('LOGINDISABLED' in self.server.capabilities)\r
        self.assertTrue('AUTH=PLAIN' in self.server.capabilities)\r
\r
\r
def test_main():\r
    tests = [TestImaplib]\r
\r
    if support.is_resource_enabled('network'):\r
        if ssl:\r
            global CERTFILE\r
            CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,\r
                                    "keycert.pem")\r
            if not os.path.exists(CERTFILE):\r
                raise support.TestFailed("Can't read certificate files!")\r
        tests.extend([\r
            ThreadedNetworkedTests, ThreadedNetworkedTestsSSL,\r
            RemoteIMAPTest, RemoteIMAP_SSLTest,\r
        ])\r
\r
    support.run_unittest(*tests)\r
\r
\r
if __name__ == "__main__":\r
    support.use_resources = ['network']\r
    test_main()\r