+++ /dev/null
-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
projects / mirror_edk2.git / blobdiff