File manager - Edit - /opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/__pycache__/testing.cpython-310.pyc
Back
o ;jvp � @ sp d Z ddlmZmZmZ z.ddlmZ ddlmZ ddl m Z ddl m Z ddl mZmZ ddlmZ dd lmZ W n eyQ d Zd Zd Z d Zd Zd Z d ZY nw dd lmZmZ dd lmZ dd lmZmZm Z dd l Z dd l Z dd l Z dd l Z dd l!Z!dd l"Z"dd l#Z#dd l$Z$e r�ddl%m&Z& nddl'm&Z& zddl(m)Z* W n ey� ddl+m*Z* Y nw e$j,dkr�e j-Z-e j.Z.ndd� Z-Z.e r�dd l/Z/nzdd l0Z/W n ey� dd l/Z/Y nw da1dd� Z2d-dd�Z3dd� Z4G d d � d e5�Z6G d d � d e/j7�Z8G d d!� d!e8�Z9G d"d#� d#e9�Z:d.d$d%�Z;de;_<G d&d'� d'e/j7�Z=G d(d)� d)e j>�Z?d*d+� Z@eAd,k�r6e@� d S d S )/a� Support classes for automated testing. * `AsyncTestCase` and `AsyncHTTPTestCase`: Subclasses of unittest.TestCase with additional support for testing asynchronous (`.IOLoop`-based) code. * `ExpectLog` and `LogTrapTestCase`: Make test logs less spammy. * `main()`: A simple test runner (wrapper around unittest.main()) with support for the tornado.autoreload module to rerun the tests when code changes. � )�absolute_import�division�print_function)�gen)�AsyncHTTPClient)� HTTPServer)�SimpleAsyncHTTPClient)�IOLoop� TimeoutError)�netutil)� SubprocessN)�gen_log�app_log)�ExceptionStackContext)�raise_exc_info�basestring_type�PY3)�StringIO)� Generator)� GeneratorType)� � c C � dS �NF� ��fr r �L/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/testing.py�<lambda>< s r i' c C s t } t d a | S )a@ Returns a (hopefully) unused port number. This function does not guarantee that the port it returns is available, only that a series of get_unused_port calls in a single process return distinct ports. .. deprecated:: Use bind_unused_port instead, which is guaranteed to find an unused port. � )� _next_port)�portr r r �get_unused_portP s r" Fc C s, t jddtj| d�d }|�� d }||fS )z�Binds a server socket to an available port on localhost. Returns a tuple (socket, port). .. versionchanged:: 4.4 Always binds to ``127.0.0.1`` without resolving the name ``localhost``. Nz 127.0.0.1)�family� reuse_portr r )r Z bind_sockets�socket�AF_INET� getsockname)r$ �sockr! r r r �bind_unused_port` s �� r) c C s, z t tj�d��W S ttfy Y dS w )z}Get the global timeout setting for async tests. Returns a float, the timeout in seconds. .. versionadded:: 3.1 ZASYNC_TEST_TIMEOUTr )�float�os�environ�get� ValueError� TypeErrorr r r r �get_async_test_timeouto s �r0 c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �_TestMethodWrappera~ Wraps a test method to raise an error if it returns a value. This is mainly used to detect undecorated generators (if a test method yields it must use a decorator to consume the generator), but will also detect other kinds of return values (these are not necessarily errors, but we alert anyway since there is no good reason to return a value from a test). c C s || _ d S �N)� orig_method)�selfr3 r r r �__init__� s z_TestMethodWrapper.__init__c O sB | j |i |��}t|t�st|�rtd��|d ur td| ��d S )NzVGenerator and coroutine test methods should be decorated with tornado.testing.gen_testz)Return value from test method ignored: %r)r3 � isinstancer � iscoroutiner/ r. �r4 �args�kwargs�resultr r r �__call__� s ��z_TestMethodWrapper.__call__c C s t | j|�S )z�Proxy all unknown attributes to the original method. This is important for some of the decorators in the `unittest` module, such as `unittest.skipIf`. )�getattrr3 )r4 �namer r r � __getattr__� s z _TestMethodWrapper.__getattr__N)�__name__� __module__� __qualname__�__doc__r5 r< r? r r r r r1 | s r1 c st e Zd ZdZd� fdd� Z� fdd�Z� fdd�Zd d � Zd d � Zd d� Z d� fdd� Z ddd�Z ddd�Z � Z S )� AsyncTestCaseak `~unittest.TestCase` subclass for testing `.IOLoop`-based asynchronous code. The unittest framework is synchronous, so the test must be complete by the time the test method returns. This means that asynchronous code cannot be used in quite the same way as usual. To write test functions that use the same ``yield``-based patterns used with the `tornado.gen` module, decorate your test methods with `tornado.testing.gen_test` instead of `tornado.gen.coroutine`. This class also provides the `stop()` and `wait()` methods for a more manual style of testing. The test method itself must call ``self.wait()``, and asynchronous callbacks should call ``self.stop()`` to signal completion. By default, a new `.IOLoop` is constructed for each test and is available as ``self.io_loop``. This `.IOLoop` should be used in the construction of HTTP clients/servers, etc. If the code being tested requires a global `.IOLoop`, subclasses should override `get_new_ioloop` to return it. The `.IOLoop`'s ``start`` and ``stop`` methods should not be called directly. Instead, use `self.stop <stop>` and `self.wait <wait>`. Arguments passed to ``self.stop`` are returned from ``self.wait``. It is possible to have multiple ``wait``/``stop`` cycles in the same test. Example:: # This test uses coroutine style. class MyTestCase(AsyncTestCase): @tornado.testing.gen_test def test_http_fetch(self): client = AsyncHTTPClient(self.io_loop) response = yield client.fetch("http://www.tornadoweb.org") # Test contents of response self.assertIn("FriendFeed", response.body) # This test uses argument passing between self.stop and self.wait. class MyTestCase2(AsyncTestCase): def test_http_fetch(self): client = AsyncHTTPClient(self.io_loop) client.fetch("http://www.tornadoweb.org/", self.stop) response = self.wait() # Test contents of response self.assertIn("FriendFeed", response.body) # This test uses an explicit callback-based style. class MyTestCase3(AsyncTestCase): def test_http_fetch(self): client = AsyncHTTPClient(self.io_loop) client.fetch("http://www.tornadoweb.org/", self.handle_fetch) self.wait() def handle_fetch(self, response): # Test contents of response (failures and exceptions here # will cause self.wait() to throw an exception and end the # test). # Exceptions thrown here are magically propagated to # self.wait() in test_http_fetch() via stack_context. self.assertIn("FriendFeed", response.body) self.stop() �runTestc sZ t t| ��|� d| _d| _d | _d | _d | _t| |d �}|d ur+t | |t |�� d S d S r ) �superrD r5 �_AsyncTestCase__stopped�_AsyncTestCase__running�_AsyncTestCase__failure�_AsyncTestCase__stop_args�_AsyncTestCase__timeoutr= �setattrr1 )r4 Z methodName�method�� __class__r r r5 � s �zAsyncTestCase.__init__c s& t t| ��� | �� | _| j�� d S r2 )rF rD �setUp�get_new_ioloop�io_loopZ make_current�r4 rN r r rP � s zAsyncTestCase.setUpc sP t �� | j�� t�� r| jt�� ur| jjdd� tt | �� � | � � d S )NT)Zall_fds) r Z uninitializerR Z clear_currentr � initialized�instance�closerF rD �tearDown�_AsyncTestCase__rethrowrS rN r r rW � s zAsyncTestCase.tearDownc C s t � S )z�Creates a new `.IOLoop` for this test. May be overridden in subclasses for tests that require a specific `.IOLoop` (usually the singleton `.IOLoop.instance()`). )r rS r r r rQ � z AsyncTestCase.get_new_ioloopc C s8 | j d u r |||f| _ n tjd|||fd� | �� dS )Nz%multiple unhandled exceptions in test)�exc_infoT)rI r �error�stop�r4 �typ�value�tbr r r �_handle_exception s �z AsyncTestCase._handle_exceptionc C s& | j d ur| j }d | _ t|� d S d S r2 )rI r )r4 Zfailurer r r Z __rethrow s �zAsyncTestCase.__rethrowNc sF t | j�� tt| ��|� W d � n1 sw Y | �� d S r2 )r ra rF rD �runrX )r4 r; rN r r rb s � zAsyncTestCase.runc K s: |du s|rJ �|p || _ | jr| j�� d| _d| _dS )z�Stops the `.IOLoop`, causing one pending (or future) call to `wait()` to return. Keyword arguments or a single positional argument passed to `stop()` are saved and will be returned by `wait()`. NFT)rJ rH rR r\ rG )r4 Z_argr: r r r r\ ' s zAsyncTestCase.stopc s� �du rt � �� jsF�r � �fdd�}� j�� j�� � |�� _ d� _� j�� � jdus5|du s5|� r6nq!� jdurF� j� � j� d� _� jsKJ �d� _� � � � j }d� _ |S )a" Runs the `.IOLoop` until stop is called or timeout has passed. In the event of a timeout, an exception will be thrown. The default timeout is 5 seconds; it may be overridden with a ``timeout`` keyword argument or globally with the ``ASYNC_TEST_TIMEOUT`` environment variable. If ``condition`` is not None, the `.IOLoop` will be restarted after `stop()` until ``condition()`` returns true. .. versionchanged:: 3.1 Added the ``ASYNC_TEST_TIMEOUT`` environment variable. Nc s8 z� � d� �� ty t�� � _Y nw � �� d S )Nz*Async operation timed out after %s seconds)ZfailureException� Exception�sysrZ rI r\ r �r4 �timeoutr r � timeout_funcH s �� � z(AsyncTestCase.wait.<locals>.timeout_funcTF) r0 rG rR Z add_timeout�timerK rH �startrI Zremove_timeoutrX rJ )r4 Z conditionrf rg r; r re r �wait5 s, � zAsyncTestCase.wait)rE r2 �NN)r@ rA rB rC r5 rP rW rQ ra rX rb r\ rj � __classcell__r r rN r rD � s = rD c sl e Zd ZdZ� fdd�Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z � fdd�Z � ZS )�AsyncHTTPTestCasear A test case that starts up an HTTP server. Subclasses must override `get_app()`, which returns the `tornado.web.Application` (or other `.HTTPServer` callback) to be tested. Tests will typically use the provided ``self.http_client`` to fetch URLs from this server. Example, assuming the "Hello, world" example from the user guide is in ``hello.py``:: import hello class TestHelloApp(AsyncHTTPTestCase): def get_app(self): return hello.make_app() def test_homepage(self): response = self.fetch('/') self.assertEqual(response.code, 200) self.assertEqual(response.body, 'Hello, world') That call to ``self.fetch()`` is equivalent to :: self.http_client.fetch(self.get_url('/'), self.stop) response = self.wait() which illustrates how AsyncTestCase can turn an asynchronous operation, like ``http_client.fetch()``, into a synchronous operation. If you need to do other asynchronous operations in tests, you'll probably need to use ``stop()`` and ``wait()`` yourself. c sN t t| ��� t� \}}|| _| �� | _| �� | _| � � | _ | j � |g� d S r2 ) rF rm rP r) �_AsyncHTTPTestCase__port�get_http_client� http_client�get_app�_app�get_http_server� http_serverZ add_sockets)r4 r( r! rN r r rP � s zAsyncHTTPTestCase.setUpc C s t | jd�S )N)rR )r rR rS r r r ro � s z!AsyncHTTPTestCase.get_http_clientc C s t | jfd| ji| �� ��S )NrR )r rr rR �get_httpserver_optionsrS r r r rs � s �z!AsyncHTTPTestCase.get_http_serverc C s t � �)zzShould be overridden by subclasses to return a `tornado.web.Application` or other `.HTTPServer` callback. )�NotImplementedErrorrS r r r rq � s zAsyncHTTPTestCase.get_appc K s& | j j| �|�| jfi |�� | �� S )a1 Convenience method to synchronously fetch a url. The given path will be appended to the local server's host and port. Any additional kwargs will be passed directly to `.AsyncHTTPClient.fetch` (and so could be used to pass ``method="POST"``, ``body="..."``, etc). )rp �fetch�get_urlr\ rj )r4 �pathr: r r r rw � s zAsyncHTTPTestCase.fetchc C s i S )zgMay be overridden by subclasses to return additional keyword arguments for the server. r rS r r r ru � s z(AsyncHTTPTestCase.get_httpserver_optionsc C s | j S )zZReturns the port used by the server. A new port is chosen for each test. )rn rS r r r � get_http_port� rY z AsyncHTTPTestCase.get_http_portc C r )N�httpr rS r r r � get_protocol� � z AsyncHTTPTestCase.get_protocolc C s d| � � | �� |f S )z>Returns an absolute url for the given path on the test server.z%s://127.0.0.1:%s%s)r| rz )r4 ry r r r rx � s �zAsyncHTTPTestCase.get_urlc sT | j �� | jj| j jt� d� t�� r | jjt� � ur!| j� � t t | �� � d S �N�rf )rt r\ rR �run_syncZclose_all_connectionsr0 r rT rp rU rV rF rm rW rS rN r r rW � s � zAsyncHTTPTestCase.tearDown)r@ rA rB rC rP ro rs rq rw ru rz r| rx rW rl r r rN r rm b s rm c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) �AsyncHTTPSTestCasezjA test case that starts an HTTPS server. Interface is generally the same as `AsyncHTTPTestCase`. c C s t | jdtdd�d�S )NTF)Z validate_cert)rR Zforce_instance�defaults)r rR �dictrS r r r ro � s �z"AsyncHTTPSTestCase.get_http_clientc C s t | �� d�S )N)Z ssl_options)r� �get_ssl_optionsrS r r r ru � s z)AsyncHTTPSTestCase.get_httpserver_optionsc C s0 t j�t�}tt j�|dd�t j�|dd�d�S )zMay be overridden by subclasses to select SSL options. By default includes a self-signed testing certificate. �testztest.crtztest.key)�certfile�keyfile)r+ ry �dirname�__file__r� �join)r4 Z module_dirr r r r� � s �z"AsyncHTTPSTestCase.get_ssl_optionsc C r )N�httpsr rS r r r r| � r} z AsyncHTTPSTestCase.get_protocolN)r@ rA rB rC ro ru r� r| r r r r r� � s r� c s. � du rt � � � fdd�}| dur|| �S |S )a> Testing equivalent of ``@gen.coroutine``, to be applied to test methods. ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not already running. ``@gen_test`` should be applied to test methods on subclasses of `AsyncTestCase`. Example:: class MyTest(AsyncHTTPTestCase): @gen_test def test_something(self): response = yield gen.Task(self.fetch('/')) By default, ``@gen_test`` times out after 5 seconds. The timeout may be overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable, or for each test with the ``timeout`` keyword argument:: class MyTest(AsyncHTTPTestCase): @gen_test(timeout=10) def test_something_slow(self): response = yield gen.Task(self.fetch('/')) .. versionadded:: 3.1 The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment variable. .. versionchanged:: 4.0 The wrapper now passes along ``*args, **kwargs`` so it can be used on functions with arguments. Nc sJ t ����fdd��}t��r|� nt�|�� t �� �� �fdd��}|S )Nc s< � | g|�R i |��}t |t�st|�r|| _|S d | _|S r2 )r6 r r7 �_test_generatorr8 r r r � pre_coroutine s �z-gen_test.<locals>.wrap.<locals>.pre_coroutinec sP z| j jtj� | g|�R i |���d�W S ty' } z| j�|� � d }~ww r~ )rR r� � functools�partialr r� �throw)r4 r9 r: �e)�cororf r r �post_coroutine s � ��z.gen_test.<locals>.wrap.<locals>.post_coroutine)r� �wraps�iscoroutinefunctionr � coroutine)r r� r� r )r� r r �wrap s zgen_test.<locals>.wrap)r0 )�funcrf r� r r r �gen_test� s (r� c s" e Zd ZdZd� fdd� Z� ZS )�LogTrapTestCaseaQ A test case that captures and discards all logging output if the test passes. Some libraries can produce a lot of logging output even when the test succeeds, so this class can be useful to minimize the noise. Simply use it as a base class for your test case. It is safe to combine with AsyncTestCase via multiple inheritance (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``) This class assumes that only one log handler is configured and that it is a `~logging.StreamHandler`. This is true for both `logging.basicConfig` and the "pretty logging" configured by `tornado.options`. It is not compatible with other log buffering mechanisms, such as those provided by some test runners. .. deprecated:: 4.1 Use the unittest module's ``--buffer`` option instead, or `.ExpectLog`. Nc s� t �� }|js t �� |jd }t|j�dks t|t j�s'tt| �� |� d S |j }zAt � |_ t � dt| � � t|j�t|j� }tt| �� |� t|j�t|j� }||krf|�|j �� � W ||_ d S W ||_ d S ||_ w )Nr r zRUNNING TEST: )�logging� getLogger�handlers� basicConfig�lenr6 � StreamHandlerrF r� rb �streamr r �info�strZfailures�errors�write�getvalue)r4 r; �logger�handlerZ old_streamZold_error_countZnew_error_countrN r r rb L s* � �zLogTrapTestCase.runr2 )r@ rA rB rC rb rl r r rN r r� 9 s r� c @ s2 e Zd ZdZd dd�Zdd� Zdd� Zd d � Zd S ) � ExpectLoga� Context manager to capture and suppress expected log output. Useful to make tests of error conditions less noisy, while still leaving unexpected log entries visible. *Not thread safe.* The attribute ``logged_stack`` is set to true if any exception stack trace was logged. Usage:: with ExpectLog('tornado.application', "Uncaught exception"): error_response = self.fetch("/some_page") .. versionchanged:: 4.3 Added the ``logged_stack`` attribute. Tc C s< t |t�r t�|�}|| _t�|�| _|| _d| _ d| _ dS )a� Constructs an ExpectLog context manager. :param logger: Logger object (or name of logger) to watch. Pass an empty string to watch the root logger. :param regex: Regular expression to match. Any log entries on the specified logger that match this regex will be suppressed. :param required: If true, an exception will be raised if the end of the ``with`` statement is reached without matching any log entries. FN) r6 r r� r� r� �re�compile�regex�required�matched� logged_stack)r4 r� r� r� r r r r5 u s zExpectLog.__init__c C s. |j rd| _|�� }| j�|�rd| _dS dS )NTF)rZ r� � getMessager� �matchr� )r4 �record�messager r r �filter� s zExpectLog.filterc C s | j �| � | S r2 )r� � addFilterrS r r r � __enter__� s zExpectLog.__enter__c C s0 | j �| � |s| jr| jstd��d S d S d S )Nz did not get expected log message)r� � removeFilterr� r� rc r] r r r �__exit__� s �zExpectLog.__exit__N)T)r@ rA rB rC r5 r� r� r� r r r r r� d s r� c K s� ddl m}m}m} |dtddd� |dtd� |d td� |d td� |d td� |d td� tjd g|tj� }|jsFt�tj tj � |j d urOd| d< |j d urXd| d< |j d urad| d < |jd urjd| d< |jd ursd| d < tdkr�t|�dkr�tdtjd� t�d� z"t|�dkr�tjdd |d�| �� W d S tjdd|d�| �� W d S ty� } z|jdkr�t�d� � t�d� � d }~ww )aa A simple test runner. This test runner is essentially equivalent to `unittest.main` from the standard library, but adds support for tornado-style option parsing and log formatting. It is *not* necessary to use this `main` function to run tests using `AsyncTestCase`; these tests are self-contained and can run with any test runner. The easiest way to run a test is via the command line:: python -m tornado.testing tornado.test.stack_context_test See the standard library unittest module for ways in which tests can be specified. Projects with many tests may wish to define a test script like ``tornado/test/runtests.py``. This script should define a method ``all()`` which returns a test suite and then call `tornado.testing.main()`. Note that even when a test script is used, the ``all()`` test suite may be overridden by naming a single test on the command line:: # Runs all tests python -m tornado.test.runtests # Runs one test python -m tornado.test.runtests tornado.test.stack_context_test Additional keyword arguments passed through to ``unittest.main()``. For example, use ``tornado.testing.main(verbosity=2)`` to show many test details as they are run. See http://docs.python.org/library/unittest.html#unittest.main for full argument list. r )�define�options�parse_command_line�exception_on_interruptTz�If true (default), ctrl-c raises a KeyboardInterrupt exception. This prints a stack trace but cannot interrupt certain operations. If false, the process is more reliably killed, but does not print a stack trace.)�type�default�help�verbose)r� �quiet�failfast�catch�bufferN� � verbosityZ catchbreak�__main__r zNo tests specified)�file)�module�argv�all)Z defaultTestr� ZPASSZFAILr ) Zsalt.ext.tornado.optionsr� r� r� �boolrd r� r� �signal�SIGINT�SIG_DFLr� r� r� r� r� r@ r� �print�stderr�exit�unittest�main� SystemExit�coder r� r[ )r: r� r� r� r� r� r r r r� � sJ "� ���r� r� )Frk )BrC � __future__r r r Zsalt.ext.tornador Zsalt.ext.tornado.httpclientr Zsalt.ext.tornado.httpserverr Z"salt.ext.tornado.simple_httpclientr Zsalt.ext.tornado.ioloopr r r Zsalt.ext.tornado.processr � ImportErrorZsalt.ext.tornado.logr r Z salt.ext.tornado.stack_contextr Zsalt.ext.tornado.utilr r r r� �inspectr� r+ r� r� r% rd �ior Z cStringIO�collections.abcr r �types� version_infor7 r� r� Z unittest2r r" r) r0 �objectr1 ZTestCaserD rm r� r� Z__test__r� �Filterr� r� r@ r r r r �<module> s� � � � Ia W+6 W �
| ver. 1.4 |
Github
|
.
| PHP 8.2.30 | Generation time: 0 |
proxy
|
phpinfo
|
Settings