cpython: c83fb59b73ea (original) (raw)

--- a/Lib/test/test_logging.py +++ b/Lib/test/test_logging.py @@ -25,24 +25,17 @@ import logging import logging.handlers import logging.config -import asynchat -import asyncore import codecs import datetime -import errno import pickle import io import gc -from http.server import HTTPServer, BaseHTTPRequestHandler import json import os import queue import re import select -import smtpd import socket -from socketserver import (ThreadingUDPServer, DatagramRequestHandler,

                     ThreadingTCPServer, StreamRequestHandler)[](#l1.24)

import struct import sys import tempfile @@ -51,11 +44,19 @@ from test.support import TestHandler, Ma import textwrap import time import unittest -from urllib.parse import urlparse, parse_qs import warnings import weakref try: 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)[](#l1.45)

except ImportError: threading = None try: @@ -611,284 +612,286 @@ class StreamHandlerTest(BaseTest):

-- The following section could be moved into a server_helper.py module

-- if it proves to be of wider utility than just test_logging

-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)[](#l1.61)

```
   self.smtp_server = server[](#l1.62)
```
```
   self.conn = conn[](#l1.63)
```
```
   self.addr = addr[](#l1.64)
```
```
   self.received_lines = [][](#l1.65)
```

   self.smtp_state = self.COMMAND[](#l1.66)

```
   self.seen_greeting = ''[](#l1.67)
```
```
   self.mailfrom = None[](#l1.68)
```
```
   self.rcpttos = [][](#l1.69)
```
```
   self.received_data = ''[](#l1.70)
```

   self.fqdn = socket.getfqdn()[](#l1.71)

```
   self.num_bytes = 0[](#l1.72)
```
```
   try:[](#l1.73)
```

       self.peer = conn.getpeername()[](#l1.74)

   except socket.error as err:[](#l1.75)

       # a race condition  may occur if the other end is closing[](#l1.76)

       # before we can get the peername[](#l1.77)

```
       self.close()[](#l1.78)
```

       if err.args[0] != errno.ENOTCONN:[](#l1.79)

```
           raise[](#l1.80)
```
```
       return[](#l1.81)
```

   self.push('220 %s %s' % (self.fqdn, smtpd.__version__))[](#l1.82)

   self.set_terminator(b'\r\n')[](#l1.83)

- - -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[](#l1.91)

            *port* attribute will hold the actual port number[](#l1.92)

            used, which can be used in client connections.[](#l1.93)

:param handler: A callable which will be called to process

               incoming messages. The handler will be passed[](#l1.95)

               the client address tuple, who the message is from,[](#l1.96)

               a list of recipients and the message data.[](#l1.97)

:param poll_interval: The interval, in seconds, used in the underlying

                     :func:`select` or :func:`poll` call by[](#l1.99)

                     :func:`asyncore.loop`.[](#l1.100)

:param sockmap: A dictionary which will be used to hold

               :class:`asyncore.dispatcher` instances used by[](#l1.102)

               :func:`asyncore.loop`. This avoids changing the[](#l1.103)

               :mod:`asyncore` module's global state.[](#l1.104)

"""
channel_class = TestSMTPChannel

def init(self, addr, handler, poll_interval, sockmap):
```
   self._localaddr = addr[](#l1.109)
```
```
   self._remoteaddr = None[](#l1.110)
```
```
   self.sockmap = sockmap[](#l1.111)
```

   asyncore.dispatcher.__init__(self, map=sockmap)[](#l1.112)

```
   try:[](#l1.113)
```

       sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)[](#l1.114)

```
       sock.setblocking(0)[](#l1.115)
```

       self.set_socket(sock, map=sockmap)[](#l1.116)

       # try to re-use a server port if possible[](#l1.117)

```
       self.set_reuse_addr()[](#l1.118)
```
```
       self.bind(addr)[](#l1.119)
```

       self.port = sock.getsockname()[1][](#l1.120)

```
       self.listen(5)[](#l1.121)
```
```
   except:[](#l1.122)
```
```
       self.close()[](#l1.123)
```
```
       raise[](#l1.124)
```
```
   self._handler = handler[](#l1.125)
```
```
   self._thread = None[](#l1.126)
```

   self.poll_interval = poll_interval[](#l1.127)

def handle_accepted(self, conn, addr):

+if threading:

class TestSMTPChannel(smtpd.SMTPChannel):
```
   """[](#l1.132)
```

   This derived class has had to be created because smtpd does not[](#l1.133)

   support use of custom channel maps, although they are allowed by[](#l1.134)

   asyncore's design. Issue #11959 has been raised to address this,[](#l1.135)

   and if resolved satisfactorily, some of this code can be removed.[](#l1.136)
   """[](#l1.137)

   Redefined only because the base class does not pass in a[](#l1.138)

   map, forcing use of a global in :mod:`asyncore`.[](#l1.139)

```
   """[](#l1.140)
```

   channel = self.channel_class(self, conn, addr, self.sockmap)[](#l1.141)

def process_message(self, peer, mailfrom, rcpttos, data):
```
   """[](#l1.144)
```

   Delegates to the handler passed in to the server's constructor.[](#l1.145)

   Typically, this will be a test case method.[](#l1.147)

   :param peer: The client (host, port) tuple.[](#l1.148)

   :param mailfrom: The address of the sender.[](#l1.149)

   :param rcpttos: The addresses of the recipients.[](#l1.150)

```
   :param data: The message.[](#l1.151)
```

   def __init__(self, server, conn, addr, sockmap):[](#l1.152)

       asynchat.async_chat.__init__(self, conn, sockmap)[](#l1.153)

       self.smtp_server = server[](#l1.154)

```
       self.conn = conn[](#l1.155)
```
```
       self.addr = addr[](#l1.156)
```

       self.received_lines = [][](#l1.157)

       self.smtp_state = self.COMMAND[](#l1.158)

       self.seen_greeting = ''[](#l1.159)

```
       self.mailfrom = None[](#l1.160)
```
```
       self.rcpttos = [][](#l1.161)
```

       self.received_data = ''[](#l1.162)

       self.fqdn = socket.getfqdn()[](#l1.163)

```
       self.num_bytes = 0[](#l1.164)
```
```
       try:[](#l1.165)
```

           self.peer = conn.getpeername()[](#l1.166)

       except socket.error as err:[](#l1.167)

           # a race condition  may occur if the other end is closing[](#l1.168)

           # before we can get the peername[](#l1.169)

```
           self.close()[](#l1.170)
```

           if err.args[0] != errno.ENOTCONN:[](#l1.171)

```
               raise[](#l1.172)
```
```
           return[](#l1.173)
```

       self.push('220 %s %s' % (self.fqdn, smtpd.__version__))[](#l1.174)

       self.set_terminator(b'\r\n')[](#l1.175)

+ +

class TestSMTPServer(smtpd.SMTPServer): """

   self._handler(peer, mailfrom, rcpttos, data)[](#l1.180)

def start(self):
```
   """[](#l1.183)
```

   Start the server running on a separate daemon thread.[](#l1.184)

```
   """[](#l1.185)
```

   self._thread = t = threading.Thread(target=self.serve_forever,[](#l1.186)

                                       args=(self.poll_interval,))[](#l1.187)

```
   t.setDaemon(True)[](#l1.188)
```
```
   t.start()[](#l1.189)
```

def serve_forever(self, poll_interval):
```
   """[](#l1.192)
```

   Run the :mod:`asyncore` loop until normal termination[](#l1.193)

```
   conditions arise.[](#l1.194)
```

   This class implements a test SMTP server.[](#l1.195)

   :param addr: A (host, port) tuple which the server listens on.[](#l1.197)

                You can specify a port value of zero: the server's[](#l1.198)

                *port* attribute will hold the actual port number[](#l1.199)

                used, which can be used in client connections.[](#l1.200)

   :param handler: A callable which will be called to process[](#l1.201)

                   incoming messages. The handler will be passed[](#l1.202)

                   the client address tuple, who the message is from,[](#l1.203)

                   a list of recipients and the message data.[](#l1.204)
   :param poll_interval: The interval, in seconds, used in the underlying[](#l1.205)
                         :func:`select` or :func:`poll` call by[](#l1.206)
                         :func:`asyncore.loop`.[](#l1.207)

```
   """[](#l1.208)
```
```
   try:[](#l1.209)
```

       asyncore.loop(poll_interval, map=self.sockmap)[](#l1.210)

```
   except select.error:[](#l1.211)
```

       # On FreeBSD 8, closing the server repeatably[](#l1.212)

       # raises this error. We swallow it if the[](#l1.213)

       # server has been closed.[](#l1.214)

       if self.connected or self.accepting:[](#l1.215)

```
           raise[](#l1.216)
```

def stop(self, timeout=None):
```
   """[](#l1.219)
```

   Stop the thread by closing the server instance.[](#l1.220)

   Wait for the server thread to terminate.[](#l1.221)

   :param timeout: How long to wait for the server thread[](#l1.223)

                   to terminate.[](#l1.224)

   :param sockmap: A dictionary which will be used to hold[](#l1.225)

                   :class:`asyncore.dispatcher` instances used by[](#l1.226)

                   :func:`asyncore.loop`. This avoids changing the[](#l1.227)

                   :mod:`asyncore` module's global state.[](#l1.228)
   """[](#l1.229)

```
   self.close()[](#l1.230)
```

   self._thread.join(timeout)[](#l1.231)

```
   self._thread = None[](#l1.232)
```

- -class ControlMixin(object):

"""
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 handler: A handler callable which will be called with a

               single parameter - the request - in order to[](#l1.243)

               process the request. This handler is called on the[](#l1.244)

               server thread, effectively meaning that requests are[](#l1.245)

               processed serially. While not quite Web scale ;-),[](#l1.246)

               this should be fine for testing applications.[](#l1.247)

:param poll_interval: The polling interval in seconds.
"""
def init(self, handler, poll_interval):
```
   self._thread = None[](#l1.251)
```

   self.poll_interval = poll_interval[](#l1.252)

```
   self._handler = handler[](#l1.253)
```

   self.ready = threading.Event()[](#l1.254)

def start(self):
```
   """[](#l1.257)
```

   Create a daemon thread to run the server, and start it.[](#l1.258)

```
   """[](#l1.259)
```

   self._thread = t = threading.Thread(target=self.serve_forever,[](#l1.260)

                                       args=(self.poll_interval,))[](#l1.261)

```
   t.setDaemon(True)[](#l1.262)
```
```
   t.start()[](#l1.263)
```

def serve_forever(self, poll_interval):
```
   """[](#l1.266)
```

   Run the server. Set the ready flag before entering the[](#l1.267)

```
   service loop.[](#l1.268)
```
```
   """[](#l1.269)
```
```
   self.ready.set()[](#l1.270)
```

   super(ControlMixin, self).serve_forever(poll_interval)[](#l1.271)

def stop(self, timeout=None):
```
   """[](#l1.274)
```

   Tell the server thread to stop, and wait for it to do so.[](#l1.275)

   :param timeout: How long to wait for the server thread[](#l1.277)

                   to terminate.[](#l1.278)

```
   """[](#l1.279)
```
```
   self.shutdown()[](#l1.280)
```

   if self._thread is not None:[](#l1.281)

   channel_class = TestSMTPChannel[](#l1.282)

   def __init__(self, addr, handler, poll_interval, sockmap):[](#l1.284)

       self._localaddr = addr[](#l1.285)

       self._remoteaddr = None[](#l1.286)

       self.sockmap = sockmap[](#l1.287)

       asyncore.dispatcher.__init__(self, map=sockmap)[](#l1.288)

```
       try:[](#l1.289)
```

           sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)[](#l1.290)

           sock.setblocking(0)[](#l1.291)

           self.set_socket(sock, map=sockmap)[](#l1.292)

           # try to re-use a server port if possible[](#l1.293)

           self.set_reuse_addr()[](#l1.294)

```
           self.bind(addr)[](#l1.295)
```

           self.port = sock.getsockname()[1][](#l1.296)

```
           self.listen(5)[](#l1.297)
```
```
       except:[](#l1.298)
```
```
           self.close()[](#l1.299)
```
```
           raise[](#l1.300)
```

       self._handler = handler[](#l1.301)

```
       self._thread = None[](#l1.302)
```

       self.poll_interval = poll_interval[](#l1.303)

   def handle_accepted(self, conn, addr):[](#l1.305)

```
       """[](#l1.306)
```

       Redefined only because the base class does not pass in a[](#l1.307)

       map, forcing use of a global in :mod:`asyncore`.[](#l1.308)

```
       """[](#l1.309)
```

       channel = self.channel_class(self, conn, addr, self.sockmap)[](#l1.310)

   def process_message(self, peer, mailfrom, rcpttos, data):[](#l1.312)

```
       """[](#l1.313)
```

       Delegates to the handler passed in to the server's constructor.[](#l1.314)

       Typically, this will be a test case method.[](#l1.316)

       :param peer: The client (host, port) tuple.[](#l1.317)

       :param mailfrom: The address of the sender.[](#l1.318)

       :param rcpttos: The addresses of the recipients.[](#l1.319)

       :param data: The message.[](#l1.320)

```
       """[](#l1.321)
```

       self._handler(peer, mailfrom, rcpttos, data)[](#l1.322)

```
   def start(self):[](#l1.324)
```
```
       """[](#l1.325)
```

       Start the server running on a separate daemon thread.[](#l1.326)

```
       """[](#l1.327)
```

       self._thread = t = threading.Thread(target=self.serve_forever,[](#l1.328)

                                           args=(self.poll_interval,))[](#l1.329)

```
       t.setDaemon(True)[](#l1.330)
```
```
       t.start()[](#l1.331)
```

   def serve_forever(self, poll_interval):[](#l1.333)

```
       """[](#l1.334)
```

       Run the :mod:`asyncore` loop until normal termination[](#l1.335)

```
       conditions arise.[](#l1.336)
```

       :param poll_interval: The interval, in seconds, used in the underlying[](#l1.337)

                             :func:`select` or :func:`poll` call by[](#l1.338)

                             :func:`asyncore.loop`.[](#l1.339)

```
       """[](#l1.340)
```
```
       try:[](#l1.341)
```

           asyncore.loop(poll_interval, map=self.sockmap)[](#l1.342)

```
       except select.error:[](#l1.343)
```

           # On FreeBSD 8, closing the server repeatably[](#l1.344)

           # raises this error. We swallow it if the[](#l1.345)

           # server has been closed.[](#l1.346)

           if self.connected or self.accepting:[](#l1.347)

```
               raise[](#l1.348)
```

   def stop(self, timeout=None):[](#l1.350)

```
       """[](#l1.351)
```

       Stop the thread by closing the server instance.[](#l1.352)

       Wait for the server thread to terminate.[](#l1.353)

       :param timeout: How long to wait for the server thread[](#l1.355)

                       to terminate.[](#l1.356)

```
       """[](#l1.357)
```

       self.close()[](#l1.358)
       self._thread.join(timeout)[](#l1.359)
       self._thread = None[](#l1.360)

```
   self.server_close()[](#l1.361)
```
```
   self.ready.clear()[](#l1.362)
```

- -class TestHTTPServer(ControlMixin, HTTPServer):

"""
An HTTP 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[](#l1.370)

               process the request.[](#l1.371)

: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):[](#l1.376)

       def __getattr__(self, name, default=None):[](#l1.377)

           if name.startswith('do_'):[](#l1.378)

               return self.process_request[](#l1.379)

           raise AttributeError(name)[](#l1.380)

       def process_request(self):[](#l1.382)

           self.server._handler(self)[](#l1.383)

       def log_message(self, format, *args):[](#l1.385)

```
           if log:[](#l1.386)
```

               super(DelegatingHTTPRequestHandler,[](#l1.387)

                     self).log_message(format, *args)[](#l1.388)

   HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)[](#l1.389)

   ControlMixin.__init__(self, handler, poll_interval)[](#l1.390)

- -class TestTCPServer(ControlMixin, ThreadingTCPServer):

"""
A TCP 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.[](#l1.398)

:param poll_interval: The polling interval in seconds.
:bind_and_activate: If True (the default), binds the server and starts it

                   listening. If False, you need to call[](#l1.401)

                   :meth:`server_bind` and :meth:`server_activate` at[](#l1.402)

                   some later time before calling :meth:`start`, so that[](#l1.403)

                   the server will set up the socket and listen on it.[](#l1.404)

allow_reuse_address = True

def init(self, addr, handler, poll_interval=0.5,

            bind_and_activate=True):[](#l1.410)

   class DelegatingTCPRequestHandler(StreamRequestHandler):[](#l1.411)

```
       def handle(self):[](#l1.413)
```

           self.server._handler(self)[](#l1.414)

   ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,[](#l1.415)

                               bind_and_activate)[](#l1.416)

   ControlMixin.__init__(self, handler, poll_interval)[](#l1.417)

def server_bind(self):

   super(TestTCPServer, self).server_bind()[](#l1.420)

   self.port = self.socket.getsockname()[1][](#l1.421)

- -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[](#l1.429)

               process the request.[](#l1.430)

:param poll_interval: The polling interval for shutdown requests,

                     in seconds.[](#l1.432)

:bind_and_activate: If True (the default), binds the server and

                   starts it listening. If False, you need to[](#l1.434)

                   call :meth:`server_bind` and[](#l1.435)

                   :meth:`server_activate` at some later time[](#l1.436)

                   before calling :meth:`start`, so that the server will[](#l1.437)

                   set up the socket and listen on it.[](#l1.438)

"""
def init(self, addr, handler, poll_interval=0.5, bind_and_activate=True):

   class DelegatingUDPRequestHandler(DatagramRequestHandler):[](#l1.441)

```
       def handle(self):[](#l1.443)
```

           self.server._handler(self)[](#l1.444)

   ThreadingUDPServer.__init__(self, addr, DelegatingUDPRequestHandler,[](#l1.445)

                               bind_and_activate)[](#l1.446)

   ControlMixin.__init__(self, handler, poll_interval)[](#l1.447)

def server_bind(self):

   super(TestUDPServer, self).server_bind()[](#l1.450)

   self.port = self.socket.getsockname()[1][](#l1.451)

class ControlMixin(object):
```
   """[](#l1.454)
```

   This mixin is used to start a server on a separate thread, and[](#l1.455)

   shut it down programmatically. Request handling is simplified - instead[](#l1.456)

   of needing to derive a suitable RequestHandler subclass, you just[](#l1.457)

   provide a callable which will be passed each received request to be[](#l1.458)

```
   processed.[](#l1.459)
```

   :param handler: A handler callable which will be called with a[](#l1.461)

                   single parameter - the request - in order to[](#l1.462)

                   process the request. This handler is called on the[](#l1.463)

                   server thread, effectively meaning that requests are[](#l1.464)

                   processed serially. While not quite Web scale ;-),[](#l1.465)

                   this should be fine for testing applications.[](#l1.466)

   :param poll_interval: The polling interval in seconds.[](#l1.467)

```
   """[](#l1.468)
```

   def __init__(self, handler, poll_interval):[](#l1.469)

```
       self._thread = None[](#l1.470)
```

       self.poll_interval = poll_interval[](#l1.471)

       self._handler = handler[](#l1.472)

       self.ready = threading.Event()[](#l1.473)

```
   def start(self):[](#l1.475)
```
```
       """[](#l1.476)
```

       Create a daemon thread to run the server, and start it.[](#l1.477)

```
       """[](#l1.478)
```

       self._thread = t = threading.Thread(target=self.serve_forever,[](#l1.479)

                                           args=(self.poll_interval,))[](#l1.480)

```
       t.setDaemon(True)[](#l1.481)
```
```
       t.start()[](#l1.482)
```

   def serve_forever(self, poll_interval):[](#l1.484)

```
       """[](#l1.485)
```

       Run the server. Set the ready flag before entering the[](#l1.486)

```
       service loop.[](#l1.487)
```
```
       """[](#l1.488)
```
```
       self.ready.set()[](#l1.489)
```

       super(ControlMixin, self).serve_forever(poll_interval)[](#l1.490)

   def stop(self, timeout=None):[](#l1.492)

```
       """[](#l1.493)
```

       Tell the server thread to stop, and wait for it to do so.[](#l1.494)

       :param timeout: How long to wait for the server thread[](#l1.496)

                       to terminate.[](#l1.497)

```
       """[](#l1.498)
```
```
       self.shutdown()[](#l1.499)
```

       if self._thread is not None:[](#l1.500)

           self._thread.join(timeout)[](#l1.501)

           self._thread = None[](#l1.502)

```
       self.server_close()[](#l1.503)
```
```
       self.ready.clear()[](#l1.504)
```

class TestHTTPServer(ControlMixin, HTTPServer):
```
   """[](#l1.507)
```

   An HTTP server which is controllable using :class:`ControlMixin`.[](#l1.508)

   :param addr: A tuple with the IP address and port to listen on.[](#l1.510)

   :param handler: A handler callable which will be called with a[](#l1.511)

                   single parameter - the request - in order to[](#l1.512)

                   process the request.[](#l1.513)

   :param poll_interval: The polling interval in seconds.[](#l1.514)

   :param log: Pass ``True`` to enable log messages.[](#l1.515)

```
   """[](#l1.516)
```

   def __init__(self, addr, handler, poll_interval=0.5, log=False):[](#l1.517)

       class DelegatingHTTPRequestHandler(BaseHTTPRequestHandler):[](#l1.518)

           def __getattr__(self, name, default=None):[](#l1.519)

               if name.startswith('do_'):[](#l1.520)

                   return self.process_request[](#l1.521)

               raise AttributeError(name)[](#l1.522)

           def process_request(self):[](#l1.524)

               self.server._handler(self)[](#l1.525)

           def log_message(self, format, *args):[](#l1.527)

```
               if log:[](#l1.528)
```

                   super(DelegatingHTTPRequestHandler,[](#l1.529)

                         self).log_message(format, *args)[](#l1.530)

       HTTPServer.__init__(self, addr, DelegatingHTTPRequestHandler)[](#l1.531)

       ControlMixin.__init__(self, handler, poll_interval)[](#l1.532)

class TestTCPServer(ControlMixin, ThreadingTCPServer):
```
   """[](#l1.535)
```

   A TCP server which is controllable using :class:`ControlMixin`.[](#l1.536)

   :param addr: A tuple with the IP address and port to listen on.[](#l1.538)

   :param handler: A handler callable which will be called with a single[](#l1.539)

                   parameter - the request - in order to process the request.[](#l1.540)

   :param poll_interval: The polling interval in seconds.[](#l1.541)

   :bind_and_activate: If True (the default), binds the server and starts it[](#l1.542)

                       listening. If False, you need to call[](#l1.543)

                       :meth:`server_bind` and :meth:`server_activate` at[](#l1.544)

                       some later time before calling :meth:`start`, so that[](#l1.545)

                       the server will set up the socket and listen on it.[](#l1.546)

```
   """[](#l1.547)
```

   allow_reuse_address = True[](#l1.549)

   def __init__(self, addr, handler, poll_interval=0.5,[](#l1.551)

                bind_and_activate=True):[](#l1.552)

       class DelegatingTCPRequestHandler(StreamRequestHandler):[](#l1.553)

```
           def handle(self):[](#l1.555)
```

               self.server._handler(self)[](#l1.556)

       ThreadingTCPServer.__init__(self, addr, DelegatingTCPRequestHandler,[](#l1.557)

                                   bind_and_activate)[](#l1.558)

       ControlMixin.__init__(self, handler, poll_interval)[](#l1.559)

```
   def server_bind(self):[](#l1.561)
```

       super(TestTCPServer, self).server_bind()[](#l1.562)

       self.port = self.socket.getsockname()[1][](#l1.563)

class TestUDPServer(ControlMixin, ThreadingUDPServer):
```
   """[](#l1.566)
```

   A UDP server which is controllable using :class:`ControlMixin`.[](#l1.567)

   :param addr: A tuple with the IP address and port to listen on.[](#l1.569)

   :param handler: A handler callable which will be called with a[](#l1.570)

                   single parameter - the request - in order to[](#l1.571)

                   process the request.[](#l1.572)

   :param poll_interval: The polling interval for shutdown requests,[](#l1.573)

                         in seconds.[](#l1.574)

   :bind_and_activate: If True (the default), binds the server and[](#l1.575)

                       starts it listening. If False, you need to[](#l1.576)

                       call :meth:`server_bind` and[](#l1.577)

                       :meth:`server_activate` at some later time[](#l1.578)

                       before calling :meth:`start`, so that the server will[](#l1.579)

                       set up the socket and listen on it.[](#l1.580)

```
   """[](#l1.581)
```

   def __init__(self, addr, handler, poll_interval=0.5, bind_and_activate=True):[](#l1.582)

       class DelegatingUDPRequestHandler(DatagramRequestHandler):[](#l1.583)

```
           def handle(self):[](#l1.585)
```

               self.server._handler(self)[](#l1.586)

       ThreadingUDPServer.__init__(self, addr, DelegatingUDPRequestHandler,[](#l1.587)

                                   bind_and_activate)[](#l1.588)

       ControlMixin.__init__(self, handler, poll_interval)[](#l1.589)

```
   def server_bind(self):[](#l1.591)
```

       super(TestUDPServer, self).server_bind()[](#l1.592)

       self.port = self.socket.getsockname()[1][](#l1.593)

- end of server_helper section

+@unittest.skipUnless(threading, 'Threading required for this test.') class SMTPHandlerTest(BaseTest): def test_basic(self): sockmap = {}

cpython: c83fb59b73ea (original) (raw)

The following imports are needed only for tests which

-- The following section could be moved into a server_helper.py module

-- if it proves to be of wider utility than just test_logging

- end of server_helper section