관리-도구
편집 파일: test_testutils.cpython-38.pyc
U �Mg�H � @ s� d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl m Z ddl mZ ddl mZ ddlmZ ddlmZ ddlmZ dd lmZ dd l mZ dd l mZ dd l mZ dd l mZ ddl mZ ddl mZ ddl m Z ddl m Z ddl m Z ddl m Z ddl m Z ddl m!Z! ddl m"Z" ddl m#Z# ddl m$Z$ ddl m%Z% ddl m&Z& ddl m'Z' dd l m(Z( dd l m)Z) dd l m*Z* dd l m+Z+ dd l m,Z, dd!l m-Z- dd"l m.Z. dd#l m/Z/ dd$l m0Z0 dd%l m1Z1 dd&l m2Z2 dd'l m3Z3 dd(l m4Z4 G d)d*� d*e �Z5G d+d,� d,e �Z6G d-d.� d.e �Z7G d/d0� d0e �Z8G d1d2� d2e �Z9e)j:j;d3d4�G d5d6� d6e ��Z<G d7d8� d8e �Z=G d9d:� d:e �Z>G d;d<� d<e �Z?dS )=z1Tests for testing utils (psutil.tests namespace).� N)�FREEBSD)�NETBSD)�POSIX)� open_binary)� open_text)� supports_ipv6)�PY3)� 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)�mock)�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��� |� dks4t�|jdksBt�d S )N� � ��retries�intervalZlogfunc s � r� � � dd q dS �Nr+ r ��pop� ��queuer2 �K/opt/hc_python/lib64/python3.8/site-packages/psutil/tests/test_testutils.py�fooI s z2TestRetryDecorator.test_retry_success.<locals>.foo� )r �list�range�AssertionError� call_count��self�sleepr6 r2 r3 r5 �test_retry_successE s z%TestRetryDecorator.test_retry_successc sT t ddd d�� fdd��}ttd��� t�t�� |� W 5 Q R X |jdksPt�d S )Nr* r+ r, c s � r� � � dd q dS r/ r0 r2 r3 r2 r5 r6 W s z2TestRetryDecorator.test_retry_failure.<locals>.foo� )r r8 r9 r �raises�ZeroDivisionErrorr; r: r< r2 r3 r5 �test_retry_failureT s z%TestRetryDecorator.test_retry_failurec C sB t tdd�dd� �}t�t�� |� W 5 Q R X |jdks>t�d S )Nr+ )� exceptionr. c S s t �d S �N)� TypeErrorr2 r2 r2 r5 r6 e s z2TestRetryDecorator.test_exception_arg.<locals>.foor )r � ValueErrorr rA rF r; r: r< r2 r2 r5 �test_exception_argc s z%TestRetryDecorator.test_exception_argc C sD t dd d d�dd� �}t�t�� |� W 5 Q R X |jdks@t�d S )Nr* r, c S s dd d S r/ r2 r2 r2 r2 r5 r6 q s z4TestRetryDecorator.test_no_interval_arg.<locals>.foor �r r rA rB r; r: r< r2 r2 r5 �test_no_interval_argm s z'TestRetryDecorator.test_no_interval_argc C sD t ddd d�dd� �}t�t�� |� W 5 Q R X |jdks@t�d S )Nr* r+ r, c S s dd d S r/ r2 r2 r2 r2 r5 r6 { s z0TestRetryDecorator.test_retries_arg.<locals>.foorI r< r2 r2 r5 �test_retries_argy s z#TestRetryDecorator.test_retries_argc C s& t �t�� tddd� W 5 Q R X d S )Nr* r+ )r- �timeout)r rA rG r )r= r> r2 r2 r5 � test_retries_and_timeout_args� s z0TestRetryDecorator.test_retries_and_timeout_argsN) �__name__� __module__� __qualname__r �patchr? rC rH rJ rK rM r2 r2 r2 r5 r) D 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 5 Q R X W 5 Q R X d S )Ni�� �psutil.tests.retry.__iter__r �Z return_value) r( �os�getpid�max�psutilZpidsr rQ �iterr rA Z NoSuchProcess)r= Znopidr2 r2 r5 �test_wait_for_pid� s z#TestSyncTestUtils.test_wait_for_pidc C sD | � � }t|d��}|�d� W 5 Q R X t|� tj�|�r@t�d S )N�wr6 �� get_testfn�open�writer' rU �path�existsr: �r= �testfn�fr2 r2 r5 �test_wait_for_file� s z$TestSyncTestUtils.test_wait_for_filec C s> | � � }t|d�� W 5 Q R X t|dd� tj�|�r:t�d S )Nr[ T)�empty)r] r^ r' rU r` ra r: �r= rc r2 r2 r5 �test_wait_for_file_empty� s z*TestSyncTestUtils.test_wait_for_file_emptyc C sJ | � � }tjdtdg�d��$ t�t�� t|� W 5 Q R X W 5 Q R X d S )NrS r rT )r] r rQ rY r rA �IOErrorr' rg r2 r2 r5 �test_wait_for_file_no_file� s z,TestSyncTestUtils.test_wait_for_file_no_filec C sH | � � }t|d��}|�d� W 5 Q R X t|dd� tj�|�sDt�d S )Nr[ r6 F)�deleter\ rb 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 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) rN rO rP rZ re rh rj rl rq r2 r2 r2 r5 rR � s rR c @ s4 e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd S ) �TestFSTestUtilsc C s& t t��}|jdkst�W 5 Q R X d S )N�r)r �__file__�moder: �r= rd r2 r2 r5 �test_open_text� s z TestFSTestUtils.test_open_textc C s& t t��}|jdkst�W 5 Q R X d S )N�rb)r rt ru r: rv r2 r2 r5 �test_open_binary� s z TestFSTestUtils.test_open_binaryc C s< | � � }t|� tj�|�s t�t|� tj�|�s8t�d S rE )r] r! rU r` �isdirr: rg r2 r2 r5 �test_safe_mkdir� s z TestFSTestUtils.test_safe_mkdirc C s� | � � }t|d��� t|� tj�|�r.t�t|� t�|� t|� tj�|�rXt�t j dt t j d�d��.}t�t �� t|� W 5 Q R X |js�t�W 5 Q R X d S )Nr[ zpsutil.tests.os.stat� )Z side_effect)r] r^ �closer" rU r` ra r: �mkdirr rQ �OSError�errno�EINVALr rA Zcalled)r= rc �mr2 r2 r5 �test_safe_rmpath� s � z TestFSTestUtils.test_safe_rmpathc C s\ | � � }t�� }t�|� t|�� t�� tj�||�ks>t�W 5 Q R X t�� |ksXt�d S rE )r] rU �getcwdr~ r r` �joinr: )r= rc �baser2 r2 r5 � test_chdir� s $zTestFSTestUtils.test_chdirN)rN rO rP rw ry r{ r� r� r2 r2 r2 r5 rr � s rr c @ s> e Zd Zdd� Zdd� Zejje dd�dd� �Z d d � Z d S ) �TestProcessUtilsc C sN | � � }t�|j�}|�� s t�t� |�� r2t�tjjr>t�tjj rJt�d S rE ) �spawn_testprocrX �Process�pid� is_runningr: r �testsZ _pids_startedZ_subprocesses_started)r= Zsubp�pr2 r2 r5 �test_reap_children� s z#TestProcessUtils.test_reap_childrenc C s� | � � \}}|j|jks t�|�� s(t�|�� s4t�t�� �� }||gksNt�t�� jdd�}t|�dksnt�||kszt�||ks�t�|�� t � � ks�t�|�� |jks�t�t |� |�� r�t�|�� s�t�t |� |�� r�t�d S )NT)� recursive� ) Zspawn_children_pairr� r: r� rX r� �children�len�ppidrU rV r% )r= �childZ grandchildr� r2 r2 r5 �test_spawn_children_pair� s" z)TestProcessUtils.test_spawn_children_pair� POSIX only��reasonc C s" | � � \}}|�� tjks t�d S rE )� spawn_zombie�statusrX Z STATUS_ZOMBIEr: )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 r�| � � \}}t|� t|� | �|j� | �|j� d S )Nz-cz3import time; [time.sleep(0.1) for x in range(100)];)�stdout�stderr�env) r� r% Z assertPidGoner� rX r� r �Popen� subprocess�PIPEr r r� )r= r� �cmdr� �parentr� r2 r2 r5 �test_terminate s> �� z TestProcessUtils.test_terminateN) rN rO rP r� r� r �mark�skipifr r� r� r2 r2 r2 r5 r� � s r� c @ sl e Zd Zdd� Zejje dd�dd� �Zdd� Z ejje dd�ejje pNe d d�d d � ��Z d d � Z dS )� TestNetUtilsc C s> t � }t�td|fd���}|�� d |ks0t�W 5 Q R X d S )Nr| ��addrr+ )r � contextlib�closingr � getsocknamer: )r= �port�sr2 r2 r5 r + s zTestNetUtils.bind_socketr� r� c C s� | � � }t|�}t�|��\ |jtjks,t�|jtj ks<t�|� � |ksLt�t j � |�s\t�t�t �|�j�srt�W 5 Q R X | � � }t|tjd�}t�|�� |jtjks�t�W 5 Q R X d S )N)�type)r] r r� r� �family�socket�AF_UNIXr: r� � SOCK_STREAMr� rU r` ra �stat�S_ISSOCK�st_mode� SOCK_DGRAM)r= �name�sockr2 r2 r5 �test_bind_unix_socket0 s z"TestNetUtils.test_bind_unix_socketc C s| dt � f}ttj|d�\}}t�|��L t�|��6 |�� |ksDt�|�� |ksTt�|�� |ksdt�W 5 Q R X W 5 Q R X d S )Nz 127.0.0.1r� ) r r$ r� �AF_INETr� r� r� r: � getpeername)r= r� �server�clientr2 r2 r5 �tcp_tcp_socketpair@ s z TestNetUtils.tcp_tcp_socketpairz*/var/run/log UNIX socket opened by defaultc C s� t �� }|�� }t|jdd��g ks(t�| �� }t|�\}}zzt j � |�sNt�t � t � |�j�sdt�|�� | dksxt�tt|jdd���dks�t�|�� |ks�t�|�� |ks�t�W 5 |�� |�� X d S )N�unix)�kindr� )rX r� �num_fdsr Znet_connectionsr: r] r& r} rU r` ra r� r� r� r� r� r� )r= r� r� r� r� r� r2 r2 r5 �test_unix_socketpairK s( � ���z!TestNetUtils.test_unix_socketpairc C s� t � ��}t�t�}t�t�}|D ]2}||j d7 < ||�tjtj� d7 < q |tj dksft �t � r~|tj dks~t �t r�tr�|tj dks�t �|tj dks�t �|tj dks�t �W 5 Q R X d S )Nr+ r� )r � collections� defaultdict�intr� � getsockoptr� � SOL_SOCKET�SO_TYPEr� r: r �AF_INET6r r r� r� r� )r= �socksZfams�typesr� r2 r2 r5 �test_create_socketsg s z TestNetUtils.test_create_socketsN)rN rO rP 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 dks4t�d S ) Nc s � d d7 < d S )N�cntr+ r2 r2 �r� r2 r5 �fun| s z(TestMemLeakClass.test_times.<locals>.funr� r � � )�times� warmup_times� )�executer: )r= r� r2 r� r5 � test_timesz s zTestMemLeakClass.test_timesc C s� t �t�� | jdd� dd� W 5 Q R X t �t�� | jdd� dd� W 5 Q R X t �t�� | jdd� dd� W 5 Q R X t �t�� | jd d� dd � W 5 Q R X t �t�� | jd d� dd � W 5 Q R X d S ) Nc S s dS �Nr r2 r2 r2 r2 r5 rn � ro z1TestMemLeakClass.test_param_err.<locals>.<lambda>r �r� c S s dS r� r2 r2 r2 r2 r5 rn � ro ���c S s dS r� r2 r2 r2 r2 r5 rn � ro )r� c S s dS r� r2 r2 r2 r2 r5 rn � ro )� tolerancec S s dS r� r2 r2 r2 r2 r5 rn � ro )r- )r rA rG r� rp r2 r2 r5 �test_param_err� s z TestMemLeakClass.test_param_errz skipped on CIr� z skipped during test coveragec C sF g }|fdd�}z,t jtdd�� | j|dd� W 5 Q R X W 5 ~X d S )Nc S s | � dd � d S )NZ�xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx� ��append��lsr2 r2 r5 r� � s z+TestMemLeakClass.test_leak_mem.<locals>.funz extra-mem��match�d r� )r rA r: r� )r= r� r� r2 r2 r5 � test_leak_mem� s z TestMemLeakClass.test_leak_memc sJ � �fdd�}g � t rdnd}tjtd| d�� ��|� W 5 Q R X d S )Nc s" t t�} ��| j� � �| � d S rE )r^ rt Z addCleanupr} r� )rd ��boxr= r2 r5 r� � s z1TestMemLeakClass.test_unclosed_files.<locals>.fun�fd�handlez unclosed r� )r r rA 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:t�d S )Nc s � � dd � d S )NZxxxxxxxxxxxxxxxxxxxxxxxxr� r� r2 r� r2 r5 r� � s z,TestMemLeakClass.test_tolerance.<locals>.funr� r i � )r� r� r� r+ )r� r� r: )r= r� r� r2 r� r5 �test_tolerance� s �z TestMemLeakClass.test_tolerancec C sd dd� }| � t|� t�t�� | � t|� W 5 Q R X dd� }t�t�� | � t|� W 5 Q R X d S )Nc S s dd d S 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� s z2TestMemLeakClass.test_execute_w_exc.<locals>.fun_2)Z execute_w_excrB r rA r r: )r= r� r� r2 r2 r5 �test_execute_w_exc� s z#TestMemLeakClass.test_execute_w_excN)rN rO rP r r� r� r r� r� r r r� r� r� r� r2 r2 r2 r5 r� x s r� c @ sV e Zd Zdd� Zdd� Zdd� Zdd� Zejj e d d �d d � �Z d d� Z dd� Z dS )�TestFakePytestc C s( t �� }|�|� t �� }|�|�}|S rE )�unittestZ TestSuiteZaddTestZTextTestRunner�run)r= �klass�suite�runner�resultr2 r2 r5 �run_test_class� s z TestFakePytest.run_test_classc C s� t �t��}dd W 5 Q R X t|jt�s.t�t jtdd��}td��W 5 Q R X z&t jtdd��}td��W 5 Q R X W n2 tk r� } zt|�dks�t�W 5 d }~X Y n X | �d��d S )Nr+ r r6 r� �barz"foo" does not match "bar"�exception not raised) r rA rB � isinstance�valuer: rG �str�fail)r= �cm�errr2 r2 r5 � test_raises� s "zTestFakePytest.test_raisesc C sV t jjdd�dd� �}|� dks$t�t jjdd�G dd� d��}|� �� dksRt�d S )Nr� r� c S s dS rm r2 r2 r2 r2 r5 r6 � s z%TestFakePytest.test_mark.<locals>.foor+ c @ s e Zd Zdd� ZdS )z%TestFakePytest.test_mark.<locals>.Fooc S s dS rm r2 rp r2 r2 r5 r� � s z)TestFakePytest.test_mark.<locals>.Foo.barN)rN rO rP r� r2 r2 r2 r5 �Foo� s r )r r� � xdist_groupr: r� )r= r6 r r2 r2 r5 � test_mark� s zTestFakePytest.test_markc C s� G dd� dt j�}| �|d��}|�� s,t�t|j�dks>t�|jd d dksTt�G dd� dt j�}| �|d��}|�� s�t�t|j�dks�t�d S )Nc @ s$ e Zd Zejjddd�dd� �ZdS )�,TestFakePytest.test_skipif.<locals>.TestCaseTr� r� c S s ddks t �d S rm �r: rp r2 r2 r5 r6 � s �0TestFakePytest.test_skipif.<locals>.TestCase.fooN�rN rO rP r r� r� r6 r2 r2 r2 r5 �TestCase� s r r6 r+ r r� c @ s$ e Zd Zejjddd�dd� �ZdS )r Fr� r� c S s ddks t �d S rm r rp r2 r2 r5 r6 � s r Nr r2 r2 r2 r5 r � s �r� r r� Z wasSuccessfulr: r� Zskipped�r= r r� r2 r2 r5 � test_skipif� s zTestFakePytest.test_skipifznot PY3r� c C sX G dd� dt j�}| �|d��}|�� s,t�t|j�dks>t�|jd d dksTt�d S )Nc @ s e Zd Zdd� ZdS )z*TestFakePytest.test_skip.<locals>.TestCasec S s t �d� ddkst�d S )Nr� r+ r )r �skipr: rp r2 r2 r5 r6 s z.TestFakePytest.test_skip.<locals>.TestCase.fooN)rN rO rP r6 r2 r2 r2 r5 r s 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 5 Q R X ttj�|d�d��}|�t� d�� � � W 5 Q R X t j � tjd|��6 | �td�� t�� }|�� dks�t�W 5 Q R X W 5 Q R X d S ) N)�dirz __init__.pyr[ z test_file.pyz� import unittest class TestCase(unittest.TestCase): def test_passed(self): pass r zFake pytest module was usedr+ )r] r rU r~ r^ r` r� r_ �textwrap�dedent�lstripr rQ �objectrX r� ZassertWarnsRegex� UserWarningr �mainZcountTestCasesr: )r= Ztmpdirrd r� r2 r2 r5 � test_main s �zTestFakePytest.test_mainc C s� t �t�� tjdtdd� W 5 Q R X z*t �t�� tjdtdd� W 5 Q R X W n tk rd Y n X | �d��t jtdd�� tjdtdd� W 5 Q R X z.t jtdd�� tjdtdd� W 5 Q R X W n tk r� Y n X | �d��d S )Nr6 r+ )� stacklevelr� r� r� )r Zwarnsr �warnings�warn�DeprecationWarningr: r rp r2 r2 r5 � test_warns s zTestFakePytest.test_warnsN)rN rO rP r� r r r r r� r� r r r r r2 r2 r2 r5 r� � s r� c @ s e Zd Zdd� Zdd� ZdS )�TestTestingUtilsc C sL t �� }t|�}|�� dd� |�|j�D �d d }|� |�� ksHt�d S )Nc S s g | ]}|d dkr|�qS )r+ r� r2 ��.0�xr2 r2 r5 � <listcomp>? s z;TestTestingUtils.test_process_namespace.<locals>.<listcomp>r )rX r� r �testrY �gettersr� r: )r= r� �nsr� r2 r2 r5 �test_process_namespace; s z'TestTestingUtils.test_process_namespacec C s: t � }dd� |�|j�D �d d }|� t�� ks6t�d S )Nc S s g | ]}|d dkr|�qS )r+ � net_if_addrsr2 r r2 r2 r5 r# D s z:TestTestingUtils.test_system_namespace.<locals>.<listcomp>r )r# rY r% rX r( r: )r= r& r� r2 r2 r5 �test_system_namespaceB s z&TestTestingUtils.test_system_namespaceN)rN rO rP r' r) r2 r2 r2 r5 r : s r c @ s e Zd Zdd� ZdS )�TestOtherUtilsc C s. t t�dd�ddd��s t�t t� �r*t�d S )Nr6 za b cr+ r� r7 )r r� � namedtupler: �tuplerp r2 r2 r5 �test_is_namedtupleI s z!TestOtherUtils.test_is_namedtupleN)rN rO rP r- r2 r2 r2 r5 r* H s r* )@�__doc__r� r� r� rU r� r� r� r r� r rX Z psutil.testsr r r Zpsutil._commonr r r Zpsutil._compatr 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) rR rr r� r� r� r r� r� r r* r2 r2 r2 r5 �<module> sx E(-LN Kv