File: //opt/osm/venv/lib/python3.10/site-packages/psutil/tests/__pycache__/test_testutils.cpython-310.pyc
o
=idI � @ s� d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl
Z
ddl Z ddl
m Z ddl
Z
ddlZ
ddl
mZ ddl
mZ ddl
mZ ddlmZ ddlmZ dd lmZ dd
lmZ dd lmZ dd lmZ dd
lmZ ddlmZ ddlmZ ddlm Z ddlm Z ddlm Z ddlm Z ddlm Z ddlm!Z! ddlm"Z" ddlm#Z# ddlm$Z$ ddlm%Z% ddlm&Z& ddlm'Z' dd lm(Z( dd lm)Z) dd lm*Z* dd lm+Z+ dd lm,Z, dd!lm-Z- dd"lm.Z. dd#lm/Z/ dd$lm0Z0 dd%lm1Z1 dd&lm2Z2 dd'lm3Z3 G d(d)� d)e �Z4G d*d+� d+e �Z5G d,d-� d-e �Z6G d.d/� d/e �Z7G d0d1� d1e �Z8e(j9j:d2d3�G d4d5� d5e ��Z;G d6d7� d7e �Z<G d8d9� d9e �Z=G d:d;� d;e �Z>dS )<z1Tests for testing utils (psutil.tests namespace).� N)�mock)�FREEBSD)�NETBSD)�POSIX)� open_binary)� open_text)�
supports_ipv6)�
CI_TESTING)�COVERAGE)�HAS_NET_CONNECTIONS_UNIX)�HERE)�
PYTHON_EXE)�PYTHON_EXE_ENV)�PsutilTestCase)�TestMemoryLeak)� bind_socket)�bind_unix_socket��
call_until)�chdir)�create_sockets)� fake_pytest)�filter_proc_net_connections)�
get_free_port)�
is_namedtuple)�process_namespace)�pytest)�
reap_children)�retry)�retry_on_failure)�
safe_mkdir)� safe_rmpath)�system_namespace)�tcp_socketpair)� terminate)�unix_socketpair)�
wait_for_file)� wait_for_pidc @ sx e Zd Ze�d�dd� �Ze�d�dd� �Ze�d�dd� �Ze�d�dd � �Ze�d�d
d � �Z e�d�d d
� �Z
dS )�TestRetryDecoratorz
time.sleepc sF t ddd d�� fdd��}ttd��� |� dksJ �|jdks!J �d S )N� � ��retries�intervalZlogfunc � � r � � � dd � sdS �Nr* r ��pop� ��queuer2 �I/opt/osm/venv/lib/python3.10/site-packages/psutil/tests/test_testutils.py�fooG �
�z2TestRetryDecorator.test_retry_success.<locals>.foo� )r �list�range�
call_count��self�sleepr6 r2 r3 r5 �test_retry_successC s
z%TestRetryDecorator.test_retry_successc sh t ddd d�� fdd��}ttd��� t�t�� |� W d � n1 s&w Y |jdks2J �d S )Nr) r* r+ c r. r/ r0 r2 r3 r2 r5 r6 U r7 z2TestRetryDecorator.test_retry_failure.<locals>.foo� )r r9 r: r �raises�ZeroDivisionErrorr; r< r2 r3 r5 �test_retry_failureR s �z%TestRetryDecorator.test_retry_failurec C sV t tdd�dd� �}t�t�� |� W d � n1 s w Y |jdks)J �d S )Nr* )� exceptionr- c S s t ��N)� TypeErrorr2 r2 r2 r5 r6 c � z2TestRetryDecorator.test_exception_arg.<locals>.foor )r �
ValueErrorr rA rF r; r< r2 r2 r5 �test_exception_arga s
�z%TestRetryDecorator.test_exception_argc C sX t dd d d�dd� �}t�t�� |� W d � n1 s w Y |jdks*J �d S )Nr) r+ c S � dd d S r/ r2 r2 r2 r2 r5 r6 o � z4TestRetryDecorator.test_no_interval_arg.<locals>.foor �r r rA rB r; r< r2 r2 r5 �test_no_interval_argk s
�z'TestRetryDecorator.test_no_interval_argc C sX t ddd d�dd� �}t�t�� |� W d � n1 s w Y |jdks*J �d S )Nr) r* r+ c S rJ r/ r2 r2 r2 r2 r5 r6 y rK z0TestRetryDecorator.test_retries_arg.<locals>.foorL r< r2 r2 r5 �test_retries_argw s
�z#TestRetryDecorator.test_retries_argc C s< t �t�� tddd� W d � d S 1 sw Y d S )Nr) r* )r, �timeout)r rA rH r )r= r> r2 r2 r5 � test_retries_and_timeout_args� s "�z0TestRetryDecorator.test_retries_and_timeout_argsN) �__name__�
__module__� __qualname__r �patchr? rC rI rM rN rP r2 r2 r2 r5 r( B s
r( c @ s< e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
S )�TestSyncTestUtilsc C s� t t�� � tt�� �d }tjdtdg�d��+ t �
tj �� t |� W d � n1 s.w Y W d � d S W d � d S 1 sFw Y d S )Ni�� �psutil.tests.retry.__iter__r �� return_value) r' �os�getpid�max�psutilZpidsr rT �iterr rA Z
NoSuchProcess)r= Znopidr2 r2 r5 �test_wait_for_pid� s
��"�z#TestSyncTestUtils.test_wait_for_pidc C sX | � � }t|d��
}|�d� W d � n1 sw Y t|� tj�|�r*J �d S )N�wr6 ��
get_testfn�open�writer& rY �path�exists�r= �testfn�fr2 r2 r5 �test_wait_for_file� s �z$TestSyncTestUtils.test_wait_for_filec C sR | � � }t|d�� W d � n1 sw Y t|dd� tj�|�r'J �d S )Nr_ T)�empty)ra rb r& rY rd re �r= rg r2 r2 r5 �test_wait_for_file_empty� s � z*TestSyncTestUtils.test_wait_for_file_emptyc C s� | � � }tjdtdg�d��* t�t�� t|� W d � n1 s#w Y W d � d S W d � d S 1 s;w Y d S )NrV r rW )ra r rT r] r rA �OSErrorr& rk r2 r2 r5 �test_wait_for_file_no_file� s
��"�z,TestSyncTestUtils.test_wait_for_file_no_filec C s\ | � � }t|d��
}|�d� W d � n1 sw Y t|dd� tj�|�s,J �d S )Nr_ r6 F)�deleter` rf r2 r2 r5 � test_wait_for_file_no_delete� s � z.TestSyncTestUtils.test_wait_for_file_no_deletec C s t dd� � d S )Nc S � dS �Nr* r2 r2 r2 r2 r5 �<lambda>� � z3TestSyncTestUtils.test_call_until.<locals>.<lambda>r �r= r2 r2 r5 �test_call_until� s z!TestSyncTestUtils.test_call_untilN) rQ rR rS r^ ri rl rn rp rv r2 r2 r2 r5 rU � s rU c @ s4 e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd S ) �TestFSTestUtilsc C �< t t��}|jdks J �W d � d S 1 sw Y d S )N�r)r �__file__�mode�r= rh r2 r2 r5 �test_open_text� �
"�z TestFSTestUtils.test_open_textc C rx )N�rb)r rz r{ r| r2 r2 r5 �test_open_binary� r~ z TestFSTestUtils.test_open_binaryc C s< | � � }t|� tj�|�sJ �t|� tj�|�s J �d S rE )ra r rY rd �isdirrk r2 r2 r5 �test_safe_mkdir� s
z TestFSTestUtils.test_safe_mkdirc C s� | � � }t|d��� t|� tj�|�rJ �t|� t�|� t|� tj�|�r,J �tj dt
t j d�d��'}t
�t
�� t|� W d � n1 sLw Y |jsVJ �W d � d S 1 saw Y d S )Nr_ zpsutil.tests.os.stat� )Z side_effect)ra rb �closer! rY rd re �mkdirr rT rm �errno�EINVALr rA Zcalled)r= rg �mr2 r2 r5 �test_safe_rmpath� s"
�
� "�z TestFSTestUtils.test_safe_rmpathc C sp | � � }t�� }t�|� t|�� t�� tj�||�ks J �W d � n1 s)w Y t�� |ks6J �d S rE )ra rY �getcwdr� r rd �join)r= rg �baser2 r2 r5 �
test_chdir� s
�zTestFSTestUtils.test_chdirN)rQ rR rS r} r� r� r� r� r2 r2 r2 r5 rw � s rw c @ s> e Zd Zdd� Zdd� Zejje dd�dd� �Z d d
� Z
d S ) �TestProcessUtilsc C sN | � � }t�|j�}|�� sJ �t� |�� rJ �tjjr J �tjjr%J �d S rE ) �
spawn_subprocr\ �Process�pid�
is_runningr �testsZ
_pids_startedZ_subprocesses_started)r= Zsubp�pr2 r2 r5 �test_reap_children� s z#TestProcessUtils.test_reap_childrenc C s� | � � \}}|j|jksJ �|�� sJ �|�� sJ �t�� �� }||gks'J �t�� jdd�}t|�dks7J �||v s=J �||v sCJ �|�� t� � ksMJ �|�� |jksVJ �t
|� |�� r`J �|�� sfJ �t
|� |�� rpJ �d S )NT)� recursive� ) Zspawn_children_pairr� r� r\ r� �children�len�ppidrY rZ r$ )r= �childZ
grandchildr� r2 r2 r5 �test_spawn_children_pair� s" z)TestProcessUtils.test_spawn_children_pair�
POSIX only��reasonc C s" | � � \}}|�� tjksJ �d S rE )� spawn_zombie�statusr\ Z
STATUS_ZOMBIE)r= Z_parent�zombier2 r2 r5 �test_spawn_zombie� s z"TestProcessUtils.test_spawn_zombiec C s� | � � }t|� | �|j� t|� t�| � � j�}t|� | �|j� t|� tddg}tj|tj tj t
d�}t|� | �|j� t|� | � � j}t|� | �|j� t|� t rw| � � \}}t|� t|� | �|j� | �|j� d S d S )Nz-cz3import time; [time.sleep(0.1) for x in range(100)];)�stdout�stderr�env)
r� r$ Zassert_pid_goner� r\ r� r
�Popen�
subprocess�PIPEr r r� )r= r� �cmdr� �parentr� r2 r2 r5 �test_terminate s@ ��
�z TestProcessUtils.test_terminateN) rQ rR rS r� r� r �mark�skipifr r� r� r2 r2 r2 r5 r� � s
r� c @ s~ e Zd Zdd� Zejje dd�dd� �Zdd� Z ejje dd�ejje
p'e d d�ejje d
d�d d � ���Z
d
d� ZdS )� TestNetUtilsc C sN t � }td|fd��}|�� d |ksJ �W d � d S 1 s w Y d S )Nr� ��addrr* )r r � getsockname)r= �port�sr2 r2 r5 r ) s "�zTestNetUtils.bind_socketr� r� c C s� | � � }t|��3}|jtjksJ �|jtjksJ �|�� |ks!J �tj �
|�s)J �t � t� |�j
�s4J �W d � n1 s>w Y | � � }t|tjd��}|jtjksWJ �W d � d S 1 sbw Y d S )N)�type)ra r �family�socket�AF_UNIXr� � SOCK_STREAMr� rY rd re �stat�S_ISSOCK�st_mode�
SOCK_DGRAM)r= �name�sockr2 r2 r5 �test_bind_unix_socket. s
�"�z"TestNetUtils.test_bind_unix_socketc C s� dt � f}ttj|d�\}}|�; |� |�� |ks J �|�� |ks$J �|�� |ks,J �W d � n1 s6w Y W d � d S W d � d S 1 sNw Y d S )Nz 127.0.0.1r� )r r# r� �AF_INETr� � getpeername)r= r� �server�clientr2 r2 r5 �test_tcp_socketpair< s
P�z TestNetUtils.test_tcp_socketpairz*/var/run/log UNIX socket opened by defaultzcan't list UNIX socketsc C s� t �� }|�� }t|jdd��rJ �| �� }t|�\}}zFtj� |�s%J �t
� t�
|�j �s0J �|�� | dks:J �t
t|jdd���dksHJ �|�� |ksPJ �|�� |ksXJ �W |�� |�� d S |�� |�� w )N�unix)�kindr� )r\ r� �num_fdsr Znet_connectionsra r% rY rd re r� r� r� r� r� r� r� )r= r� r� r� r� r� r2 r2 r5 �test_unix_socketpairE s( �� �
z!TestNetUtils.test_unix_socketpairc C s� t � �c}t�t�}t�t�}|D ]}||j d7 < ||�tjtj� d7 < q|tj dks3J �t
� r?|tj dks?J �t rLt
rL|tj dksLJ �|tj dksUJ �|tj dks^J �W d � d S 1 siw Y d S )Nr* r� )r � collections� defaultdict�intr� �
getsockoptr� �
SOL_SOCKET�SO_TYPEr� r �AF_INET6r r r� r� r� )r= �socksZfams�typesr� r2 r2 r5 �test_create_socketsb s
"�z TestNetUtils.test_create_socketsN)rQ rR rS r r r� r� r r� r� r r r r� r� r2 r2 r2 r5 r� ( s
�� r� �serial�r� c @ sh e Zd Ze� dd� �Zdd� Ze� ejje dd�ejje
dd�dd � ���Z d
d � Z d d
� Z
dd� ZdS )�TestMemLeakClassc s8 � fdd�}ddi� | j |ddd� � d dksJ �d S ) Nc s � d d7 < d S )N�cntr* r2 r2 �r� r2 r5 �funw s z(TestMemLeakClass.test_times.<locals>.funr� r �
� )�times� warmup_times� )�execute)r= r� r2 r� r5 �
test_timesu s zTestMemLeakClass.test_timesc C s2 t �t�� | jdd� dd� W d � n1 sw Y t �t�� | jdd� dd� W d � n1 s7w Y t �t�� | jdd� dd� W d � n1 sUw Y t �t�� | jd d� dd
� W d � n1 ssw Y t �t�� | jd d� dd � W d � d S 1 s�w Y d S )
Nc S rq �Nr r2 r2 r2 r2 r5 rs � rt z1TestMemLeakClass.test_param_err.<locals>.<lambda>r �r� c S rq r� r2 r2 r2 r2 r5 rs � rt ���c S rq r� r2 r2 r2 r2 r5 rs � rt )r� c S rq r� r2 r2 r2 r2 r5 rs � rt )� tolerancec S rq r� r2 r2 r2 r2 r5 rs � rt )r, )r rA rH r� ru r2 r2 r5 �test_param_err~ s � � � � "�z TestMemLeakClass.test_param_errz
skipped on CIr� z skipped during test coveragec
C s� g }|fdd�}z_t jt jjdd��J t�t�� ��& t�t�� �� | j |dd� W d � n1 s4w Y W d � n1 sCw Y W d � W ~d S W d � W ~d S 1 s_w Y W ~d S ~w )Nc S s | � dd � d S )NZ�xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx� ��append��lsr2 r2 r5 r� � � z+TestMemLeakClass.test_leak_mem.<locals>.funz extra-mem��match�d r� )
r rA �fail� Exception�
contextlib�redirect_stdout�io�StringIO�redirect_stderrr� )r= r� r� r2 r2 r5 �
test_leak_mem� s( � � �� �� ��
z TestMemLeakClass.test_leak_memc sd � �fdd�}g � t r
dnd}tjtjjd| d�� ��|� W d � d S 1 s+w Y d S )Nc s" t t�} ��| j� � �| � d S rE )rb rz Z
addCleanupr� r� )rh ��boxr= r2 r5 r� � s z1TestMemLeakClass.test_unclosed_files.<locals>.fun�fd�handlez unclosed r� )r r rA r� r� r� )r= r� r� r2 r� r5 �test_unclosed_files� s "�z$TestMemLeakClass.test_unclosed_filesc s> � fdd�}g � d}| j ||ddd� t� �|d ks J �d S )Nc s � � dd � d S )NZxxxxxxxxxxxxxxxxxxxxxxxxr� r� r2 r� r2 r5 r� � r� z,TestMemLeakClass.test_tolerance.<locals>.funr� r i � )r� r� r� r* )r� r� )r= r� r� r2 r� r5 �test_tolerance� s �z TestMemLeakClass.test_tolerancec C s� dd� }| � t|� t�t�� | � t|� W d � n1 s w Y dd� }t�tjj�� | � t|� W d � d S 1 sBw Y d S )Nc S rJ r/ r2 r2 r2 r2 r5 �fun_1� s z2TestMemLeakClass.test_execute_w_exc.<locals>.fun_1c S s d S rE r2 r2 r2 r2 r5 �fun_2� � z2TestMemLeakClass.test_execute_w_exc.<locals>.fun_2)Z
execute_w_excrB r rA rm r� r� )r= r� r� r2 r2 r5 �test_execute_w_exc� s �"�z#TestMemLeakClass.test_execute_w_excN)rQ rR rS r r� r� r r� r� r r
r� r� r� r r2 r2 r2 r5 r� s s
r� c @ sL e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dS )�TestFakePytestc C s0 t �� }|�|� t jt�� d�}|�|�}|S )N)�stream)�unittestZ TestSuiteZaddTestZTextTestRunnerr� r� �run)r= �klass�suite�runner�resultr2 r2 r5 �run_test_class� s
z TestFakePytest.run_test_classc
C s� t �t�� }dd W d � n1 sw Y t|jt�s!J �t jtdd��}td��1 s0w Y zt jtdd��}td��1 sEw Y W n tyf } zt|�dks[J �W Y d }~d S d }~ww t� d�S )Nr* r r6 r� �barz"foo" does not match "bar"�exception not raised)
r rA rB �
isinstance�valuerH �AssertionError�strr r� )r= �cm�errr2 r2 r5 � test_raises� s
��� ��
zTestFakePytest.test_raisesc C sV t jjdd�dd� �}|� dksJ �t jjdd�G dd� d��}|� �� dks)J �d S )Nr� r� c S rq rr r2 r2 r2 r2 r5 r6 � rG z%TestFakePytest.test_mark.<locals>.foor* c @ � e Zd Zdd� ZdS )z%TestFakePytest.test_mark.<locals>.Fooc S rq rr r2 ru r2 r2 r5 r
� r� z)TestFakePytest.test_mark.<locals>.Foo.barN)rQ rR rS r
r2 r2 r2 r5 �Foo� s r )r r� � xdist_groupr
)r= r6 r r2 r2 r5 � test_mark� s
zTestFakePytest.test_markc C s� G dd� dt j�}| �|d��}|�� sJ �t|j�dks J �|jd d dks*J �G dd� dt j�}| �|d��}|�� s@J �t|j�dksIJ �d S )Nc @ �$ e Zd Zejjddd�dd� �ZdS )�,TestFakePytest.test_skipif.<locals>.TestCaseTr� r� c S � ddksJ �d S rr r2 ru r2 r2 r5 r6 � � �0TestFakePytest.test_skipif.<locals>.TestCase.fooN�rQ rR rS r r� r� r6 r2 r2 r2 r5 �TestCase� � r r6 r* r r� c @ r )r Fr� r� c S r rr r2 ru r2 r2 r5 r6 � r r Nr r2 r2 r2 r5 r � r �r r r Z
wasSuccessfulr� Zskipped�r= r r r2 r2 r5 � test_skipif� s zTestFakePytest.test_skipifc C sX G dd� dt j�}| �|d��}|�� sJ �t|j�dks J �|jd d dks*J �d S )Nc @ r )z*TestFakePytest.test_skip.<locals>.TestCasec S s t �d� ddks J �d S )Nr� r* r )r �skipru r2 r2 r5 r6 � s
z.TestFakePytest.test_skip.<locals>.TestCase.fooN)rQ rR rS r6 r2 r2 r2 r5 r � � r r6 r* r r� r r r2 r2 r5 � test_skip� s
zTestFakePytest.test_skipc C s | j td�}t�|� ttj�|d�d�� W d � n1 s w Y ttj�|d�d��}|�t� d��
� � W d � n1 sDw Y t j �
tjd|��, t�t�� �� t�� }W d � n1 siw Y |�� dksvJ �W d � d S 1 s�w Y d S )N)�dirz __init__.pyr_ z test_file.pyz� import unittest
class TestCase(unittest.TestCase):
def test_passed(self):
pass
r r* )ra r rY r� rb rd r� rc �textwrap�dedent�lstripr rT �objectr\ r� r� r� r� r� r �mainZcountTestCases)r= Ztmpdirrh r r2 r2 r5 � test_main s
� � �
�"�zTestFakePytest.test_mainc C s6 t �t�� tjdtdd� W d � n1 sw Y z t �t�� tjdtdd� W d � n1 s6w Y W n tyE Y nw t�d�S t jtdd�� tjdtdd� W d � n1 sew Y z!t jtdd�� tjdtdd� W d � n1 s�w Y W n
ty� Y d S w t�d�S )Nr6 r* )�
stacklevelr r� r
) r Zwarns� UserWarning�warnings�warn�DeprecationWarningr r r� ru r2 r2 r5 �
test_warns s0 � �� �
� �� �
zTestFakePytest.test_warnsc C s. t �t jj�� t �d��1 sw Y d S )Nr� )r rA r� r� ru r2 r2 r5 � test_fail5 s
�zTestFakePytest.test_failN) rQ rR rS r r r r! r$ r+ r1 r2 r2 r2 r2 r5 r � s r c @ s e Zd Zdd� Zdd� ZdS )�TestTestingUtilsc C sL t �� }t|�}|�� tdd� |�|j�D ��d }|� |�� ks$J �d S )Nc s � � | ] }|d dkr|V qdS )r* r� Nr2 ��.0�xr2 r2 r5 � <genexpr>? � � z:TestTestingUtils.test_process_namespace.<locals>.<genexpr>r )r\ r� r �test�nextr] �gettersr� )r= r� �nsr� r2 r2 r5 �test_process_namespace; s
z'TestTestingUtils.test_process_namespacec C s: t � }tdd� |�|j�D ��d }|� t�� ksJ �d S )Nc s r4 )r* � net_if_addrsNr2 r5 r2 r2 r5 r8 D r9 z9TestTestingUtils.test_system_namespace.<locals>.<genexpr>r )r" r; r] r< r\ r? )r= r= r� r2 r2 r5 �test_system_namespaceB s z&TestTestingUtils.test_system_namespaceN)rQ rR rS r> r@ r2 r2 r2 r5 r3 : s r3 c @ r )�TestOtherUtilsc C s. t t�dd�ddd��sJ �t t� �rJ �d S )Nr6 za b cr* r� r8 )r r� �
namedtuple�tupleru r2 r2 r5 �test_is_namedtupleI s z!TestOtherUtils.test_is_namedtupleN)rQ rR rS rD r2 r2 r2 r5 rA H r# rA )?�__doc__r� r� r� r� rY r� r� r� r& r r. r r\ Z psutil.testsr r r Zpsutil._commonr r r r r
r r r
r r r r r r r r r r r r r r r r r r r! r" r# r$ r% r& r' r( rU rw r� r� r� r r� r r3 rA r2 r2 r2 r5 �<module> sz E(-L KNx