관리-도구
편집 파일: test_connections.cpython-38.pyc
U �MgS � @ s\ d Z ddlZddlZddlZddlmZ ddlmZ ddlmZ ddlmZ ddlm Z ddl Z ddl m Z dd l m Z dd l m Z dd l mZ dd l mZ dd l mZ ddl mZ ddl mZ ddlmZ ddlmZ ddlmZ ddlmZ ddlmZ ddlmZ 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(ed"e)� �Z*d#d$� Z+e!j,j-d%d&�G d'd(� d(e��Z.G d)d*� d*e.�Z/e!j,j-d%d&�G d+d,� d,e.��Z0e!j,j-d%d&�G d-d.� d.e.��Z1G d/d0� d0e.�Z2e!j,j3ed1d2�G d3d4� d4e.��Z4G d5d6� d6e�Z5dS )7zFTests for psutil.net_connections() and Process.net_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)�HAS_NET_CONNECTIONS_UNIX)� SKIP_SYSCONS)�PsutilTestCase)� bind_socket)�bind_unix_socket)�check_connection_ntuple)�create_sockets)�filter_proc_net_connections)�pytest)� reap_children)�retry_on_failure)�skip_on_access_denied)�tcp_socketpair)�unix_socketpair)� wait_for_file�SOCK_SEQPACKETc C s$ t �� j| d�}| dkr t|�S |S )N��kind)�all�unix)�psutil�Process�net_connectionsr )r# �cons� r* �M/opt/hc_python/lib64/python3.8/site-packages/psutil/tests/test_connections.py�this_proc_net_connections2 s r, �serial)�namec @ s& e Zd Zdd� Zdd� Zd dd�ZdS ) �ConnectionTestCasec C s t dd�g kst�d S �Nr$ r" �r, �AssertionError��selfr* r* r+ �setUp; s zConnectionTestCase.setUpc C s t dd�g kst�d S r0 r1 r3 r* r* r+ �tearDown> s zConnectionTestCase.tearDownr$ c sf zt j|d�}W n" t jk r2 tr,Y dS � Y nX � fdd�|D �}|�� |�� ||ksbt�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�cr8 r* r+ � <listcomp>Q s zBConnectionTestCase.compare_procsys_connections.<locals>.<listcomp>)r& r( Z AccessDeniedr �sortr2 )r4 r9 Z proc_consr# Zsys_consr* r8 r+ �compare_procsys_connectionsB s z.ConnectionTestCase.compare_procsys_connectionsN)r$ )�__name__� __module__� __qualname__r5 r6 r? r* r* r* r+ r/ 9 s r/ c @ s4 e Zd Zejjedd�dd� �Zdd� Zdd� Z d S ) �TestBasicOperations� requires root��reasonc C s0 t � � tjdd�D ] }t|� qW 5 Q R X d S r0 )r r& r( r �r4 �connr* r* r+ � test_systemX s z TestBasicOperations.test_systemc C s. t � � tdd�D ] }t|� qW 5 Q R X d S r0 )r r, r rG r* r* r+ � test_process^ s z TestBasicOperations.test_processc C sF t �t�� tdd� W 5 Q R X t �t�� tjdd� W 5 Q R X d S )Nz???r" )r Zraises� ValueErrorr, r& r( r3 r* r* r+ �test_invalid_kindc s z%TestBasicOperations.test_invalid_kindN) r@ rA rB r �mark�skipifr rI rJ rL r* r* r* r+ rC W s rC c @ s� e Zd ZdZdd� Zdd� Zdd� Zejj e � dd �d d � �Z d d � Z ejj e � dd �dd� �Z ejj e dd �dd� �Zejj e dd �dd� �ZdS )�TestUnconnectedSocketsz;Tests sockets which are open but not connected to anything.c C st t dd�}tdd� |D ��}ts$tr0||�� S t|�dks@t�|d jdkrh||�� j|�� ksht�|d S d S )Nr$ r" c S s g | ]}|j |f�qS r* )�fdr: r* r* r+ r= p s z=TestUnconnectedSockets.get_conn_from_sock.<locals>.<listcomp>� r r7 )r, �dictr r �fileno�lenr2 rP )r4 �sockr) Zsmapr* r* r+ �get_conn_from_sockn s z)TestUnconnectedSockets.get_conn_from_sockc C s� | � |�}t|� |jdkr.|j|�� ks.t�|j|jks>t�|j|�tj tj �ksXt�|� � }|szt rzt |t�rz|�� }|jtkr�|dd� }|j|ks�t�|jtkr�tr�tdd�}| 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). r7 N� r$ r" )rV r rP rS r2 �family�type� getsockopt�socket� SOL_SOCKET�SO_TYPE� getsocknamer � isinstance�bytes�decoder �laddrr r r, r? �os�getpid)r4 rU rH rb r) r* r* r+ � check_socket{ s z#TestUnconnectedSockets.check_socketc C sN d}t ttt|d���.}| �|�}|jdks0t�|jtj ks@t�W 5 Q R X d S �N�� 127.0.0.1r ��addrr* ) r r r r re �raddrr2 �statusr& � CONN_LISTEN�r4 rj rU rH r* r* r+ � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedrE c C sN d}t ttt|d���.}| �|�}|jdks0t�|jtj ks@t�W 5 Q R X d S �N)�::1r ri r* ) r r r r re rk r2 rl r& rm rn r* r* r+ � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sN d}t ttt|d���.}| �|�}|jdks0t�|jtj ks@t�W 5 Q R X d S rf ) r r r r re rk r2 rl r& � CONN_NONErn r* r* r+ � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sN d}t ttt|d���.}| �|�}|jdks0t�|jtj ks@t�W 5 Q R X d S rp ) r r r r re rk r2 rl r& rs rn r* r* r+ � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6� POSIX onlyc C sP | � � }tt|td���.}| �|�}|jdks2t�|jtj ksBt�W 5 Q R X d S �N)rY � � � get_testfnr r r re rk r2 rl r& rs �r4 �testfnrU rH r* r* r+ � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sP | � � }tt|td���.}| �|�}|jdks2t�|jtj ksBt�W 5 Q R X d S rw ry r{ r* r* r+ � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)r@ rA rB �__doc__rV re ro r rM rN r rr rt ru r r} r~ r* r* r* r+ rO j s rO c @ sB e Zd ZdZejjedd�dd� �Zejje dd�dd� �Z d S ) �TestConnectedSocketzFTest socket pairs which are actually connected to each other. zunreliable on SUONSrE c C s� d}t dd�g kst�tt|d�\}}zFt dd�}t|�dksBt�|d jtjksVt�|d jtjksjt�W 5 |�� |�� X d S )Nrg �tcp4r" ri rW r rQ ) r, r2 r r �closerT rl r& ZCONN_ESTABLISHED)r4 rj �server�clientr) r* r* r+ �test_tcp� s z TestConnectedSocket.test_tcprv c C s | � � }t|�\}}z�tdd�}|d jr<|d jr<t|��|d jrX|d jrXt|��ts`trndd� |D �}t |�dks~t�t s�ts�t s�t r�|d jdks�t�|d jdks�t�||d jp�|d jks�t�n |d jp�|d j|ks�t�W 5 |�� |�� X d S ) Nr% r" r rQ c S s g | ]}|j d kr|�qS )z /var/run/log)rk r: r* r* r+ r= � s z1TestConnectedSocket.test_unix.<locals>.<listcomp>rW rx ) rz r r� r, rb rk r2 r r rT r r r )r4 r| r� r� r) r* r* r+ � test_unix� s z TestConnectedSocket.test_unixN) r@ rA rB r r rM rN r r� r r� r* r* r* r+ r� � s r� c @ s. e Zd Zdd� Zeed�dd� �Zdd� ZdS ) � TestFiltersc C s� dd� }t � �� |dtttgtttg� |dttgttg� |dtgttg� |dttgtg� |dtgtg� |dtgtg� |d ttgtg� |d tgtg� |d tgtg� tr�|d tgtttg� W 5 Q R X d S ) Nc S sb t | d�D ] }|j|ks t�|j|ks t�q ts^tj| d�D ] }|j|ksNt�|j|ks<t�q<d S )Nr" )r, rX r2 rY r r& r( )r# �families�typesrH r* r* r+ �check s z'TestFilters.test_filters.<locals>.checkr$ �inet�inet4�tcpr� �tcp6�udp�udp4�udp6r% )r r r r r r r! r )r4 r� r* r* r+ � test_filters s* ��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 }t�� �� D ]�}|�� }t|�d k�s8t�|D ]�}|j| jk�rh|||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<�qd S )Nc s� d}t |� |j|kst�|j|ks(t�|j|ks6t�|j|ksDt�|j|ksRt�|D ]2} | j| d�} | |kr|| g ks�t�qV| g ksVt�qVtr�� � | j |g� d S )N) r$ r� r� �inet6r� r� r� r� r� r� r" ) r rX r2 rY rb rk rl r( r r? r9 ) �procrH rX rY rb rk rl �kindsZ all_kindsr# r) r3 r* r+ � check_conn$ s z+TestFilters.test_combos.<locals>.check_conna4 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(0.1) for x in range(100)] 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(0.1) for x in range(100)] )�dirrh )rX rj r| rq T)�deleterQ r* )r$ r� r� r� r� )r$ r� r� r� r� r9 )r$ r� r� r� r� )r$ r� r� r� r� ) r �textwrap�dedentrc �path�basenamerz �getcwd�format�intr r �pyrun�evalr r r& r' �childrenr( rT r2 r9 r rm r rs �getattr)r4 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) rH r* r3 r+ � test_combos s� � � � � � � � �zTestFilters.test_combosc C s� t � ��� tdd�}t|�t� r$dndks.t�|D ]$}|jttfksHt�|jt ks2t�q2tdd�}t|�dksrt�|d jtks�t�|d jt ks�t�t� r�tdd�}t|�dks�t�|d jtks�t�|d jt ks�t�tdd�}t|�t� r�dndks�t�|D ]*}|jttfk�st�|jt k�st��qtd d�}t|�dk�sJt�|d jtk�s^t�|d jt k�srt�t� �r�td d�}t|�dk�s�t�|d jtk�s�t�|d jt k�s�t�td d�}t|�t� �r�d ndk�s�t�|D ].}|jttfk�st�|jt t fk�s�t��q�t� �rntd d�}t|�dk�s>t�|D ]*}|jtk�sVt�|jt t fk�sBt��qBt �r�t �s�t �s�tdd�}t|�dk�s�t�|D ]*}|jtk�s�t�|jt t fk�s�t��q�W 5 Q R X d S )Nr� r" rW rQ r� r r� r� r� r� r� � r� r% � )r r, rT r r2 rX r r rY r r r r r r )r4 r) rH r* r* r+ � test_count� s\ zTestFilters.test_countN)r@ rA rB r� r r r� r� r* r* r* r+ r� s r� rD rE c @ s& e Zd ZdZdd� Ze� dd� �ZdS )�TestSystemWideConnectionsz Tests for net_connections().c C s� dd� }t � �h ddlm} |�� D ]L\}}|dkr:ts:q$|\}}t�|�}t|�tt|��ksdt �||||� q$W 5 Q R X d S )Nc S s< | D ]2}|j |kst�|j tkr.|j|ks.t�t|� qd S )N)rX r2 r rY r )r) r� �types_rH r* r* r+ r� � s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmapr% ) r �psutil._commonr� �itemsr r& r( rT �setr2 )r4 r� r� r# �groupsr� r� r) r* r* r+ �test_it� s z!TestSystemWideConnections.test_itc s� t � �}t|�}W 5 Q R X g �d}g }t|�D ]:}| �� }|�|� t�d| �}| �|�}��|j� q.|D ] }t |� qn�fdd�t j dd�D �} �D ]B� t� fdd�| D ��|ks�t �t � � �} t| � d��|ks�t �q�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(0.1) for x in range(100)] c s g | ]}|j � kr|�qS r* r8 �r; �x)�pidsr* r+ r= s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r$ r" c s g | ]}|j � kr|�qS r* r8 r� r8 r* r+ r= ! s )r rT �rangerz �appendr� r� r� r9 r r& r( r2 r' ) r4 �socks�expected�times�fnames�_�fname�srcZsprocZsysconsr� r* )r9 r� r+ �test_multi_sockets_procs� s, � � z2TestSystemWideConnections.test_multi_sockets_procsN)r@ rA rB r r� r r� r* r* r* r+ r� � s r� c @ s e Zd Zdd� ZdS )�TestMiscc C s� g }g }t t�D ]\}|�d�rtt|�}t|�}|�� s@t|��t|ksLt�||ksXt�|�|� |�|� qtr~tj tj t r�tj d S )NZCONN_) r� r& � startswithr� �str�isupperr2 r� r Z CONN_IDLEZ CONN_BOUNDr ZCONN_DELETE_TCB)r4 Zints�strsr. �numZstr_r* r* r+ � test_net_connection_constants' s z&TestMisc.test_net_connection_constantsN)r@ rA rB r� r* r* r* r+ r� & s r� )6r rc r[ r� � contextlibr r r r r r& 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 r� �objectr! r, rM Z xdist_groupr/ rC rO r� r� rN r� r� r* r* r* r+ �<module> sb ] 7 g?