File: //opt/cloudlinux/venv/lib64/python3.11/site-packages/mock/__pycache__/backports.cpython-311.pyc
�
� h�
� � � 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 �C/opt/cloudlinux/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