i 1dZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddl mZddlmZddlmZmZ ddlZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZe j@dk(rddl!m"Z"nddlm"Z"dZ#dZ$dd6Z?Gd7d8e jZ@ejfd9ZAd:ZBd;ZCy#e$rdZY~wxYw)=zUtilities shared by tests.N)mock) HTTPServer)WSGIRequestHandler WSGIServer) base_events)events)futures) selectors)tasks) coroutine)loggerwin32) socketpaircVtytjtjSN)ssl SSLContextPROTOCOL_SSLv23L/var/www/html/aiagents/fe/lib/python3.12/site-packages/asyncio/test_utils.pydummy_ssl_contextr)s {~~c1122rctd}|}|j|}d|_ |j||j y#|j wxYw)Ncyrrrrroncezrun_briefly..once1s rF)r create_task_log_destroy_pendingrun_until_completeclose)looprgents r run_brieflyr$0s[  &C A#A "  s AA!ctj|z}|sa|0|tjz }|dkrtj|jt j d||s`yy)NrgMbP?r!)timer TimeoutErrorrr sleep)r!predtimeoutdeadlines r run_untilr-?scyy{W$Hf  ,G!|**,,  E => frcD|j|jy)zloop.stop() schedules _raise_stop_error() and run_forever() runs until _raise_stop_error() callback. this wont work if test waits for some IO events, because _raise_stop_error() runs before any of io events callbacks. N)stop run_foreverr&s rrun_oncer1Is  IIKrceZdZdZdZy)SilentWSGIRequestHandlerc*tjSr)ioStringIOselfs r get_stderrz#SilentWSGIRequestHandler.get_stderrUs{{}rcyrr)r8formatargss r log_messagez$SilentWSGIRequestHandler.log_messageX rN)__name__ __module__ __qualname__r9r=rrrr3r3Ss  rr3c(eZdZdZfdZdZxZS)SilentWSGIServercdt|\}}|j|j||fSrsuper get_request settimeoutrequest_timeoutr8request client_addr __class__s rrHzSilentWSGIServer.get_request`s4$w244//0 ##rcyrrr8rLclient_addresss r handle_errorzSilentWSGIServer.handle_errorer>r)r?r@rArJrHrR __classcell__rNs@rrCrC\sO$  rrCceZdZdZy)SSLWSGIServerMixincvtjjtjjtdd}tjj |sLtjjtjjtjdd}tjj|d}tjj|d}t j|||d} |j||||jy#t$rYywxYw) Nz..teststest test_asyncioz ssl_key.pemz ssl_cert.pemT)keyfilecertfile server_side) ospathjoindirname__file__isdirr wrap_socketRequestHandlerClassr OSError)r8rLrQherer[r\ssocks rfinish_requestz!SSLWSGIServerMixin.finish_requestks ww||BGGOOH5tWEww}}T"77<< < &8D'',,t]377<<n5(/)1,02   $ $UND A KKM   s#D,, D87D8N)r?r@rArirrrrVrVisrrVc eZdZy) SSLWSGIServerNr?r@rArrrrkrksrrkF)use_sslc#Kd}|r|n|}||tj|j_t j fd}|j  jj|jy#jj|jwxYww)Nc$d}dg}|||dgS)Nz200 OK)z Content-typez text/plains Test messager)environstart_responsestatusheaderss rappz_run_test_server..apps"12vw'  rc(jdS)Ng?) poll_interval) serve_forever)httpdsrz"_run_test_server..su***>r)target) r3set_appserver_addressaddress threadingThreadstartshutdown server_closer`)r}rm server_clsserver_ssl_clsrt server_class server_threadrxs @r_run_test_serverrs!&->*L ": ;E MM#((EM$$>@M     sA"C&B*1C2C  CAF_UNIXceZdZdZy)UnixHTTPServerc^tjj|d|_d|_y)N 127.0.0.1P) socketserverUnixStreamServer server_bind server_name server_portr7s rrzUnixHTTPServer.server_binds&  ) ) 5 5d ;*D !D rN)r?r@rArrrrrrs "rrc(eZdZdZdZfdZxZS)UnixWSGIServerrDcNtj||jyr)rr setup_environr7s rrzUnixWSGIServer.server_binds  & &t ,    rcdt|\}}|j|j|dfS)N)rrFrKs rrHzUnixWSGIServer.get_requests7#(7#6#8 G[   t33 4-- -r)r?r@rArJrrHrSrTs@rrrs ! . .rrceZdZdZy)SilentUnixWSGIServercyrrrPs rrRz!SilentUnixWSGIServer.handle_errors rN)r?r@rArRrrrrrs rrc eZdZy)UnixSSLWSGIServerNrlrrrrrs rrcntj5}|jcdddS#1swYyxYwr)tempfileNamedTemporaryFilename)files rgen_unix_socket_pathrs,  ( ( * d99   s +4c#Kt} | tj|y#t$rYywxYw# tj|w#t$rYwwxYwxYwwr)rr^unlinkrf)r_s runix_socket_pathrs[#% J  $    $  sR A"8)A" 5A"5A"AAA AAAAA"c#Kt5}t||ttEd{dddy7 #1swYyxYwwN)r}rmrr)rrrr)rmr_s rrun_test_unix_serverrsR   J4'g3G7HJ J J J J J J Js$ A868 A8AArhostportrmc#RKt||f|ttEd{y7wr)rrCrkrs rrun_test_serverrs&t g+;/<>>>s '%'ci}t|D]4}|jdr|jdr&td||<6t d|f|j z|S)N__ return_value TestProtocol)dir startswithendswith MockCallbacktype __bases__)basedctrs rmake_test_protocolrsc CD 4 ??4 T]]4%8  d3D 4 ?4$.. 8# > @@rc,eZdZdZddZdZdZdZy) TestSelectorci|_yrkeysr7s r__init__zTestSelector.__init__s  rNcTtj|d||}||j|<|S)Nr)r SelectorKeyr)r8fileobjr datakeys rregisterzTestSelector.registers+##GQ=  ' rc8|jj|Sr)rpop)r8rs r unregisterzTestSelector.unregistersyy}}W%%rcgSrr)r8r+s rselectzTestSelector.selects rc|jSrrr7s rget_mapzTestSelector.get_maps yyrr)r?r@rArrrrrrrrrrs &rrceZdZdZdfd ZdZdZfdZdZdZ dZ d Z d Z d Z d Zfd ZfdZdZdZxZS)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. ct|| d}d|_nd|_||_t |jd|_d|_g|_t|_ i|_ i|_ |jy)Nc3Kdywrrrrrr"zTestLoop.__init__..gens sFTrg& .>) rGr_check_on_close_gennext_time_clock_resolution_timersr _selectorreaderswritersreset_counters)r8r"rNs rrzTestLoop.__init__su  ; #(D #'D E  TYY !% %   rc|jSrrr7s rr'z TestLoop.time-s zzrc4|r|xj|z c_yy)zMove test time forward.Nr)r8advances r advance_timezTestLoop.advance_time0s  JJ' !J rct||jr' |jj dt dy#t $rYywxYw)NrzTime generator is not finished)rGr rrsendAssertionError StopIteration)r8rNs rr zTestLoop.close5sR      G q!%%EFF !  sA AAcLtj||||j|<yr)r Handlerr8fdcallbackr<s r add_readerzTestLoop.add_reader?!==4> Rrcj|j|xxdz cc<||jvr|j|=yyNrTF)remove_reader_countrr8rs r remove_readerzTestLoop.remove_readerB6   $)$   R rc*||jvsJdj||j|}|j|k(s!Jdj|j||j|k(s!Jdj|j|yNzfd {} is not registeredz {!r} != {!r})rr; _callback_argsr8rrr<handles r assert_readerzTestLoop.assert_readerJT\\!G#<#C#CB#GG!b!8+ (^-B-B   h.( (+||t# ^%:%: LL$&  #rcLtj||||j|<yr)r rrrs r add_writerzTestLoop.add_writerRrrcj|j|xxdz cc<||jvr|j|=yyr)remove_writer_countrrs r remove_writerzTestLoop.remove_writerUrrc*||jvsJdj||j|}|j|k(s!Jdj|j||j|k(s!Jdj|j|yr)rr;rrrs r assert_writerzTestLoop.assert_writer]rrc|tjt|_tjt|_yr) collections defaultdictintrrr7s rrzTestLoop.reset_counterses(#.#:#:3#? #.#:#:3#? rct||jD].}|jj |}|j |0g|_yr)rG _run_oncerrrr)r8whenrrNs rrzTestLoop._run_onceisJ LL 'DiinnT*G   g & ' rc\|jj|t| ||g|Sr)rappendrGcall_at)r8r rr<rNs rr zTestLoop.call_atps+ D!wtX555rcyrr)r8 event_lists r_process_eventszTestLoop._process_eventstsrcyrrr7s r_write_to_selfzTestLoop._write_to_selfwr>rr)r?r@rA__doc__rr'rr rrrrrrrrr rrrSrTs@rrrsU$*" G? ? @6 rrc 2tjdddgi|S)Nspec__call__r)rMock)kwargss rrr{s 99 1:, 1& 11rceZdZdZdZy) 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')) cpttjt||tjSr)boolresearchstrS)r8others r__eq__zMockPattern.__eq__s"BIIc$i566rN)r?r@rArr!rrrrrs 7rrcPtj|}|td||S)Nzunable to get the source of )r _get_function_source ValueError)funcsources rget_function_sourcer's*  ( ( .F ~TCDD Mrc&eZdZdddZddZdZy)TestCaseT)cleanupcr|Jtjd|r|j|jyyr)r set_event_loop addCleanupr )r8r!r*s rr,zTestCase.set_event_loops4d#  OODJJ ' rNc>t|}|j||Sr)rr,)r8r"r!s r new_test_loopzTestCase.new_test_loops} D! rcvtjd|jtjdy)N)NNN)r r, assertEqualsysexc_infor7s rtearDownzTestCase.tearDowns)d# );rRs     "@!<<7)!3 ? 1   z  2 &(8 */2 69"66 "..(~  .0D  (-JJ 'a>> A9))(s {((s l2 7# 7=x  =*  # #*K  Cs GGG