← Back
Editing: backports.cpython-311.pyc
� �|oi� � � � d dl Z e j dd� dk r7d dlZd dlZd dlmZ d dlmZmZm Z d dl m Z d� Z d� Z d � Z G d � d e � � ZdS d d lmZ d d l mZ dS )� N� )� � )� _is_coroutine)�ismethod� isfunction� CO_COROUTINE)�TestCasec �| � t | t j � � r!| j } t | t j � � �!| S �N)� isinstance� functools�partial�func)r s �`/builddir/build/BUILD/cloudlinux-venv-1.0.10/venv/lib/python3.11/site-packages/mock/backports.py�_unwrap_partialr s: � ���y� 0�1�1� ��9�D� ��y� 0�1�1� �� � c �� � t | � � r| j } t | � � �t | � � } t | � � sdS t | j j |z � � S )z�Return true if ``f`` is a function (or a method or functools.partial wrapper wrapping a function) whose code object has the given ``flag`` set in its flags.F)r �__func__r r �bool�__code__�co_flags)�f�flags r �_has_code_flagr sa � � �q�k�k� �� �A� �q�k�k� � �A� � ���!�}�}� ��5��A�J�'�$�.�/�/�/r c �\ � t | t � � pt | dd� � t u S )z}Return true if the object is a coroutine function. Coroutine functions are defined with "async def" syntax. r N)r r �getattrr )�objs r �iscoroutinefunctionr s0 � � �3� � -� -� A� �C� �$� /� /�=� @� r c �> � � e Zd Zd� fd� Zd� Zd� Zd� Zd � fd� Z� xZS ) �IsolatedAsyncioTestCase�runTestc �f �� t � � � |� � d | _ d | _ d S r )�super�__init__�_asyncioTestLoop�_asyncioCallsQueue)�self� methodName� __class__s �r r% z IsolatedAsyncioTestCase.__init__( s1 �� � �G�G� � �Z� (� (� (�$(�D� !�&*�D� #� #� #r c � �� K � t j � � x| _ }|� d � � |� � � � d {V ��}|� � � |�J ��3r )�asyncio�Queuer' � set_result�get� task_done)r( �fut�queue�querys r �_asyncioLoopRunnerz*IsolatedAsyncioTestCase._asyncioLoopRunner- sm � � � �.5�m�o�o� =�D� #�e� �N�N�4� � � � %� #�i�i�k�k�)�)�)�)�)�)�����!�!�!� �}�}�}� %r c �J � | j �J �t j � � }t j |� � |� d� � || _ |� � � }|� | � |� � � � | _ |� |� � d S )NT) r& r, �new_event_loop�set_event_loop� set_debug� create_future� create_taskr4 �_asyncioCallsTask�run_until_complete)r( �loopr1 s r �_setupAsyncioLoopz)IsolatedAsyncioTestCase._setupAsyncioLoop5 s� � ��(�0�0�0��)�+�+�D� � "�4� (� (� (� �N�N�4� � � �$(�D� !��$�$�&�&�C�%)�%5�%5�d�6M�6M�c�6R�6R�%S�%S�D� "� � #� #�C� (� (� (� (� (r c � � | j �J �| j }d | _ | j � d � � |� | j � � � � � |� |� � � � � t j d � � |� � � d S # t j d � � |� � � w xY wr ) r& r' � put_nowaitr< �join�shutdown_asyncgensr, r7 �close)r( r= s r �_tearDownAsyncioLoopz,IsolatedAsyncioTestCase._tearDownAsyncioLoop? s� � ��(�4�4�4��(�D�$(�D� !� � #� .� .�t� 4� 4� 4� � #� #�D�$;�$@�$@�$B�$B� C� C� C� ��'�'��(?�(?�(A�(A�B�B�B��&�t�,�,�,�� � � � � � � �� �&�t�,�,�,�� � � � � � ���s � 'B0 �0*CNc �� �� | � � � t � � � |� � | � � � S # | � � � w xY wr )r> r$ �runrD )r( �resultr* s �r rF zIsolatedAsyncioTestCase.runM sZ �� � � "� "� $� $� $� ,� �w�w�{�{�6�*�*��)�)�+�+�+�+���)�)�+�+�+�+���s � A � A")r" r ) �__name__� __module__� __qualname__r% r4 r> rD rF � __classcell__)r* s @r r! r! & s� �� � � � � � +� +� +� +� +� +� %� %� %� )� )� )� � � � ,� ,� ,� ,� ,� ,� ,� ,� ,� ,r r! )r )r! )�sys� version_infor, r �asyncio.coroutinesr �inspectr r r �unittestr r r r r! � r r �<module>rR s �� � � � � ��B�Q�B��&� � � � � � � � � � �0�0�0�0�0�0�:�:�:�:�:�:�:�:�:�:�!�!�!�!�!�!�� � � 0� 0� 0� � � �,,� ,,� ,,� ,,� ,,�(� ,,� ,,� ,,� ,,� ,,�b ,�+�+�+�+�+�0�0�0�0�0�0�0�0r
Save File
Cancel