B v9a@ @sdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlmZddlmZddlmZmZy ddlZWnek rdZYnXddlmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!d dZ"e"dZ#e"dZ$e"dZ%e"dZ&ddddddddddd Z'ddZ(d d!d"d#Z)d$d%Z*d&d'Z+d`d)d*Z,d+d,Z-Gd-d.d.eZ.Gd/d0d0eZ/Gd1d2d2Z0Gd3d4d4e0e/Z1d5d6d7d8Z2e3ed9rdGd:d;d;e j4eZ5Gdd?d?e6Z7Gd@dAdAe0e7Z8dBdCZ9ej:dDdEZ;ej:d5d6dFdGZGdNdOdOej?Z@GdPdQdQejAZBdRdSZCGdTdUdUeDZEGdVdWdWZFdXdYZGGdZd[d[ejHZHej:d\d]ZIejJejKejLfd^d_ZMdS)azUtilities shared by tests.N)mock) HTTPServer)WSGIRequestHandler WSGIServer) base_events)events)format_helpers)futures)tasks)logger)supportcCs^ttdr*tjtj|}tj|r*|Stjtjtd|}tj|rR|St |dS)N TEST_HOME_DIRz..) hasattrr ospathjoinr isfiledirname__file__FileNotFoundError)filenamefullnamer/usr/lib/python3.7/utils.py data_file&s   rz ssl_cert.pemz ssl_key.pemz keycert3.pemz pycacert.pem)z)http://testca.pythontest.net/testca/ocsp/)z0http://testca.pythontest.net/testca/pycacert.cer)z2http://testca.pythontest.net/testca/revocation.crl))) countryNameXY))organizationNamezPython Software Foundation CA)) commonNamez our-ca-serverzJul 7 14:23:16 2028 GMTzAug 29 14:23:16 2018 GMTZCB2D80995A69525C)))rr))Z localityNamezCastle Anthrax))rzPython Software Foundation))r localhost))ZDNSr) ZOCSPZ caIssuersZcrlDistributionPointsZissuerZnotAfterZ notBeforeZ serialNumberZsubjectZsubjectAltNameversioncCs*ttj}|ttd|_tj|_|S)NF) ssl SSLContextZPROTOCOL_TLS_SERVERload_cert_chainONLYCERTONLYKEYcheck_hostname CERT_NONE verify_mode)Zserver_contextrrrsimple_server_sslcontextHs   r*T)disable_verifycCs"ttj}d|_|rtj|_|S)NF)r"r#ZPROTOCOL_TLS_CLIENTr'r(r))r+Zclient_contextrrrsimple_client_sslcontextPs  r,cCstdkr dSttjSdS)N)r"r#Z PROTOCOL_TLSrrrrdummy_ssl_contextXsr-cCs<dd}|}||}d|_z||Wd|XdS)NcsdS)Nrrrrronce`szrun_briefly..onceF)Z create_taskZ_log_destroy_pendingrun_until_completeclose)loopr.gentrrr run_briefly_s r4cCsTt|}xB|sN|dk r8|t}|dkr8t|tjd|dqWdS)NrgMbP?)r1)timeZ monotonicr TimeoutErrorr/r Zsleep)r1ZpredtimeoutZdeadlinerrr run_untilms  r9cCs||j|dS)zLegacy 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. N)Z call_soonstopZ run_forever)r1rrrrun_oncews r;c@seZdZddZddZdS)SilentWSGIRequestHandlercCstS)N)ioStringIO)selfrrr get_stderrsz#SilentWSGIRequestHandler.get_stderrcGsdS)Nr)r?formatargsrrr log_messagesz$SilentWSGIRequestHandler.log_messageN)__name__ __module__ __qualname__r@rCrrrrr<sr<cs(eZdZdZfddZddZZS)SilentWSGIServercs"t\}}||j||fS)N)super get_request settimeoutrequest_timeout)r?request client_addr) __class__rrrJs zSilentWSGIServer.get_requestcCsdS)Nr)r?rMclient_addressrrr handle_errorszSilentWSGIServer.handle_error)rDrErFrLrJrQ __classcell__rr)rOrrGs rGc@seZdZddZdS)SSLWSGIServerMixincCsVt}|tt|j|dd}y|||||Wntk rPYnXdS)NT)Z server_side) r"r#r$r%r&Z wrap_socketZRequestHandlerClassr0OSError)r?rMrPcontextZssockrrrfinish_requests  z!SSLWSGIServerMixin.finish_requestN)rDrErFrVrrrrrSsrSc@s eZdZdS) SSLWSGIServerN)rDrErFrrrrrWsrWF)use_sslc#svdd}|r|n|}||t|j_tjfddd}|z VWd| XdS)NcSsd}dg}|||dgS)Nz200 OK)z Content-typez text/plains Test messager)environZstart_responsestatusZheadersrrrapps z_run_test_server..appcs jddS)Ng?)Z poll_interval)Z serve_foreverr)httpdrrz"_run_test_server..)target) r<Zset_appZserver_addressaddress threadingZThreadstartshutdownZ server_closer)r`rX server_clsserver_ssl_clsr[Z server_classZ server_threadr)r\r_run_test_servers    rfZAF_UNIXc@seZdZddZdS)UnixHTTPServercCstj|d|_d|_dS)Nz 127.0.0.1P) socketserverUnixStreamServer server_bindZ server_nameZ server_port)r?rrrrks zUnixHTTPServer.server_bindN)rDrErFrkrrrrrgsrgcs(eZdZdZddZfddZZS)UnixWSGIServerrHcCst||dS)N)rgrkZ setup_environ)r?rrrrks zUnixWSGIServer.server_bindcs"t\}}||j|dfS)N)z 127.0.0.1)rIrJrKrL)r?rMrN)rOrrrJs zUnixWSGIServer.get_request)rDrErFrLrkrJrRrr)rOrrlsrlc@seZdZddZdS)SilentUnixWSGIServercCsdS)Nr)r?rMrPrrrrQsz!SilentUnixWSGIServer.handle_errorN)rDrErFrQrrrrrnsrnc@s eZdZdS)UnixSSLWSGIServerN)rDrErFrrrrrosroc Cst}|jSQRXdS)N)tempfileZNamedTemporaryFilename)filerrrgen_unix_socket_paths rsccs<t}z |VWdyt|Wntk r4YnXXdS)N)rsrunlinkrT)rrrrunix_socket_paths ruc cs,t}t||ttdEdHWdQRXdS)N)r`rXrdre)rurfrnro)rXrrrrrun_test_unix_serversrvz 127.0.0.1)hostportrXccst||f|ttdEdHdS)N)r`rXrdre)rfrGrW)rwrxrXrrrrun_test_server s rycCsPi}x4t|D](}|dr(|dr(qtdd||<qWtd|f|j|S)N__) return_valueZ TestProtocol)dir startswithendswith MockCallbacktype __bases__)baseZdctrqrrrmake_test_protocols rc@s6eZdZddZd ddZddZdd Zd d ZdS) TestSelectorcCs i|_dS)N)keys)r?rrr__init__szTestSelector.__init__NcCst|d||}||j|<|S)Nr) selectorsZ SelectorKeyr)r?fileobjrdatakeyrrrregisters zTestSelector.registercCs |j|S)N)rpop)r?rrrr unregister$szTestSelector.unregistercCsgS)Nr)r?r8rrrselect'szTestSelector.selectcCs|jS)N)r)r?rrrget_map*szTestSelector.get_map)N)rDrErFrrrrrrrrrrs  rcseZdZdZd.fdd ZddZddZfd d Zd d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!d"Zd#d$Zfd%d&Zdd'fd(d) Zd*d+Zd,d-ZZS)/TestLoopaLoop 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. Ncsvt|dkr"dd}d|_nd|_||_t|jd|_d|_g|_t|_ i|_ i|_ | t |_dS)Ncss dVdS)Nrrrrrr2EszTestLoop.__init__..genFTrg& .>)rIr_check_on_close_gennext_timeZ_clock_resolution_timersrZ _selectorreaderswritersreset_countersweakrefZWeakValueDictionary _transports)r?r2)rOrrrAs  zTestLoop.__init__cCs|jS)N)r)r?rrrr6Xsz TestLoop.timecCs|r|j|7_dS)zMove test time forward.N)r)r?advancerrr advance_time[szTestLoop.advance_timecsBt|jr>y|jdWntk r4Yn XtddS)NrzTime generator is not finished)rIr0rrsend StopIterationAssertionError)r?)rOrrr0`s zTestLoop.closecGst|||d|j|<dS)N)rHandler)r?fdcallbackrBrrr _add_readerjszTestLoop._add_readercCs0|j|d7<||jkr(|j|=dSdSdS)NTF)remove_reader_countr)r?rrrr_remove_readerms  zTestLoop._remove_readercGsh||jkrtd|d|j|}|j|krDtd|jd||j|krdtd|jd|dS)Nzfd z is not registeredzunexpected callback: z != zunexpected callback args: )rr _callback_args)r?rrrBhandlerrr assert_readerus    zTestLoop.assert_readercCs||jkrtd|ddS)Nzfd z is registered)rr)r?rrrrassert_no_readers zTestLoop.assert_no_readercGst|||d|j|<dS)N)rrr)r?rrrBrrr _add_writerszTestLoop._add_writercCs0|j|d7<||jkr(|j|=dSdSdS)NrTF)remove_writer_countr)r?rrrr_remove_writers  zTestLoop._remove_writercGs^||jkstd||j|}|j|ks>td|j||j|ksZtd|j|dS)Nzfd {} is not registeredz {!r} != {!r})rrrArr)r?rrrBrrrr assert_writers   zTestLoop.assert_writerc Cs~t|tsFyt|}Wn*tttfk rDtd|dYnXy|j|}Wntk rhYnXt d||dS)NzInvalid file object: {!r}z.File descriptor {!r} is used by transport {!r}) isinstanceintfilenoAttributeError TypeError ValueErrorrArKeyError RuntimeError)r?rZ transportrrr_ensure_fd_no_transports z TestLoop._ensure_fd_no_transportcGs|||j||f|S)zAdd a reader callback.)rr)r?rrrBrrr add_readers zTestLoop.add_readercCs||||S)zRemove a reader callback.)rr)r?rrrr remove_readers zTestLoop.remove_readercGs|||j||f|S)zAdd a writer callback..)rr)r?rrrBrrr add_writers zTestLoop.add_writercCs||||S)zRemove a writer callback.)rr)r?rrrr remove_writers zTestLoop.remove_writercCstt|_tt|_dS)N) collections defaultdictrrr)r?rrrrs zTestLoop.reset_counterscs:tx$|jD]}|j|}||qWg|_dS)N)rI _run_oncerrrr)r?whenr)rOrrrs    zTestLoop._run_once)rUcs&|j|tj||f|d|iS)NrU)rappendrIcall_at)r?rrrUrB)rOrrrs zTestLoop.call_atcCsdS)Nr)r?Z event_listrrr_process_eventsszTestLoop._process_eventscCsdS)Nr)r?rrr_write_to_selfszTestLoop._write_to_self)N)rDrErF__doc__rr6rr0rrrrrrrrrrrrrrrrrrRrr)rOrr.s,   rcKstjfddgi|S)Nspec__call__)rZMock)kwargsrrrrsrc@seZdZdZddZdS) MockPatternzA 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')) cCsttt||tjS)N)boolresearchstrS)r?otherrrr__eq__szMockPattern.__eq__N)rDrErFrrrrrrrsrc@seZdZddZddZdS)MockInstanceOfcCs ||_dS)N)_type)r?rrrrrszMockInstanceOf.__init__cCs t||jS)N)rr)r?rrrrrszMockInstanceOf.__eq__N)rDrErFrrrrrrrsrcCs$t|}|dkr td|f|S)Nzunable to get the source of %r)rZ_get_function_sourcer)funcsourcerrrget_function_sources rc@sHeZdZeddZddddZddd Zd d Zd d ZddZ dS)TestCasecCs&|j}|dk r|jdd|dS)NT)wait)Z_default_executorrcr0)r1executorrrr close_loops zTestCase.close_loopT)cleanupcCs,|dk s ttd|r(||j|dS)N)rrset_event_loopZ addCleanupr)r?r1rrrrrs  zTestCase.set_event_loopNcCst|}|||S)N)rr)r?r2r1rrr new_test_loops zTestCase.new_test_loopcCs |jt_dS)N)_get_running_loopr)r?rrrunpatch_get_running_loopsz!TestCase.unpatch_get_running_loopcCs tj|_ddt_t|_dS)NcSsdS)Nrrrrrr] r^z TestCase.setUp..)rrr Zthreading_setup_thread_cleanup)r?rrrsetUp s zTestCase.setUpcCsB|td|td|tj|j t dS)N)NNN) rrrZ assertEqualsysexc_infoZ doCleanupsr Zthreading_cleanuprZ reap_children)r?rrrtearDowns   zTestCase.tearDown)N) rDrErF staticmethodrrrrrrrrrrrs   rc cs2tj}zttjddVWdt|XdS)zrContext manager to disable asyncio logger. For example, it can be used to ignore warnings in debug mode. rN)r levelZsetLevelloggingZCRITICAL)Z old_levelrrrdisable_loggers  rcCs*ttj}||_||_||_d|j_|S)z'Create a mock of a non-blocking socket.g)rZ MagicMocksocketprotorfamilyZ gettimeoutr{)rrrZsockrrrmock_nonblocking_socket*s  r)r5)Nrr contextlibr=rrrrrrirrprar6ZunittestrrZ http.serverrZwsgiref.simple_serverrrr" ImportErrorZasynciorrrr r Z asyncio.logr Ztestr rr%r&ZSIGNED_CERTFILEZ SIGNING_CAZPEERCERTr*r,r-r4r9r;r<rGrSrWrfrrjrgrlrnrorscontextmanagerrurvryrZ BaseSelectorrZ BaseEventLooprrrrrrrrZ IPPROTO_TCPZ SOCK_STREAMZAF_INETrrrrrs                   % *