EmbeddedPkg: Extend NvVarStoreFormattedLib LIBRARY_CLASS

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

import asyncore\r
import unittest\r
import select\r
import os\r
import socket\r
import sys\r
import time\r
import warnings\r
import errno\r
\r
from test import test_support\r
from test.test_support import TESTFN, run_unittest, unlink\r
from StringIO import StringIO\r
\r
try:\r
    import threading\r
except ImportError:\r
    threading = None\r
\r
HOST = test_support.HOST\r
\r
class dummysocket:\r
    def __init__(self):\r
        self.closed = False\r
\r
    def close(self):\r
        self.closed = True\r
\r
    def fileno(self):\r
        return 42\r
\r
class dummychannel:\r
    def __init__(self):\r
        self.socket = dummysocket()\r
\r
    def close(self):\r
        self.socket.close()\r
\r
class exitingdummy:\r
    def __init__(self):\r
        pass\r
\r
    def handle_read_event(self):\r
        raise asyncore.ExitNow()\r
\r
    handle_write_event = handle_read_event\r
    handle_close = handle_read_event\r
    handle_expt_event = handle_read_event\r
\r
class crashingdummy:\r
    def __init__(self):\r
        self.error_handled = False\r
\r
    def handle_read_event(self):\r
        raise Exception()\r
\r
    handle_write_event = handle_read_event\r
    handle_close = handle_read_event\r
    handle_expt_event = handle_read_event\r
\r
    def handle_error(self):\r
        self.error_handled = True\r
\r
# used when testing senders; just collects what it gets until newline is sent\r
def capture_server(evt, buf, serv):\r
    try:\r
        serv.listen(5)\r
        conn, addr = serv.accept()\r
    except socket.timeout:\r
        pass\r
    else:\r
        n = 200\r
        while n > 0:\r
            r, w, e = select.select([conn], [], [])\r
            if r:\r
                data = conn.recv(10)\r
                # keep everything except for the newline terminator\r
                buf.write(data.replace('\n', ''))\r
                if '\n' in data:\r
                    break\r
            n -= 1\r
            time.sleep(0.01)\r
\r
        conn.close()\r
    finally:\r
        serv.close()\r
        evt.set()\r
\r
\r
class HelperFunctionTests(unittest.TestCase):\r
    def test_readwriteexc(self):\r
        # Check exception handling behavior of read, write and _exception\r
\r
        # check that ExitNow exceptions in the object handler method\r
        # bubbles all the way up through asyncore read/write/_exception calls\r
        tr1 = exitingdummy()\r
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)\r
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)\r
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)\r
\r
        # check that an exception other than ExitNow in the object handler\r
        # method causes the handle_error method to get called\r
        tr2 = crashingdummy()\r
        asyncore.read(tr2)\r
        self.assertEqual(tr2.error_handled, True)\r
\r
        tr2 = crashingdummy()\r
        asyncore.write(tr2)\r
        self.assertEqual(tr2.error_handled, True)\r
\r
        tr2 = crashingdummy()\r
        asyncore._exception(tr2)\r
        self.assertEqual(tr2.error_handled, True)\r
\r
    # asyncore.readwrite uses constants in the select module that\r
    # are not present in Windows systems (see this thread:\r
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)\r
    # These constants should be present as long as poll is available\r
\r
    @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')\r
    def test_readwrite(self):\r
        # Check that correct methods are called by readwrite()\r
\r
        attributes = ('read', 'expt', 'write', 'closed', 'error_handled')\r
\r
        expected = (\r
            (select.POLLIN, 'read'),\r
            (select.POLLPRI, 'expt'),\r
            (select.POLLOUT, 'write'),\r
            (select.POLLERR, 'closed'),\r
            (select.POLLHUP, 'closed'),\r
            (select.POLLNVAL, 'closed'),\r
            )\r
\r
        class testobj:\r
            def __init__(self):\r
                self.read = False\r
                self.write = False\r
                self.closed = False\r
                self.expt = False\r
                self.error_handled = False\r
\r
            def handle_read_event(self):\r
                self.read = True\r
\r
            def handle_write_event(self):\r
                self.write = True\r
\r
            def handle_close(self):\r
                self.closed = True\r
\r
            def handle_expt_event(self):\r
                self.expt = True\r
\r
            def handle_error(self):\r
                self.error_handled = True\r
\r
        for flag, expectedattr in expected:\r
            tobj = testobj()\r
            self.assertEqual(getattr(tobj, expectedattr), False)\r
            asyncore.readwrite(tobj, flag)\r
\r
            # Only the attribute modified by the routine we expect to be\r
            # called should be True.\r
            for attr in attributes:\r
                self.assertEqual(getattr(tobj, attr), attr==expectedattr)\r
\r
            # check that ExitNow exceptions in the object handler method\r
            # bubbles all the way up through asyncore readwrite call\r
            tr1 = exitingdummy()\r
            self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)\r
\r
            # check that an exception other than ExitNow in the object handler\r
            # method causes the handle_error method to get called\r
            tr2 = crashingdummy()\r
            self.assertEqual(tr2.error_handled, False)\r
            asyncore.readwrite(tr2, flag)\r
            self.assertEqual(tr2.error_handled, True)\r
\r
    def test_closeall(self):\r
        self.closeall_check(False)\r
\r
    def test_closeall_default(self):\r
        self.closeall_check(True)\r
\r
    def closeall_check(self, usedefault):\r
        # Check that close_all() closes everything in a given map\r
\r
        l = []\r
        testmap = {}\r
        for i in range(10):\r
            c = dummychannel()\r
            l.append(c)\r
            self.assertEqual(c.socket.closed, False)\r
            testmap[i] = c\r
\r
        if usedefault:\r
            socketmap = asyncore.socket_map\r
            try:\r
                asyncore.socket_map = testmap\r
                asyncore.close_all()\r
            finally:\r
                testmap, asyncore.socket_map = asyncore.socket_map, socketmap\r
        else:\r
            asyncore.close_all(testmap)\r
\r
        self.assertEqual(len(testmap), 0)\r
\r
        for c in l:\r
            self.assertEqual(c.socket.closed, True)\r
\r
    def test_compact_traceback(self):\r
        try:\r
            raise Exception("I don't like spam!")\r
        except:\r
            real_t, real_v, real_tb = sys.exc_info()\r
            r = asyncore.compact_traceback()\r
        else:\r
            self.fail("Expected exception")\r
\r
        (f, function, line), t, v, info = r\r
        self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')\r
        self.assertEqual(function, 'test_compact_traceback')\r
        self.assertEqual(t, real_t)\r
        self.assertEqual(v, real_v)\r
        self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))\r
\r
\r
class DispatcherTests(unittest.TestCase):\r
    def setUp(self):\r
        pass\r
\r
    def tearDown(self):\r
        asyncore.close_all()\r
\r
    def test_basic(self):\r
        d = asyncore.dispatcher()\r
        self.assertEqual(d.readable(), True)\r
        self.assertEqual(d.writable(), True)\r
\r
    def test_repr(self):\r
        d = asyncore.dispatcher()\r
        self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))\r
\r
    def test_log(self):\r
        d = asyncore.dispatcher()\r
\r
        # capture output of dispatcher.log() (to stderr)\r
        fp = StringIO()\r
        stderr = sys.stderr\r
        l1 = "Lovely spam! Wonderful spam!"\r
        l2 = "I don't like spam!"\r
        try:\r
            sys.stderr = fp\r
            d.log(l1)\r
            d.log(l2)\r
        finally:\r
            sys.stderr = stderr\r
\r
        lines = fp.getvalue().splitlines()\r
        self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2])\r
\r
    def test_log_info(self):\r
        d = asyncore.dispatcher()\r
\r
        # capture output of dispatcher.log_info() (to stdout via print)\r
        fp = StringIO()\r
        stdout = sys.stdout\r
        l1 = "Have you got anything without spam?"\r
        l2 = "Why can't she have egg bacon spam and sausage?"\r
        l3 = "THAT'S got spam in it!"\r
        try:\r
            sys.stdout = fp\r
            d.log_info(l1, 'EGGS')\r
            d.log_info(l2)\r
            d.log_info(l3, 'SPAM')\r
        finally:\r
            sys.stdout = stdout\r
\r
        lines = fp.getvalue().splitlines()\r
        expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]\r
\r
        self.assertEqual(lines, expected)\r
\r
    def test_unhandled(self):\r
        d = asyncore.dispatcher()\r
        d.ignore_log_types = ()\r
\r
        # capture output of dispatcher.log_info() (to stdout via print)\r
        fp = StringIO()\r
        stdout = sys.stdout\r
        try:\r
            sys.stdout = fp\r
            d.handle_expt()\r
            d.handle_read()\r
            d.handle_write()\r
            d.handle_connect()\r
            d.handle_accept()\r
        finally:\r
            sys.stdout = stdout\r
\r
        lines = fp.getvalue().splitlines()\r
        expected = ['warning: unhandled incoming priority event',\r
                    'warning: unhandled read event',\r
                    'warning: unhandled write event',\r
                    'warning: unhandled connect event',\r
                    'warning: unhandled accept event']\r
        self.assertEqual(lines, expected)\r
\r
    def test_issue_8594(self):\r
        # XXX - this test is supposed to be removed in next major Python\r
        # version\r
        d = asyncore.dispatcher(socket.socket())\r
        # make sure the error message no longer refers to the socket\r
        # object but the dispatcher instance instead\r
        self.assertRaisesRegexp(AttributeError, 'dispatcher instance',\r
                                getattr, d, 'foo')\r
        # cheap inheritance with the underlying socket is supposed\r
        # to still work but a DeprecationWarning is expected\r
        with warnings.catch_warnings(record=True) as w:\r
            warnings.simplefilter("always")\r
            family = d.family\r
            self.assertEqual(family, socket.AF_INET)\r
            self.assertEqual(len(w), 1)\r
            self.assertTrue(issubclass(w[0].category, DeprecationWarning))\r
\r
    def test_strerror(self):\r
        # refers to bug #8573\r
        err = asyncore._strerror(errno.EPERM)\r
        if hasattr(os, 'strerror'):\r
            self.assertEqual(err, os.strerror(errno.EPERM))\r
        err = asyncore._strerror(-1)\r
        self.assertTrue(err != "")\r
\r
\r
class dispatcherwithsend_noread(asyncore.dispatcher_with_send):\r
    def readable(self):\r
        return False\r
\r
    def handle_connect(self):\r
        pass\r
\r
class DispatcherWithSendTests(unittest.TestCase):\r
    usepoll = False\r
\r
    def setUp(self):\r
        pass\r
\r
    def tearDown(self):\r
        asyncore.close_all()\r
\r
    @unittest.skipUnless(threading, 'Threading required for this test.')\r
    @test_support.reap_threads\r
    def test_send(self):\r
        evt = threading.Event()\r
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\r
        sock.settimeout(3)\r
        port = test_support.bind_port(sock)\r
\r
        cap = StringIO()\r
        args = (evt, cap, sock)\r
        t = threading.Thread(target=capture_server, args=args)\r
        t.start()\r
        try:\r
            # wait a little longer for the server to initialize (it sometimes\r
            # refuses connections on slow machines without this wait)\r
            time.sleep(0.2)\r
\r
            data = "Suppose there isn't a 16-ton weight?"\r
            d = dispatcherwithsend_noread()\r
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
            d.connect((HOST, port))\r
\r
            # give time for socket to connect\r
            time.sleep(0.1)\r
\r
            d.send(data)\r
            d.send(data)\r
            d.send('\n')\r
\r
            n = 1000\r
            while d.out_buffer and n > 0:\r
                asyncore.poll()\r
                n -= 1\r
\r
            evt.wait()\r
\r
            self.assertEqual(cap.getvalue(), data*2)\r
        finally:\r
            t.join()\r
\r
\r
class DispatcherWithSendTests_UsePoll(DispatcherWithSendTests):\r
    usepoll = True\r
\r
@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),\r
                     'asyncore.file_wrapper required')\r
class FileWrapperTest(unittest.TestCase):\r
    def setUp(self):\r
        self.d = "It's not dead, it's sleeping!"\r
        with file(TESTFN, 'w') as h:\r
            h.write(self.d)\r
\r
    def tearDown(self):\r
        unlink(TESTFN)\r
\r
    def test_recv(self):\r
        fd = os.open(TESTFN, os.O_RDONLY)\r
        w = asyncore.file_wrapper(fd)\r
        os.close(fd)\r
\r
        self.assertNotEqual(w.fd, fd)\r
        self.assertNotEqual(w.fileno(), fd)\r
        self.assertEqual(w.recv(13), "It's not dead")\r
        self.assertEqual(w.read(6), ", it's")\r
        w.close()\r
        self.assertRaises(OSError, w.read, 1)\r
\r
\r
    def test_send(self):\r
        d1 = "Come again?"\r
        d2 = "I want to buy some cheese."\r
        fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)\r
        w = asyncore.file_wrapper(fd)\r
        os.close(fd)\r
\r
        w.write(d1)\r
        w.send(d2)\r
        w.close()\r
        self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)\r
\r
    @unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),\r
                         'asyncore.file_dispatcher required')\r
    def test_dispatcher(self):\r
        fd = os.open(TESTFN, os.O_RDONLY)\r
        data = []\r
        class FileDispatcher(asyncore.file_dispatcher):\r
            def handle_read(self):\r
                data.append(self.recv(29))\r
        s = FileDispatcher(fd)\r
        os.close(fd)\r
        asyncore.loop(timeout=0.01, use_poll=True, count=2)\r
        self.assertEqual(b"".join(data), self.d)\r
\r
\r
class BaseTestHandler(asyncore.dispatcher):\r
\r
    def __init__(self, sock=None):\r
        asyncore.dispatcher.__init__(self, sock)\r
        self.flag = False\r
\r
    def handle_accept(self):\r
        raise Exception("handle_accept not supposed to be called")\r
\r
    def handle_connect(self):\r
        raise Exception("handle_connect not supposed to be called")\r
\r
    def handle_expt(self):\r
        raise Exception("handle_expt not supposed to be called")\r
\r
    def handle_close(self):\r
        raise Exception("handle_close not supposed to be called")\r
\r
    def handle_error(self):\r
        raise\r
\r
\r
class TCPServer(asyncore.dispatcher):\r
    """A server which listens on an address and dispatches the\r
    connection to a handler.\r
    """\r
\r
    def __init__(self, handler=BaseTestHandler, host=HOST, port=0):\r
        asyncore.dispatcher.__init__(self)\r
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
        self.set_reuse_addr()\r
        self.bind((host, port))\r
        self.listen(5)\r
        self.handler = handler\r
\r
    @property\r
    def address(self):\r
        return self.socket.getsockname()[:2]\r
\r
    def handle_accept(self):\r
        sock, addr = self.accept()\r
        self.handler(sock)\r
\r
    def handle_error(self):\r
        raise\r
\r
\r
class BaseClient(BaseTestHandler):\r
\r
    def __init__(self, address):\r
        BaseTestHandler.__init__(self)\r
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
        self.connect(address)\r
\r
    def handle_connect(self):\r
        pass\r
\r
\r
class BaseTestAPI(unittest.TestCase):\r
\r
    def tearDown(self):\r
        asyncore.close_all()\r
\r
    def loop_waiting_for_flag(self, instance, timeout=5):\r
        timeout = float(timeout) / 100\r
        count = 100\r
        while asyncore.socket_map and count > 0:\r
            asyncore.loop(timeout=0.01, count=1, use_poll=self.use_poll)\r
            if instance.flag:\r
                return\r
            count -= 1\r
            time.sleep(timeout)\r
        self.fail("flag not set")\r
\r
    def test_handle_connect(self):\r
        # make sure handle_connect is called on connect()\r
\r
        class TestClient(BaseClient):\r
            def handle_connect(self):\r
                self.flag = True\r
\r
        server = TCPServer()\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    def test_handle_accept(self):\r
        # make sure handle_accept() is called when a client connects\r
\r
        class TestListener(BaseTestHandler):\r
\r
            def __init__(self):\r
                BaseTestHandler.__init__(self)\r
                self.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
                self.bind((HOST, 0))\r
                self.listen(5)\r
                self.address = self.socket.getsockname()[:2]\r
\r
            def handle_accept(self):\r
                self.flag = True\r
\r
        server = TestListener()\r
        client = BaseClient(server.address)\r
        self.loop_waiting_for_flag(server)\r
\r
    def test_handle_read(self):\r
        # make sure handle_read is called on data received\r
\r
        class TestClient(BaseClient):\r
            def handle_read(self):\r
                self.flag = True\r
\r
        class TestHandler(BaseTestHandler):\r
            def __init__(self, conn):\r
                BaseTestHandler.__init__(self, conn)\r
                self.send('x' * 1024)\r
\r
        server = TCPServer(TestHandler)\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    def test_handle_write(self):\r
        # make sure handle_write is called\r
\r
        class TestClient(BaseClient):\r
            def handle_write(self):\r
                self.flag = True\r
\r
        server = TCPServer()\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    def test_handle_close(self):\r
        # make sure handle_close is called when the other end closes\r
        # the connection\r
\r
        class TestClient(BaseClient):\r
\r
            def handle_read(self):\r
                # in order to make handle_close be called we are supposed\r
                # to make at least one recv() call\r
                self.recv(1024)\r
\r
            def handle_close(self):\r
                self.flag = True\r
                self.close()\r
\r
        class TestHandler(BaseTestHandler):\r
            def __init__(self, conn):\r
                BaseTestHandler.__init__(self, conn)\r
                self.close()\r
\r
        server = TCPServer(TestHandler)\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    @unittest.skipIf(sys.platform.startswith("sunos"),\r
                     "OOB support is broken on Solaris")\r
    def test_handle_expt(self):\r
        # Make sure handle_expt is called on OOB data received.\r
        # Note: this might fail on some platforms as OOB data is\r
        # tenuously supported and rarely used.\r
\r
        class TestClient(BaseClient):\r
            def handle_expt(self):\r
                self.flag = True\r
\r
        class TestHandler(BaseTestHandler):\r
            def __init__(self, conn):\r
                BaseTestHandler.__init__(self, conn)\r
                self.socket.send(chr(244), socket.MSG_OOB)\r
\r
        server = TCPServer(TestHandler)\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    def test_handle_error(self):\r
\r
        class TestClient(BaseClient):\r
            def handle_write(self):\r
                1.0 / 0\r
            def handle_error(self):\r
                self.flag = True\r
                try:\r
                    raise\r
                except ZeroDivisionError:\r
                    pass\r
                else:\r
                    raise Exception("exception not raised")\r
\r
        server = TCPServer()\r
        client = TestClient(server.address)\r
        self.loop_waiting_for_flag(client)\r
\r
    def test_connection_attributes(self):\r
        server = TCPServer()\r
        client = BaseClient(server.address)\r
\r
        # we start disconnected\r
        self.assertFalse(server.connected)\r
        self.assertTrue(server.accepting)\r
        # this can't be taken for granted across all platforms\r
        #self.assertFalse(client.connected)\r
        self.assertFalse(client.accepting)\r
\r
        # execute some loops so that client connects to server\r
        asyncore.loop(timeout=0.01, use_poll=self.use_poll, count=100)\r
        self.assertFalse(server.connected)\r
        self.assertTrue(server.accepting)\r
        self.assertTrue(client.connected)\r
        self.assertFalse(client.accepting)\r
\r
        # disconnect the client\r
        client.close()\r
        self.assertFalse(server.connected)\r
        self.assertTrue(server.accepting)\r
        self.assertFalse(client.connected)\r
        self.assertFalse(client.accepting)\r
\r
        # stop serving\r
        server.close()\r
        self.assertFalse(server.connected)\r
        self.assertFalse(server.accepting)\r
\r
    def test_create_socket(self):\r
        s = asyncore.dispatcher()\r
        s.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
        self.assertEqual(s.socket.family, socket.AF_INET)\r
        self.assertEqual(s.socket.type, socket.SOCK_STREAM)\r
\r
    def test_bind(self):\r
        s1 = asyncore.dispatcher()\r
        s1.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
        s1.bind((HOST, 0))\r
        s1.listen(5)\r
        port = s1.socket.getsockname()[1]\r
\r
        s2 = asyncore.dispatcher()\r
        s2.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
        # EADDRINUSE indicates the socket was correctly bound\r
        self.assertRaises(socket.error, s2.bind, (HOST, port))\r
\r
    def test_set_reuse_addr(self):\r
        sock = socket.socket()\r
        try:\r
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\r
        except socket.error:\r
            unittest.skip("SO_REUSEADDR not supported on this platform")\r
        else:\r
            # if SO_REUSEADDR succeeded for sock we expect asyncore\r
            # to do the same\r
            s = asyncore.dispatcher(socket.socket())\r
            self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,\r
                                                 socket.SO_REUSEADDR))\r
            s.create_socket(socket.AF_INET, socket.SOCK_STREAM)\r
            s.set_reuse_addr()\r
            self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,\r
                                                 socket.SO_REUSEADDR))\r
        finally:\r
            sock.close()\r
\r
\r
class TestAPI_UseSelect(BaseTestAPI):\r
    use_poll = False\r
\r
@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')\r
class TestAPI_UsePoll(BaseTestAPI):\r
    use_poll = True\r
\r
\r
def test_main():\r
    tests = [HelperFunctionTests, DispatcherTests, DispatcherWithSendTests,\r
             DispatcherWithSendTests_UsePoll, TestAPI_UseSelect,\r
             TestAPI_UsePoll, FileWrapperTest]\r
    run_unittest(*tests)\r
\r
if __name__ == "__main__":\r
    test_main()\r