관리-도구
편집 파일: test_connections.cpython-37.pyc
B �=�_�S � @ sr d Z ddlZddlZddlZddlZddlmZ ddlmZ ddlmZ ddlm Z ddlm Z ddl Z ddl m Z dd l m Z dd l mZ dd l mZ dd l mZ dd l mZ ddl mZ ddl mZ ddlmZ ddlmZ ddlmZ ddlmZ ddlmZ 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( e �)� Z*e+ed"e,� �Z-ej.dd#� d$kZ/e"G d%d&� d&e ��Z0G d'd(� d(e0�Z1e"G d)d*� d*e0��Z2e"G d+d,� d,e0��Z3G d-d.� d.e0�Z4e&�5e$d/�G d0d1� d1e0��Z6G d2d3� d3e �Z7e8d4k�rndd5l9m:Z: e:e;� dS )6z;Tests for net_connections() and Process.connections() APIs.� N)�closing)�AF_INET)�AF_INET6)� SOCK_DGRAM)� SOCK_STREAM)�FREEBSD)�LINUX)�MACOS)�NETBSD)�OPENBSD)�POSIX)�SUNOS)�WINDOWS)� supports_ipv6)�PY3)�AF_UNIX)� bind_socket)�bind_unix_socket)�check_connection_ntuple)�create_sockets)�HAS_CONNECTIONS_UNIX)�PsutilTestCase)� reap_children)�retry_on_failure)� serialrun)�skip_on_access_denied)� SKIP_SYSCONS)�tcp_socketpair)�unittest)�unix_socketpair)� wait_for_file�SOCK_SEQPACKET� )� � c @ s& e Zd Zdd� Zdd� Zd dd�ZdS ) �ConnectionTestCasec C s$ t s ts tjdd�}|r t|��d S )N�all)�kind)r r �thisproc� connections�AssertionError)�self�cons� r- �P/opt/alt/python37/lib64/python3.7/site-packages/psutil/tests/test_connections.py�setUp8 s zConnectionTestCase.setUpc C s$ t s ts tjdd�}|r t|��d S )Nr& )r' )r r r( r) r* )r+ r, r- r- r. �tearDown> s zConnectionTestCase.tearDownr& c sd yt j|d�}W n t jk r0 tr*dS � Y nX � fdd�|D �}|�� |�� | �||� dS )z�Given a process PID and its list of connections compare those against system-wide connections retrieved via psutil.net_connections. )r' Nc s" g | ]}|j � kr|d d� �qS )N���)�pid)�.0�c)r2 r- r. � <listcomp>T s zBConnectionTestCase.compare_procsys_connections.<locals>.<listcomp>)�psutil�net_connections� AccessDeniedr �sort� assertEqual)r+ r2 Z proc_consr' Zsys_consr- )r2 r. �compare_procsys_connectionsE s z.ConnectionTestCase.compare_procsys_connectionsN)r& )�__name__� __module__� __qualname__r/ r0 r; r- r- r- r. r% 5 s r% c @ s0 e Zd Ze�ed�dd� �Zdd� Zdd� ZdS ) �TestBasicOperationsz requires rootc C s4 t � �$ x tjdd�D ] }t|� qW W d Q R X d S )Nr& )r' )r r6 r7 r )r+ �connr- r- r. � test_system\ s z TestBasicOperations.test_systemc C s8 t � �( x t�� jdd�D ] }t|� qW W d Q R X d S )Nr& )r' )r r6 �Processr) r )r+ r@ r- r- r. � test_processb s z TestBasicOperations.test_processc C s( | j ttjdd� | j ttjdd� d S )Nz???)r' )Z assertRaises� ValueErrorr( r) r6 r7 )r+ r- r- r. �test_invalid_kindg s z%TestBasicOperations.test_invalid_kindN) r< r= r> r �skipIfr rA rC rE r- r- r- r. r? Z s r? c @ s� e Zd ZdZdd� Zdd� Zdd� Ze�e � d�d d � �Z d d � Z e�e � d�d d� �Z e�e d�dd� �Ze�e d�dd� �ZdS )�TestUnconnectedSocketsz;Tests sockets which are open but not connected to anything.c C sv t jdd�}tdd� |D ��}ts&tr2||�� S | �t|�d� |d jdkrj| �||�� j|�� � |d S d S )Nr& )r' c S s g | ]}|j |f�qS r- )�fd)r3 r4 r- r- r. r5 r s z=TestUnconnectedSockets.get_conn_from_sock.<locals>.<listcomp>� r r1 ) r( r) �dictr r �filenor: �lenrH )r+ �sockr, Zsmapr- r- r. �get_conn_from_sockp s z)TestUnconnectedSockets.get_conn_from_sockc C s� | � |�}t|� |jdkr.| �|j|�� � | �|j|j� | �|j|�tj tj �� |� � }|szt rzt |t�rz|�� }|jtkr�|dd� }|jtkr�tr�n| �|j|� |jtkr�tr�tjdd�}| jt�� |dd� |S )z�Given a socket, makes sure it matches the one obtained via psutil. It assumes this process created one connection only (the one supposed to be checked). r1 Nr" r& )r' )rN r rH r: rK �family�type� getsockopt�socket� SOL_SOCKET�SO_TYPE� getsocknamer � isinstance�bytes�decoder r r �laddrr r( r) r; �os�getpid)r+ rM r@ rY r, r- r- r. � check_socket} s& z#TestUnconnectedSockets.check_socketc C sJ d}t ttt|d���*}| �|�}|jr,t�| �|jt j � W d Q R X d S )N)z 127.0.0.1r )�addr) r r r r r\ �raddrr* r: �statusr6 � CONN_LISTEN)r+ r] rM r@ r- r- r. � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedc C sJ d}t ttt|d���*}| �|�}|jr,t�| �|jt j � W d Q R X d S )N)z::1r )r] ) r r r r r\ r^ r* r: r_ r6 r` )r+ r] rM r@ r- r- r. � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sJ d}t ttt|d���*}| �|�}|jr,t�| �|jt j � W d Q R X d S )N)z 127.0.0.1r )r] ) r r r r r\ r^ r* r: r_ r6 � CONN_NONE)r+ r] rM r@ r- r- r. � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sJ d}t ttt|d���*}| �|�}|jr,t�| �|jt j � W d Q R X d S )N)z::1r )r] ) r r r r r\ r^ r* r: r_ r6 rc )r+ r] rM r@ r- r- r. � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6z POSIX onlyc C sL | � � }tt|td���*}| �|�}|jr.t�| �|jt j � W d Q R X d S )N)rP ) � get_testfnr r r r\ r^ r* r: r_ r6 rc )r+ �testfnrM r@ r- r- r. � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sL | � � }tt|td���*}| �|�}|jr.t�| �|jt j � W d Q R X d S )N)rP ) rf r r r r\ r^ r* r: r_ r6 rc )r+ rg rM r@ r- r- r. � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)r< r= r> �__doc__rN r\ ra r rF r rb rd re r rh ri r- r- r- r. rG l s #rG c @ s: e Zd ZdZe�ed�dd� �Ze�e d�dd� �Z dS ) �TestConnectedSocketzJTest socket pairs which are are actually connected to each other. zunreliable on SUONSc C s� d}t jdd�rt�tt|d�\}}zHt jdd�}| �t|�d� | �|d jtj � | �|d jtj � W d |� � |� � X d S )N)z 127.0.0.1r �tcp4)r' )r] r" r rI ) r( r) r* r r r: rL r_ r6 �CONN_ESTABLISHED�close)r+ r] �server�clientr, r- r- r. �test_tcp� s z TestConnectedSocket.test_tcpz POSIX onlyc C sd | � � }t|�\}}�z6tjdd�}|d jr<|d jr<t�|d jrT|d jrTt�ts\trjdd� |D �}| j t |�d|d� t s�ts�t r�| � |d jd � | � |d jd � | � ||d jp�|d j� n~t �rxv|d j|d j|d j|d jfD ]}| � |d � q�W n<| � |d j�p&|d j|� | � |d j�pD|d j|� W d |�� |�� X d S ) N�unix)r' r rI c S s g | ]}|j d kr|�qS )z /var/run/log)r^ )r3 r4 r- r- r. r5 � s z1TestConnectedSocket.test_unix.<locals>.<listcomp>r" )�msg� )rf r r( r) rY r^ r* r r r: rL r r r rn )r+ rg ro rp r, r] r- r- r. � test_unix� s* "z TestConnectedSocket.test_unixN) r< r= r> rj r rF r rq r ru r- r- r- r. rk � s rk c @ s. e Zd Zdd� Zeed�dd� �Zdd� ZdS ) � TestFiltersc s� � fdd�}t � �� |dtttgtttg� |dttgttg� |dtgttg� |dttgtg� |dtgtg� |dtgtg� |d ttgtg� |d tgtg� |d tgtg� tr�|d tgtttg� W d Q R X d S ) Nc sl x0t j| d�D ] }� �|j|� � �|j|� qW tshx0tj| d�D ] }� �|j|� � �|j|� qDW d S )N)r' )r( r) �assertInrO rP r r6 r7 )r' �families�typesr@ )r+ r- r. �check s z'TestFilters.test_filters.<locals>.checkr& �inet�inet4�tcprl �tcp6�udp�udp4�udp6rr )r r r r r r r! r )r+ rz r- )r+ r. � test_filters sB zTestFilters.test_filters)Zonly_ifc s� t � � fdd�}t�d�}t�d�}tj�� jt�� d��}|jt t �d|d�}|jt t �d|d�}|jt t �d|d�}|jt t �d|d�}� � |�} t t|d d ��} � � |�} t t|d d ��} t� r�� � |�} t t|d d ��}� � |�}t t|d d ��}nd } d }d }d }x�t�� D ]�}|�� }� �t|�d � x�|D ]�}|j| jk�rf|||t t| d tjd � n�|j| jk�r�|||t t| d tjd� nZ|jt| dd �k�r�|||t t|d tjd� n,|jt|dd �k�r:|||t t|d tjd� �q:W �qW d S )Nc s� d}t |� � �|j|� � �|j|� � �|j|� � �|j|� � �|j|� x6|D ].} | j| d�} | |krz| s�t�qX| rXt| ��qXW t r�� � | j |g� d S )N) r& r{ r| �inet6r} rl r~ r r� r� )r' ) r r: rO rP rY r^ r_ r) r* r r; r2 ) �procr@ rO rP rY r^ r_ ZkindsZ all_kindsr' r, )r+ r- r. � check_conn> s z+TestFilters.test_combos.<locals>.check_conna import socket, time s = socket.socket({family}, socket.SOCK_STREAM) s.bind(('{addr}', 0)) s.listen(5) with open('{testfn}', 'w') as f: f.write(str(s.getsockname()[:2])) time.sleep(60) a import socket, time s = socket.socket({family}, socket.SOCK_DGRAM) s.bind(('{addr}', 0)) with open('{testfn}', 'w') as f: f.write(str(s.getsockname()[:2])) time.sleep(60) )�dirz 127.0.0.1)rO r] rg z::1T)�deleterI r- )r& r{ r| r} rl )r& r{ r| r r� r2 )r& r{ r� r} r~ )r& r{ r� r r� ) r �textwrap�dedentrZ �path�basenamerf �getcwd�format�intr r �pyrun�evalr r r( �childrenr) r: rL r2 r r6 r` r rc �getattr)r+ r� Z tcp_templateZ udp_templateZtestfileZ tcp4_templateZ udp4_templateZ tcp6_templateZ udp6_templateZ tcp4_procZ tcp4_addrZ udp4_procZ udp4_addrZ tcp6_procZ tcp6_addrZ udp6_procZ udp6_addr�pr, r@ r- )r+ r. � test_combos: s` zTestFilters.test_combosc C s� t � ��� tjdd�}| �t|�t� r*dnd� x,|D ]$}| �|jtt f� | �|j t � q6W tjdd�}| �t|�d� | �|d jt� | �|d j t � t� r�tjdd�}| �t|�d� | �|d jt � | �|d j t � tjdd�}| �t|�t� �rdnd� x.|D ]&}| �|jtt f� | �|j t � �qW tjd d�}| �t|�d� | �|d jt� | �|d j t � t� �r�tjd d�}| �t|�d� | �|d jt � | �|d j t � tjd d�}| �t|�t� �r�d nd� x2|D ]*}| �|jtt f� | �|j t t f� �q�W t� �rttjd d�}| �t|�d� x.|D ]&}| �|jt � | �|j t t f� �qJW t �r�t�s�t�s�tjdd�}| �t|�d� x.|D ]&}| �|jt� | �|j t t f� �q�W W d Q R X d S )Nr} )r' r" rI rl r r~ r r� r� r{ � r� rr r# )r r( r) r: rL r rw rO r r rP r r r r r r )r+ r, r@ r- r- r. � test_count� s\ zTestFilters.test_countN)r< r= r> r� r r r� r� r- r- r- r. rv s +arv z requires rootc @ s& e Zd ZdZdd� Ze� dd� �ZdS )�TestSystemWideConnectionsz Tests for net_connections().c s� � fdd�}t � �l ddlm} xX|�� D ]L\}}|dkr@ts@q*|\}}t�|�}� �t|�tt |��� ||||� q*W W d Q R X d S )Nc sH xB| D ]:}� j |j||d� |jtkr8� j |j||d� t|� qW d S )N)rs )rw rO r rP r )r, rx �types_r@ )r+ r- r. rz � s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmaprr ) r �psutil._commonr� �itemsr r6 r7 r: rL �set)r+ rz r� r' �groupsrx r� r, r- )r+ r. �test_it� s z!TestSystemWideConnections.test_itc s� t � �}t|�}W d Q R X g �d}g }xFt|�D ]:}| �� }|�|� t�d| �}| �|�}��|j� q0W x|D ] }t |� qtW �fdd�t j dd�D �} xJ�D ]B� | � t� fdd�| D ��|� t � � �} | � t| �d��|� q�W d S )N� a import time, os from psutil.tests import create_sockets with create_sockets(): with open(r'%s', 'w') as f: f.write("hello") time.sleep(60) c s g | ]}|j � kr|�qS r- )r2 )r3 �x)�pidsr- r. r5 s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r& )r' c s g | ]}|j � kr|�qS r- )r2 )r3 r� )r2 r- r. r5 s )r rL �rangerf �appendr� r� r� r2 r r6 r7 r: rB r) ) r+ ZsocksZexpected�times�fnames�i�fname�srcZsprocZsysconsr� r- )r2 r� r. �test_multi_sockets_procs� s( z2TestSystemWideConnections.test_multi_sockets_procsN)r< r= r> rj r� r r� r- r- r- r. r� � s r� c @ s e Zd Zdd� ZdS )�TestMiscc C s� g }g }xht t�D ]\}|�d�rtt|�}t|�}|�� sBt|��| �t|� | �||� |�|� |�|� qW t r�tj tj t r�tj d S )NZCONN_)r� r6 � startswithr� �str�isupperr* Z assertNotInr� r Z CONN_IDLEZ CONN_BOUNDr ZCONN_DELETE_TCB)r+ �ints�strs�name�numZstr_r- r- r. �test_connection_constants s z"TestMisc.test_connection_constantsN)r< r= r> r� r- r- r- r. r� s r� �__main__)� run_from_name)<rj rZ rR �sysr� � contextlibr r r r r r6 r r r r r r r r r� r Zpsutil._compatr Z psutil.testsr r r r r r r r r r r r r r r r rB r( r� �objectr! � version_infoZ PYTHON_39r% r? rG rk rv rF r� r� r< Zpsutil.tests.runnerr� �__file__r- r- r- r. �<module> sl $ b = J>