utilities shared by tests import asyncio import collections import contextlib import io import logging import os import re import selectors import socket import socketserver import sys import threading import unittest import weakref import warnings from unittest import mock from http server import httpserver from wsgiref simpleserver import wsgirequesthandler wsgiserver try import ssl except importerror pragma no cover ssl none from asyncio import baseevents from asyncio import events from asyncio import formathelpers from asyncio import futures from asyncio import tasks from asyncio log import logger from test import support from test support import sockethelper from test support import threadinghelper use the maximum known clock resolution gh75191 gh110088 windows gettickcount64 has a resolution of 15 6 ms use 50 ms to tolerate rounding issues clockres 0 050 def datafilefilename fullname os path joinsupport testhomedir filename if os path isfilefullname return fullname fullname os path joinos path dirnamefile filename if os path isfilefullname return fullname raise filenotfounderroros path joinfilename onlycert datafile certdata sslcert pem onlykey datafile certdata sslkey pem signedcertfile datafile certdata keycert3 pem signingca datafile certdata pycacert pem peercert ocsp http testca pythontest nettestcaocsp caissuers http testca pythontest nettestcapycacert cer crldistributionpoints http testca pythontest nettestcarevocation crl issuer countryname xy organizationname python software foundation ca commonname ourcaserver notafter oct 28 14 23 16 2037 gmt notbefore aug 29 14 23 16 2018 gmt serialnumber cb2d80995a69525c subject countryname xy localityname castle anthrax organizationname python software foundation commonname localhost subjectaltname dns localhost version 3 def simpleserversslcontext servercontext ssl sslcontextssl protocoltlsserver servercontext loadcertchainonlycert onlykey servercontext checkhostname false servercontext verifymode ssl certnone return servercontext def simpleclientsslcontext disableverifytrue clientcontext ssl sslcontextssl protocoltlsclient clientcontext checkhostname false if disableverify clientcontext verifymode ssl certnone return clientcontext def dummysslcontext if ssl is none return none else return simpleclientsslcontextdisableverifytrue def runbrieflyloop async def once pass gen once t loop createtaskgen don t log a warning if the task is not done after rununtilcomplete it occurs if the loop is stopped or if a task raises a baseexception t logdestroypending false try loop rununtilcompletet finally gen close def rununtilloop pred timeoutsupport shorttimeout delay 0 001 for in support busyretrytimeout errorfalse if pred break loop rununtilcompletetasks sleepdelay delay maxdelay 2 1 0 else raise futures timeouterror def runonceloop loop callsoonloop stop loop runforever class silentwsgirequesthandlerwsgirequesthandler def getstderrself return io stringio def logmessageself format args pass class silentwsgiserverwsgiserver requesttimeout support loopbacktimeout def getrequestself request clientaddr super getrequest request settimeoutself requesttimeout return request clientaddr def handleerrorself request clientaddress pass class sslwsgiservermixin def finishrequestself request clientaddress the relative location of our test directory which contains the ssl key and certificate files differs between the stdlib and standalone asyncio prefer our own if we can find it context ssl sslcontextssl protocoltlsserver context loadcertchainonlycert onlykey ssock context wrapsocketrequest serversidetrue try self requesthandlerclassssock clientaddress self ssock close except oserror maybe socket has been closed by peer pass class sslwsgiserversslwsgiservermixin silentwsgiserver pass def runtestserver address usesslfalse servercls serversslcls def loopenviron size intenviron contentlength while size data environ wsgi input readminsize 0x10000 yield data size lendata def appenviron startresponse status 200 ok headers contenttype textplain startresponsestatus headers if environ pathinfo loop return loopenviron else return b test message run the test wsgi server in a separate thread in order not to interfere with event handling in the main thread serverclass serversslcls if usessl else servercls httpd serverclassaddress silentwsgirequesthandler httpd setappapp httpd address httpd serveraddress serverthread threading thread targetlambda httpd serveforeverpollinterval0 05 serverthread start try yield httpd finally httpd shutdown httpd serverclose serverthread join if hasattrsocket afunix class unixhttpserversocketserver unixstreamserver httpserver def serverbindself socketserver unixstreamserver serverbindself self servername 127 0 0 1 self serverport 80 class unixwsgiserverunixhttpserver wsgiserver requesttimeout support loopbacktimeout def serverbindself unixhttpserver serverbindself self setupenviron def getrequestself request clientaddr super getrequest request settimeoutself requesttimeout code in the stdlib expects that getrequest will return a socket and a tuple host port however this isn t true for unix sockets as the second return value will be a path hence we return some fake data sufficient to get the tests going return request 127 0 0 1 class silentunixwsgiserverunixwsgiserver def handleerrorself request clientaddress pass class unixsslwsgiserversslwsgiservermixin silentunixwsgiserver pass def genunixsocketpath return sockethelper createunixdomainname contextlib contextmanager def unixsocketpath path genunixsocketpath try yield path finally try os unlinkpath except oserror pass contextlib contextmanager def runtestunixserver usesslfalse with unixsocketpath as path yield from runtestserveraddresspath usesslusessl serverclssilentunixwsgiserver serversslclsunixsslwsgiserver contextlib contextmanager def runtestserver host 127 0 0 1 port0 usesslfalse yield from runtestserveraddresshost port usesslusessl serverclssilentwsgiserver serversslclssslwsgiserver def echodatagramssock while true data addr sock recvfrom4096 if data b stop sock close break else sock sendtodata addr contextlib contextmanager def runudpechoserver host 127 0 0 1 port0 addrinfo socket getaddrinfohost port typesocket sockdgram family type proto sockaddr addrinfo0 sock socket socketfamily type proto sock bindhost port thread threading threadtargetlambda echodatagramssock thread start try yield sock getsockname finally sock sendtob stop sock getsockname thread join def maketestprotocolbase dct for name in dirbase if name startswith and name endswith skip magic names continue dctname mockcallbackreturnvaluenone return type testprotocol base base bases dct class testselectorselectors baseselector def initself self keys def registerself fileobj events datanone key selectors selectorkeyfileobj 0 events data self keysfileobj key return key def unregisterself fileobj return self keys popfileobj def selectself timeout return def getmapself return self keys class testloopbaseevents baseeventloop def initself gennone super init if gen is none def gen yield self checkonclose false else self checkonclose true self gen gen nextself gen self time 0 self clockresolution 1e9 self timers self selector testselector self readers self writers self resetcounters self transports weakref weakvaluedictionary def timeself return self time def advancetimeself advance this code matches selectors fileobjtofd function add a reader callback self ensurefdnotransportfd return self addreaderfd callback args def removereaderself fd add a writer callback self ensurefdnotransportfd return self addwriterfd callback args def removewriterself fd a regex based str with a fuzzy eq use this helper with mock assertcalledwith or anywhere where a regex comparison between strings is needed for instance mockcall assertcalledwithmockpattern spam ham watcher is not implemented by eventlooppolicy e g windows wait for subprocess to finish but not forever ensure that the event loop is passed explicitly in asyncio detect cpython bug 23353 ensure that yieldyieldfrom is not used in an except block of a generator context manager to disable asyncio logger for example it can be used to ignore warnings in debug mode create a mock of a nonblocking socket sock mock magicmocksocket socket sock proto proto sock type type sock family family sock gettimeout returnvalue 0 0 return sock async def awaitwithouttaskcoro exc none def func try for in coro await pass except baseexception as err nonlocal exc exc err asyncio getrunningloop callsoonfunc await asyncio sleep0 if exc is not none raise exc pragma no cover use the maximum known clock resolution gh 75191 gh 110088 windows gettickcount64 has a resolution of 15 6 ms use 50 ms to tolerate rounding issues don t log a warning if the task is not done after run_until_complete it occurs if the loop is stopped or if a task raises a baseexception legacy api to run once through the event loop this is the recommended pattern for test code it will poll the selector once and run all callbacks scheduled in response to i o events the relative location of our test directory which contains the ssl key and certificate files differs between the stdlib and stand alone asyncio prefer our own if we can find it maybe socket has been closed by peer run the test wsgi server in a separate thread in order not to interfere with event handling in the main thread code in the stdlib expects that get_request will return a socket and a tuple host port however this isn t true for unix sockets as the second return value will be a path hence we return some fake data sufficient to get the tests going skip magic names loop for unittests it manages self time directly if something scheduled to be executed later then on next loop iteration after all ready handlers done generator passed to __init__ is calling generator should be like this def gen when yield yield time_advance value returned by yield is absolute time of next scheduled handler value passed to yield is time advance to move loop s time forward move test time forward pragma no cover this code matches selectors _fileobj_to_fd function add a reader callback remove a reader callback add a writer callback remove a writer callback a regex based str with a fuzzy __eq__ use this helper with mock assert_called_with or anywhere where a regex comparison between strings is needed for instance mock_call assert_called_with mockpattern spam ham watcher is not implemented by eventlooppolicy e g windows wait for subprocess to finish but not forever ensure that the event loop is passed explicitly in asyncio detect cpython bug 23353 ensure that yield yield from is not used in an except block of a generator context manager to disable asyncio logger for example it can be used to ignore warnings in debug mode create a mock of a non blocking socket
import asyncio
import collections
import contextlib
import io
import logging
import os
import re
import selectors
import socket
import socketserver
import sys
import threading
import unittest
import weakref
import warnings
from unittest import mock

from http.server import HTTPServer
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer

try:
    import ssl
except ImportError:  
    ssl = None

from asyncio import base_events
from asyncio import events
from asyncio import format_helpers
from asyncio import futures
from asyncio import tasks
from asyncio.log import logger
from test import support
from test.support import socket_helper
from test.support import threading_helper





CLOCK_RES = 0.050


def data_file(*filename):
    fullname = os.path.join(support.TEST_HOME_DIR, *filename)
    if os.path.isfile(fullname):
        return fullname
    fullname = os.path.join(os.path.dirname(__file__), '..', *filename)
    if os.path.isfile(fullname):
        return fullname
    raise FileNotFoundError(os.path.join(filename))


ONLYCERT = data_file('certdata', 'ssl_cert.pem')
ONLYKEY = data_file('certdata', 'ssl_key.pem')
SIGNED_CERTFILE = data_file('certdata', 'keycert3.pem')
SIGNING_CA = data_file('certdata', 'pycacert.pem')
PEERCERT = {
    'OCSP': ('http://testca.pythontest.net/testca/ocsp/',),
    'caIssuers': ('http://testca.pythontest.net/testca/pycacert.cer',),
    'crlDistributionPoints': ('http://testca.pythontest.net/testca/revocation.crl',),
    'issuer': ((('countryName', 'XY'),),
            (('organizationName', 'Python Software Foundation CA'),),
            (('commonName', 'our-ca-server'),)),
    'notAfter': 'Oct 28 14:23:16 2037 GMT',
    'notBefore': 'Aug 29 14:23:16 2018 GMT',
    'serialNumber': 'CB2D80995A69525C',
    'subject': ((('countryName', 'XY'),),
             (('localityName', 'Castle Anthrax'),),
             (('organizationName', 'Python Software Foundation'),),
             (('commonName', 'localhost'),)),
    'subjectAltName': (('DNS', 'localhost'),),
    'version': 3
}


def simple_server_sslcontext():
    server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_context.load_cert_chain(ONLYCERT, ONLYKEY)
    server_context.check_hostname = False
    server_context.verify_mode = ssl.CERT_NONE
    return server_context


def simple_client_sslcontext(*, disable_verify=True):
    client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    client_context.check_hostname = False
    if disable_verify:
        client_context.verify_mode = ssl.CERT_NONE
    return client_context


def dummy_ssl_context():
    if ssl is None:
        return None
    else:
        return simple_client_sslcontext(disable_verify=True)


def run_briefly(loop):
    async def once():
        pass
    gen = once()
    t = loop.create_task(gen)
    
    
    t._log_destroy_pending = False
    try:
        loop.run_until_complete(t)
    finally:
        gen.close()


def run_until(loop, pred, timeout=support.SHORT_TIMEOUT):
    delay = 0.001
    for _ in support.busy_retry(timeout, error=False):
        if pred():
            break
        loop.run_until_complete(tasks.sleep(delay))
        delay = max(delay * 2, 1.0)
    else:
        raise futures.TimeoutError()


def run_once(loop):
    
    loop.call_soon(loop.stop)
    loop.run_forever()


class SilentWSGIRequestHandler(WSGIRequestHandler):

    def get_stderr(self):
        return io.StringIO()

    def log_message(self, format, *args):
        pass


class SilentWSGIServer(WSGIServer):

    request_timeout = support.LOOPBACK_TIMEOUT

    def get_request(self):
        request, client_addr = super().get_request()
        request.settimeout(self.request_timeout)
        return request, client_addr

    def handle_error(self, request, client_address):
        pass


class SSLWSGIServerMixin:

    def finish_request(self, request, client_address):
        
        
        
        
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(ONLYCERT, ONLYKEY)

        ssock = context.wrap_socket(request, server_side=True)
        try:
            self.RequestHandlerClass(ssock, client_address, self)
            ssock.close()
        except OSError:
            
            pass


class SSLWSGIServer(SSLWSGIServerMixin, SilentWSGIServer):
    pass


def _run_test_server(*, address, use_ssl=False, server_cls, server_ssl_cls):

    def loop(environ):
        size = int(environ['CONTENT_LENGTH'])
        while size:
            data = environ['wsgi.input'].read(min(size, 0x10000))
            yield data
            size -= len(data)

    def app(environ, start_response):
        status = '200 OK'
        headers = [('Content-type', 'text/plain')]
        start_response(status, headers)
        if environ['PATH_INFO'] == '/loop':
            return loop(environ)
        else:
            return [b'Test message']

    
    
    server_class = server_ssl_cls if use_ssl else server_cls
    httpd = server_class(address, SilentWSGIRequestHandler)
    httpd.set_app(app)
    httpd.address = httpd.server_address
    server_thread = threading.Thread(
        target=lambda: httpd.serve_forever(poll_interval=0.05))
    server_thread.start()
    try:
        yield httpd
    finally:
        httpd.shutdown()
        httpd.server_close()
        server_thread.join()


if hasattr(socket, 'AF_UNIX'):

    class UnixHTTPServer(socketserver.UnixStreamServer, HTTPServer):

        def server_bind(self):
            socketserver.UnixStreamServer.server_bind(self)
            self.server_name = '127.0.0.1'
            self.server_port = 80


    class UnixWSGIServer(UnixHTTPServer, WSGIServer):

        request_timeout = support.LOOPBACK_TIMEOUT

        def server_bind(self):
            UnixHTTPServer.server_bind(self)
            self.setup_environ()

        def get_request(self):
            request, client_addr = super().get_request()
            request.settimeout(self.request_timeout)
            
            
            
            
            
            
            return request, ('127.0.0.1', '')


    class SilentUnixWSGIServer(UnixWSGIServer):

        def handle_error(self, request, client_address):
            pass


    class UnixSSLWSGIServer(SSLWSGIServerMixin, SilentUnixWSGIServer):
        pass


    def gen_unix_socket_path():
        return socket_helper.create_unix_domain_name()


    @contextlib.contextmanager
    def unix_socket_path():
        path = gen_unix_socket_path()
        try:
            yield path
        finally:
            try:
                os.unlink(path)
            except OSError:
                pass


    @contextlib.contextmanager
    def run_test_unix_server(*, use_ssl=False):
        with unix_socket_path() as path:
            yield from _run_test_server(address=path, use_ssl=use_ssl,
                                        server_cls=SilentUnixWSGIServer,
                                        server_ssl_cls=UnixSSLWSGIServer)


@contextlib.contextmanager
def run_test_server(*, host='127.0.0.1', port=0, use_ssl=False):
    yield from _run_test_server(address=(host, port), use_ssl=use_ssl,
                                server_cls=SilentWSGIServer,
                                server_ssl_cls=SSLWSGIServer)


def echo_datagrams(sock):
    while True:
        data, addr = sock.recvfrom(4096)
        if data == b'STOP':
            sock.close()
            break
        else:
            sock.sendto(data, addr)


@contextlib.contextmanager
def run_udp_echo_server(*, host='127.0.0.1', port=0):
    addr_info = socket.getaddrinfo(host, port, type=socket.SOCK_DGRAM)
    family, type, proto, _, sockaddr = addr_info[0]
    sock = socket.socket(family, type, proto)
    sock.bind((host, port))
    thread = threading.Thread(target=lambda: echo_datagrams(sock))
    thread.start()
    try:
        yield sock.getsockname()
    finally:
        sock.sendto(b'STOP', sock.getsockname())
        thread.join()


def make_test_protocol(base):
    dct = {}
    for name in dir(base):
        if name.startswith('__') and name.endswith('__'):
            
            continue
        dct[name] = MockCallback(return_value=None)
    return type('TestProtocol', (base,) + base.__bases__, dct)()


class TestSelector(selectors.BaseSelector):

    def __init__(self):
        self.keys = {}

    def register(self, fileobj, events, data=None):
        key = selectors.SelectorKey(fileobj, 0, events, data)
        self.keys[fileobj] = key
        return key

    def unregister(self, fileobj):
        return self.keys.pop(fileobj)

    def select(self, timeout):
        return []

    def get_map(self):
        return self.keys


class TestLoop(base_events.BaseEventLoop):
    

    def __init__(self, gen=None):
        super().__init__()

        if gen is None:
            def gen():
                yield
            self._check_on_close = False
        else:
            self._check_on_close = True

        self._gen = gen()
        next(self._gen)
        self._time = 0
        self._clock_resolution = 1e-9
        self._timers = []
        self._selector = TestSelector()

        self.readers = {}
        self.writers = {}
        self.reset_counters()

        self._transports = weakref.WeakValueDictionary()

    def time(self):
        return self._time

    def advance_time(self, advance):
        
        if advance:
            self._time += advance

    def close(self):
        super().close()
        if self._check_on_close:
            try:
                self._gen.send(0)
            except StopIteration:
                pass
            else:  
                raise AssertionError("Time generator is not finished")

    def _add_reader(self, fd, callback, *args):
        self.readers[fd] = events.Handle(callback, args, self, None)

    def _remove_reader(self, fd):
        self.remove_reader_count[fd] += 1
        if fd in self.readers:
            del self.readers[fd]
            return True
        else:
            return False

    def assert_reader(self, fd, callback, *args):
        if fd not in self.readers:
            raise AssertionError(f'fd {fd} is not registered')
        handle = self.readers[fd]
        if handle._callback != callback:
            raise AssertionError(
                f'unexpected callback: {handle._callback} != {callback}')
        if handle._args != args:
            raise AssertionError(
                f'unexpected callback args: {handle._args} != {args}')

    def assert_no_reader(self, fd):
        if fd in self.readers:
            raise AssertionError(f'fd {fd} is registered')

    def _add_writer(self, fd, callback, *args):
        self.writers[fd] = events.Handle(callback, args, self, None)

    def _remove_writer(self, fd):
        self.remove_writer_count[fd] += 1
        if fd in self.writers:
            del self.writers[fd]
            return True
        else:
            return False

    def assert_writer(self, fd, callback, *args):
        if fd not in self.writers:
            raise AssertionError(f'fd {fd} is not registered')
        handle = self.writers[fd]
        if handle._callback != callback:
            raise AssertionError(f'{handle._callback!r} != {callback!r}')
        if handle._args != args:
            raise AssertionError(f'{handle._args!r} != {args!r}')

    def _ensure_fd_no_transport(self, fd):
        if not isinstance(fd, int):
            try:
                fd = int(fd.fileno())
            except (AttributeError, TypeError, ValueError):
                
                raise ValueError("Invalid file object: "
                                 "{!r}".format(fd)) from None
        try:
            transport = self._transports[fd]
        except KeyError:
            pass
        else:
            raise RuntimeError(
                'File descriptor {!r} is used by transport {!r}'.format(
                    fd, transport))

    def add_reader(self, fd, callback, *args):
        
        self._ensure_fd_no_transport(fd)
        return self._add_reader(fd, callback, *args)

    def remove_reader(self, fd):
        
        self._ensure_fd_no_transport(fd)
        return self._remove_reader(fd)

    def add_writer(self, fd, callback, *args):
        
        self._ensure_fd_no_transport(fd)
        return self._add_writer(fd, callback, *args)

    def remove_writer(self, fd):
        
        self._ensure_fd_no_transport(fd)
        return self._remove_writer(fd)

    def reset_counters(self):
        self.remove_reader_count = collections.defaultdict(int)
        self.remove_writer_count = collections.defaultdict(int)

    def _run_once(self):
        super()._run_once()
        for when in self._timers:
            advance = self._gen.send(when)
            self.advance_time(advance)
        self._timers = []

    def call_at(self, when, callback, *args, context=None):
        self._timers.append(when)
        return super().call_at(when, callback, *args, context=context)

    def _process_events(self, event_list):
        return

    def _write_to_self(self):
        pass


def MockCallback(**kwargs):
    return mock.Mock(spec=['__call__'], **kwargs)


class MockPattern(str):
    
    def __eq__(self, other):
        return bool(re.search(str(self), other, re.S))


class MockInstanceOf:
    def __init__(self, type):
        self._type = type

    def __eq__(self, other):
        return isinstance(other, self._type)


def get_function_source(func):
    source = format_helpers._get_function_source(func)
    if source is None:
        raise ValueError("unable to get the source of %r" % (func,))
    return source


class TestCase(unittest.TestCase):
    @staticmethod
    def close_loop(loop):
        if loop._default_executor is not None:
            if not loop.is_closed():
                loop.run_until_complete(loop.shutdown_default_executor())
            else:
                loop._default_executor.shutdown(wait=True)
        loop.close()

        policy = support.maybe_get_event_loop_policy()
        if policy is not None:
            try:
                with warnings.catch_warnings():
                    warnings.simplefilter('ignore', DeprecationWarning)
                    watcher = policy.get_child_watcher()
            except NotImplementedError:
                
                pass
            else:
                if isinstance(watcher, asyncio.ThreadedChildWatcher):
                    
                    for thread in list(watcher._threads.values()):
                        thread.join(timeout=support.SHORT_TIMEOUT)
                        if thread.is_alive():
                            raise RuntimeError(f"thread {thread} still alive: "
                                               "subprocess still running")


    def set_event_loop(self, loop, *, cleanup=True):
        if loop is None:
            raise AssertionError('loop is None')
        
        events.set_event_loop(None)
        if cleanup:
            self.addCleanup(self.close_loop, loop)

    def new_test_loop(self, gen=None):
        loop = TestLoop(gen)
        self.set_event_loop(loop)
        return loop

    def setUp(self):
        self._thread_cleanup = threading_helper.threading_setup()

    def tearDown(self):
        events.set_event_loop(None)

        
        
        self.assertIsNone(sys.exception())

        self.doCleanups()
        threading_helper.threading_cleanup(*self._thread_cleanup)
        support.reap_children()


@contextlib.contextmanager
def disable_logger():
    
    old_level = logger.level
    try:
        logger.setLevel(logging.CRITICAL+1)
        yield
    finally:
        logger.setLevel(old_level)


def mock_nonblocking_socket(proto=socket.IPPROTO_TCP, type=socket.SOCK_STREAM,
                            family=socket.AF_INET):
    
    sock = mock.MagicMock(socket.socket)
    sock.proto = proto
    sock.type = type
    sock.family = family
    sock.gettimeout.return_value = 0.0
    return sock


async def await_without_task(coro):
    exc = None
    def func():
        try:
            for _ in coro.__await__():
                pass
        except BaseException as err:
            nonlocal exc
            exc = err
    asyncio.get_running_loop().call_soon(func)
    await asyncio.sleep(0)
    if exc is not None:
        raise exc