mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-27 02:39:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

Skip some more tests in the absence of threading.

Browse source
This commit is contained in:
Vinay Sajip 2011-05-17 07:15:53 +01:00
parent 362ce37963
commit ce7c978140
1 changed files with 251 additions and 248 deletions
Download patch file Download diff file Expand all files Collapse all files

499
Lib/test/test_logging.py
View file

@ -25,24 +25,17 @@ import logging
import logging.handlers import logging.handlers
import logging.config import logging.config
import asynchat
import asyncore
import codecs import codecs
import datetime import datetime
import errno
import pickle import pickle
import io import io
import gc import gc
from http.server import HTTPServer, BaseHTTPRequestHandler
import json import json
import os import os
import queue import queue
import re import re
import select import select
import smtpd
import socket import socket
from socketserver import (ThreadingUDPServer, DatagramRequestHandler,
ThreadingTCPServer, StreamRequestHandler)
import struct import struct
import sys import sys
import tempfile import tempfile
@ -51,11 +44,19 @@ from test.support import TestHandler, Matcher
import textwrap import textwrap
import time import time
import unittest import unittest
from urllib.parse import urlparse, parse_qs
import warnings import warnings
import weakref import weakref
try: try:
import threading import threading
# The following imports are needed only for tests which
import asynchat
import asyncore
import errno
from http.server import HTTPServer, BaseHTTPRequestHandler
import smtpd
from urllib.parse import urlparse, parse_qs
from socketserver import (ThreadingUDPServer, DatagramRequestHandler,
ThreadingTCPServer, StreamRequestHandler)
except ImportError: except ImportError:
threading = None threading = None
try: try:
@ -611,284 +612,286 @@ class StreamHandlerTest(BaseTest):
# -- The following section could be moved into a server_helper.py module # -- The following section could be moved into a server_helper.py module
# -- if it proves to be of wider utility than just test_logging # -- if it proves to be of wider utility than just test_logging
class TestSMTPChannel(smtpd.SMTPChannel): if threading:
""" class TestSMTPChannel(smtpd.SMTPChannel):
This derived class has had to be created because smtpd does not
support use of custom channel maps, although they are allowed by
asyncore's design. Issue #11959 has been raised to address this,
and if resolved satisfactorily, some of this code can be removed.
"""
def __init__(self, server, conn, addr, sockmap):
asynchat.async_chat.__init__(self, conn, sockmap)
self.smtp_server = server
self.conn = conn
self.addr = addr
self.received_lines = []
self.smtp_state = self.COMMAND
self.seen_greeting = ''
self.mailfrom = None
self.rcpttos = []
self.received_data = ''
self.fqdn = socket.getfqdn()
self.num_bytes = 0
try:
self.peer = conn.getpeername()
except socket.error as err:
# a race condition may occur if the other end is closing
# before we can get the peername
self.close()
if err.args[0] != errno.ENOTCONN:
raise
return
self.push('220 %s %s' % (self.fqdn, smtpd.__version__))
self.set_terminator(b'\r\n')
class TestSMTPServer(smtpd.SMTPServer):
"""
This class implements a test SMTP server.
:param addr: A (host, port) tuple which the server listens on.
You can specify a port value of zero: the server's
*port* attribute will hold the actual port number
used, which can be used in client connections.
:param handler: A callable which will be called to process
incoming messages. The handler will be passed
the client address tuple, who the message is from,
a list of recipients and the message data.
:param poll_interval: The interval, in seconds, used in the underlying
:func:`select` or :func:`poll` call by
:func:`asyncore.loop`.
:param sockmap: A dictionary which will be used to hold
:class:`asyncore.dispatcher` instances used by
:func:`asyncore.loop`. This avoids changing the
:mod:`asyncore` module's global state.
"""
channel_class = TestSMTPChannel
def __init__(self, addr, handler, poll_interval, sockmap):
self._localaddr = addr
self._remoteaddr = None
self.sockmap = sockmap
asyncore.dispatcher.__init__(self, map=sockmap)
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(0)
self.set_socket(sock, map=sockmap)
# try to re-use a server port if possible
self.set_reuse_addr()
self.bind(addr)
self.port = sock.getsockname()[1]
self.listen(5)
except:
self.close()
raise
self._handler = handler
self._thread = None
self.poll_interval = poll_interval
def handle_accepted(self, conn, addr):
""" """
Redefined only because the base class does not pass in a This derived class has had to be created because smtpd does not
map, forcing use of a global in :mod:`asyncore`. support use of custom channel maps, although they are allowed by
asyncore's design. Issue #11959 has been raised to address this,
and if resolved satisfactorily, some of this code can be removed.
""" """
channel = self.channel_class(self, conn, addr, self.sockmap) def __init__(self, server, conn, addr, sockmap):
asynchat.async_chat.__init__(self, conn, sockmap)
self.smtp_server = server
self.conn = conn
self.addr = addr
self.received_lines = []
self.smtp_state = self.COMMAND
self.seen_greeting = ''
self.mailfrom = None
self.rcpttos = []
self.received_data = ''
self.fqdn = socket.getfqdn()
self.num_bytes = 0
try:
self.peer = conn.getpeername()
except socket.error as err:
# a race condition may occur if the other end is closing
# before we can get the peername
self.close()
if err.args[0] != errno.ENOTCONN:
raise
return
self.push('220 %s %s' % (self.fqdn, smtpd.__version__))
self.set_terminator(b'\r\n')
def process_message(self, peer, mailfrom, rcpttos, data):
"""
Delegates to the handler passed in to the server's constructor.
Typically, this will be a test case method. class TestSMTPServer(smtpd.SMTPServer):
:param peer: The client (host, port) tuple.
:param mailfrom: The address of the sender.
:param rcpttos: The addresses of the recipients.
:param data: The message.
""" """
self._handler(peer, mailfrom, rcpttos, data) This class implements a test SMTP server.
def start(self): :param addr: A (host, port) tuple which the server listens on.
""" You can specify a port value of zero: the server's
Start the server running on a separate daemon thread. *port* attribute will hold the actual port number
""" used, which can be used in client connections.
self._thread = t = threading.Thread(target=self.serve_forever, :param handler: A callable which will be called to process
args=(self.poll_interval,)) incoming messages. The handler will be passed
t.setDaemon(True) the client address tuple, who the message is from,
t.start() a list of recipients and the message data.
def serve_forever(self, poll_interval):
"""
Run the :mod:`asyncore` loop until normal termination
conditions arise.
:param poll_interval: The interval, in seconds, used in the underlying :param poll_interval: The interval, in seconds, used in the underlying
:func:`select` or :func:`poll` call by :func:`select` or :func:`poll` call by
:func:`asyncore.loop`. :func:`asyncore.loop`.
:param sockmap: A dictionary which will be used to hold
:class:`asyncore.dispatcher` instances used by
:func:`asyncore.loop`. This avoids changing the
:mod:`asyncore` module's global state.
""" """
try: channel_class = TestSMTPChannel
asyncore.loop(poll_interval, map=self.sockmap)
except select.error: def __init__(self, addr, handler, poll_interval, sockmap):
# On FreeBSD 8, closing the server repeatably self._localaddr = addr
# raises this error. We swallow it if the self._remoteaddr = None
# server has been closed. self.sockmap = sockmap
if self.connected or self.accepting: asyncore.dispatcher.__init__(self, map=sockmap)
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(0)
self.set_socket(sock, map=sockmap)
# try to re-use a server port if possible
self.set_reuse_addr()
self.bind(addr)
self.port = sock.getsockname()[1]
self.listen(5)
except:
self.close()
raise raise
self._handler = handler
self._thread = None
self.poll_interval = poll_interval
def stop(self, timeout=None): def handle_accepted(self, conn, addr):
""" """
Stop the thread by closing the server instance. Redefined only because the base class does not pass in a
Wait for the server thread to terminate. map, forcing use of a global in :mod:`asyncore`.
"""
channel = self.channel_class(self, conn, addr, self.sockmap)
:param timeout: How long to wait for the server thread def process_message(self, peer, mailfrom, rcpttos, data):
to terminate. """
""" Delegates to the handler passed in to the server's constructor.
self.close()
self._thread.join(timeout)
self._thread = None
class ControlMixin(object): Typically, this will be a test case method.
""" :param peer: The client (host, port) tuple.
This mixin is used to start a server on a separate thread, and :param mailfrom: The address of the sender.
shut it down programmatically. Request handling is simplified - instead :param rcpttos: The addresses of the recipients.
of needing to derive a suitable RequestHandler subclass, you just :param data: The message.
provide a callable which will be passed each received request to be """
processed. self._handler(peer, mailfrom, rcpttos, data)
:param handler: A handler callable which will be called with a def start(self):
single parameter - the request - in order to """
process the request. This handler is called on the Start the server running on a separate daemon thread.
server thread, effectively meaning that requests are """
processed serially. While not quite Web scale ;-), self._thread = t = threading.Thread(target=self.serve_forever,
this should be fine for testing applications. args=(self.poll_interval,))
:param poll_interval: The polling interval in seconds. t.setDaemon(True)
""" t.start()
def __init__(self, handler, poll_interval):
self._thread = None
self.poll_interval = poll_interval
self._handler = handler
self.ready = threading.Event()
def start(self): def serve_forever(self, poll_interval):
""" """
Create a daemon thread to run the server, and start it. Run the :mod:`asyncore` loop until normal termination
""" conditions arise.
self._thread = t = threading.Thread(target=self.serve_forever, :param poll_interval: The interval, in seconds, used in the underlying
args=(self.poll_interval,)) :func:`select` or :func:`poll` call by
t.setDaemon(True) :func:`asyncore.loop`.
t.start() """
try:
asyncore.loop(poll_interval, map=self.sockmap)
except select.error:
# On FreeBSD 8, closing the server repeatably
# raises this error. We swallow it if the
# server has been closed.
if self.connected or self.accepting:
raise
def serve_forever(self, poll_interval): def stop(self, timeout=None):
""" """
Run the server. Set the ready flag before entering the Stop the thread by closing the server instance.
service loop. Wait for the server thread to terminate.
"""
self.ready.set()
super(ControlMixin, self).serve_forever(poll_interval)
def stop(self, timeout=None): :param timeout: How long to wait for the server thread
""" to terminate.
Tell the server thread to stop, and wait for it to do so. """
self.close()
:param timeout: How long to wait for the server thread
to terminate.
"""
self.shutdown()
if self._thread is not None:
self._thread.join(timeout) self._thread.join(timeout)
self._thread = None self._thread = None
self.server_close()
self.ready.clear()
class TestHTTPServer(ControlMixin, HTTPServer): class ControlMixin(object):
""" """
An HTTP server which is controllable using :class:`ControlMixin`. This mixin is used to start a server on a separate thread, and
shut it down programmatically. Request handling is simplified - instead
of needing to derive a suitable RequestHandler subclass, you just
provide a callable which will be passed each received request to be
processed.
:param addr: A tuple with the IP address and port to listen on. :param handler: A handler callable which will be called with a
:param handler: A handler callable which will be called with a single parameter - the request - in order to
single parameter - the request - in order to process the request. This handler is called on the
process the request. server thread, effectively meaning that requests are
:param poll_interval: The polling interval in seconds. processed serially. While not quite Web scale ;-),
:param log: Pass ``True`` to enable log messages. this should be fine for testing applications.
""" :param poll_interval: The polling interval in seconds.
def __init__(self, addr, handler, poll_interval=0.5, log=False): """
class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler): def __init__(self, handler, poll_interval):
def __getattr__(self, name, default=None): self._thread = None
if name.startswith('do_'): self.poll_interval = poll_interval
return self.process_request self._handler = handler
raise AttributeError(name) self.ready = threading.Event()
def process_request(self): def start(self):
self.server._handler(self) """
Create a daemon thread to run the server, and start it.
"""
self._thread = t = threading.Thread(target=self.serve_forever,
args=(self.poll_interval,))
t.setDaemon(True)
t.start()
def log_message(self, format, *args): def serve_forever(self, poll_interval):
if log: """
super(DelegatingHTTPRequestHandler, Run the server. Set the ready flag before entering the
self).log_message(format, *args) service loop.
HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler) """
ControlMixin.__init__(self, handler, poll_interval) self.ready.set()
super(ControlMixin, self).serve_forever(poll_interval)
class TestTCPServer(ControlMixin, ThreadingTCPServer): def stop(self, timeout=None):
""" """
A TCP server which is controllable using :class:`ControlMixin`. Tell the server thread to stop, and wait for it to do so.
:param addr: A tuple with the IP address and port to listen on. :param timeout: How long to wait for the server thread
:param handler: A handler callable which will be called with a single to terminate.
parameter - the request - in order to process the request. """
:param poll_interval: The polling interval in seconds. self.shutdown()
:bind_and_activate: If True (the default), binds the server and starts it if self._thread is not None:
listening. If False, you need to call self._thread.join(timeout)
:meth:`server_bind` and :meth:`server_activate` at self._thread = None
some later time before calling :meth:`start`, so that self.server_close()
the server will set up the socket and listen on it. self.ready.clear()
"""
allow_reuse_address = True class TestHTTPServer(ControlMixin, HTTPServer):
"""
An HTTP server which is controllable using :class:`ControlMixin`.
def __init__(self, addr, handler, poll_interval=0.5, :param addr: A tuple with the IP address and port to listen on.
bind_and_activate=True): :param handler: A handler callable which will be called with a
class DelegatingTCPRequestHandler(StreamRequestHandler): single parameter - the request - in order to
process the request.
:param poll_interval: The polling interval in seconds.
:param log: Pass ``True`` to enable log messages.
"""
def __init__(self, addr, handler, poll_interval=0.5, log=False):
class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):
def __getattr__(self, name, default=None):
if name.startswith('do_'):
return self.process_request
raise AttributeError(name)
def handle(self): def process_request(self):
self.server._handler(self) self.server._handler(self)
ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,
bind_and_activate)
ControlMixin.__init__(self, handler, poll_interval)
def server_bind(self): def log_message(self, format, *args):
super(TestTCPServer, self).server_bind() if log:
self.port = self.socket.getsockname()[1] super(DelegatingHTTPRequestHandler,
self).log_message(format, *args)
HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)
ControlMixin.__init__(self, handler, poll_interval)
class TestUDPServer(ControlMixin, ThreadingUDPServer): class TestTCPServer(ControlMixin, ThreadingTCPServer):
""" """
A UDP server which is controllable using :class:`ControlMixin`. A TCP server which is controllable using :class:`ControlMixin`.
:param addr: A tuple with the IP address and port to listen on. :param addr: A tuple with the IP address and port to listen on.
:param handler: A handler callable which will be called with a :param handler: A handler callable which will be called with a single
single parameter - the request - in order to parameter - the request - in order to process the request.
process the request. :param poll_interval: The polling interval in seconds.
:param poll_interval: The polling interval for shutdown requests, :bind_and_activate: If True (the default), binds the server and starts it
in seconds. listening. If False, you need to call
:bind_and_activate: If True (the default), binds the server and :meth:`server_bind` and :meth:`server_activate` at
starts it listening. If False, you need to some later time before calling :meth:`start`, so that
call :meth:`server_bind` and the server will set up the socket and listen on it.
:meth:`server_activate` at some later time """
before calling :meth:`start`, so that the server will
set up the socket and listen on it.
"""
def __init__(self, addr, handler, poll_interval=0.5, bind_and_activate=True):
class DelegatingUDPRequestHandler(DatagramRequestHandler):
def handle(self): allow_reuse_address = True
self.server._handler(self)
ThreadingUDPServer.__init__(self, addr, DelegatingUDPRequestHandler,
bind_and_activate)
ControlMixin.__init__(self, handler, poll_interval)
def server_bind(self): def __init__(self, addr, handler, poll_interval=0.5,
super(TestUDPServer, self).server_bind() bind_and_activate=True):
self.port = self.socket.getsockname()[1] class DelegatingTCPRequestHandler(StreamRequestHandler):
def handle(self):
self.server._handler(self)
ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,
bind_and_activate)
ControlMixin.__init__(self, handler, poll_interval)
def server_bind(self):
super(TestTCPServer, self).server_bind()
self.port = self.socket.getsockname()[1]
class TestUDPServer(ControlMixin, ThreadingUDPServer):
"""
A UDP server which is controllable using :class:`ControlMixin`.
:param addr: A tuple with the IP address and port to listen on.
:param handler: A handler callable which will be called with a
single parameter - the request - in order to
process the request.
:param poll_interval: The polling interval for shutdown requests,
in seconds.
:bind_and_activate: If True (the default), binds the server and
starts it listening. If False, you need to
call :meth:`server_bind` and
:meth:`server_activate` at some later time
before calling :meth:`start`, so that the server will
set up the socket and listen on it.
"""
def __init__(self, addr, handler, poll_interval=0.5, bind_and_activate=True):
class DelegatingUDPRequestHandler(DatagramRequestHandler):
def handle(self):
self.server._handler(self)
ThreadingUDPServer.__init__(self, addr, DelegatingUDPRequestHandler,
bind_and_activate)
ControlMixin.__init__(self, handler, poll_interval)
def server_bind(self):
super(TestUDPServer, self).server_bind()
self.port = self.socket.getsockname()[1]
# - end of server_helper section # - end of server_helper section
@unittest.skipUnless(threading, 'Threading required for this test.')
class SMTPHandlerTest(BaseTest): class SMTPHandlerTest(BaseTest):
def test_basic(self): def test_basic(self):
sockmap = {} sockmap = {}