Advanced Web Server Manager
Complete File Manager & Terminal - Standalone Version
By Sid Gifari | Gifari Industries
Current path:
/
/
usr
/
local
/
lib64
/
python3.6
/
site-packages
/
psutil
/
tests
/
__pycache__
✏️
Editing: test_connections.cpython-36.pyc
3 �̪e�T � @ s` 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)ed"e*� �Z+d#d$� Z,e$G d%d&� d&e ��Z-G d'd(� d(e-�Z.e$G d)d*� d*e-��Z/e$G d+d,� d,e-��Z0G d-d.� d.e-�Z1ej2ed/�G d0d1� d1e-��Z3G d2d3� d3e �Z4e5d4k�r\dd5l6m7Z7 e7e8� 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)�HAS_CONNECTIONS_UNIX)� SKIP_SYSCONS)�PsutilTestCase)� bind_socket)�bind_unix_socket)�check_connection_ntuple)�create_sockets)�filter_proc_connections)� reap_children)�retry_on_failure)� serialrun)�skip_on_access_denied)�tcp_socketpair)�unix_socketpair)� wait_for_file�SOCK_SEQPACKETc C s$ t j� j| d�}| dkr t|�S |S )N)�kind�all�unix)r# r$ )�psutil�Process� connectionsr )r" �cons� r) �I/usr/local/lib64/python3.6/site-packages/psutil/tests/test_connections.py�this_proc_connections3 s r+ c @ s& e Zd Zdd� Zdd� Zd dd�ZdS ) �ConnectionTestCasec C s | j tdd�g � d S )Nr# )r" )� assertEqualr+ )�selfr) r) r* �setUp< s zConnectionTestCase.setUpc C s | j tdd�g � d S )Nr# )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 �}|j� |j� | j||� 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)r3 r) r* � <listcomp>R s zBConnectionTestCase.compare_procsys_connections.<locals>.<listcomp>)r% �net_connections� AccessDeniedr �sortr- )r. r3 Z proc_consr" Zsys_consr) )r3 r* �compare_procsys_connectionsC s z.ConnectionTestCase.compare_procsys_connectionsN)r# )�__name__� __module__� __qualname__r/ r0 r: r) r) r) r* r, : s r, c @ s0 e Zd Zejed�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 r% r7 r )r. �connr) r) r* � test_systemY s z TestBasicOperations.test_systemc C s2 t � �" xtdd�D ] }t|� qW W d Q R X d S )Nr# )r" )r r+ r )r. r? r) r) r* � test_process_ s z TestBasicOperations.test_processc C s& | j ttdd� | j ttjdd� d S )Nz???)r" )Z assertRaises� ValueErrorr+ r% r7 )r. r) r) r* �test_invalid_kindd s z%TestBasicOperations.test_invalid_kindN) r; r< r= �unittest�skipIfr r@ rA rC r) r) r) r* r> X s r> c @ s� e Zd ZdZdd� Zdd� Zdd� Zeje � d�d d � �Z d d � Z eje � d�d d� �Z eje d�dd� �Zeje d�dd� �ZdS )�TestUnconnectedSocketsz;Tests sockets which are open but not connected to anything.c C st t dd�}tdd� |D ��}ts$tr0||j� S | jt|�d� |d jdkrh| j||j� j|j� � |d S d S )Nr# )r" c S s g | ]}|j |f�qS r) )�fd)r4 r5 r) r) r* r6 o s z=TestUnconnectedSockets.get_conn_from_sock.<locals>.<listcomp>r1 r r2 )r+ �dictr r �filenor- �lenrG )r. �sockr( Zsmapr) r) r* �get_conn_from_sockm s z)TestUnconnectedSockets.get_conn_from_sockc C s� | j |�}t|� |jdkr.| j|j|j� � | j|j|j� | j|j|jtj tj �� |j � }| r|t r|t |t�r||j� }|jtkr�|dd� }| j|j|� |jtkr�tr�tdd�}| jtj� |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 N� r# )r" r2 )rL r rG r- rI �family�type� getsockopt�socket� SOL_SOCKET�SO_TYPE� getsocknamer � isinstance�bytes�decoder �laddrr r r+ r: �os�getpid)r. rK r? rX r( r) r) r* � check_socketz s" z#TestUnconnectedSockets.check_socketc C sN d}t ttt|d���.}| j|�}| j|jf � | j|jtj � W d Q R X d S )N� 127.0.0.1r )�addr)r\ r ) r r r r r[ r- �raddr�statusr% � CONN_LISTEN)r. r] rK r? r) r) r* � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedc C sN d}t ttt|d���.}| j|�}| j|jf � | j|jtj � W d Q R X d S )N�::1r )r] )rb r ) r r r r r[ r- r^ r_ r% r` )r. r] rK r? r) r) r* � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sN d}t ttt|d���.}| j|�}| j|jf � | j|jtj � W d Q R X d S )N� 127.0.0.1r )r] )rd r ) r r r r r[ r- r^ r_ r% � CONN_NONE)r. r] rK r? r) r) r* � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sN d}t ttt|d���.}| j|�}| j|jf � | j|jtj � W d Q R X d S )N�::1r )r] )rg r ) r r r r r[ r- r^ r_ r% re )r. r] rK r? r) r) r* � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6z POSIX onlyc C sP | j � }tt|td���.}| j|�}| j|jd� | j|jtj � W d Q R X d S )N)rO � ) � get_testfnr r r r[ r- r^ r_ r% re )r. �testfnrK r? r) r) r* � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sP | j � }tt|td���.}| j|�}| j|jd� | j|jtj � W d Q R X d S )N)rO ri ) rj r r r r[ r- r^ r_ r% re )r. rk rK r? r) r) r* � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)r; r< r= �__doc__rL r[ ra rD rE r rc rf rh r rl rm r) r) r) r* rF i s rF c @ s: e Zd ZdZejed�dd� �Zeje d�dd� �Z dS ) �TestConnectedSocketzFTest socket pairs which are actually connected to each other. zunreliable on SUONSc C s� d}| j tdd�g � tt|d�\}}zFtdd�}| j t|�d� | j |d jtj� | j |d jtj� W d |j� |j� X d S ) N� 127.0.0.1r �tcp4)r" )r] rM r1 )rp r ) r- r+ r r rJ r_ r% �CONN_ESTABLISHED�close)r. r] �server�clientr( r) r) r* �test_tcp� s z TestConnectedSocket.test_tcpz POSIX onlyc C s | j � }t|�\}}z�tdd�}|d jo2|d j s>t|��|d joP|d j s\t|��tsdtrrdd� |D �}| jt |�d|d� t s�ts�t s�t r�| j|d jd � | j|d jd � | j||d jp�|d j� n | j|d jp�|d j|� W d |j � |j � X d S ) Nr$ )r" r r1 c S s g | ]}|j d kr|�qS )z /var/run/log)r^ )r4 r5 r) r) r* r6 � s z1TestConnectedSocket.test_unix.<locals>.<listcomp>rM )�msgri )rj r r+ rX r^ �AssertionErrorr r r- rJ r r r rs )r. rk rt ru r( r) r) r* � test_unix� s z TestConnectedSocket.test_unixN) r; r< r= rn rD rE r rv r ry r) r) r) r* ro � s ro 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 sj x.t | d�D ] }� j|j|� � j|j|� q W tsfx0tj| d�D ] }� j|j|� � j|j|� qBW d S )N)r" )r+ �assertInrN rO r r% r7 )r" �families�typesr? )r. r) r* �check s z'TestFilters.test_filters.<locals>.checkr# �inet�inet4�tcprq �tcp6�udp�udp4�udp6r$ )r r r r r r r! r )r. r~ r) )r. r* � test_filters s& zTestFilters.test_filters)Zonly_ifc s t � � fdd�}tjd�}tjd�}tjj� jtj� d��}|jt t �d|d�}|jt t �d|d�}|jt t �d|d�}|jt t �d|d�}� j |�} t t|d d ��} � j |�} t t|d d ��} t� r�� j |�} t t|d d ��}� j |�}t t|d d ��}nd } d }d }d }� x�tj� j� D � ]�}|j� }� jt|�d � x�|D ]�}|j| jk�rn|||t t| f tjd� n�|j| jk�r�|||t t| f tjd� nZ|jt| dd �k�r�|||t t|f tjd� n,|jt|dd �k�rB|||t t|f tjd� �qBW �q W d S )Nc s� d }t |� � j|j|� � j|j|� � j|j|� � j|j|� � j|j|� x:|D ]2} | j| d �} | |kr~� j| g � qX� j| g � qXW t r�� j | j |g� d S ) Nr# r r� �inet6r� rq r� r� r� r� )r" ) r# r r� r� r� rq r� r� r� r� ) r r- rN rO rX r^ r_ r' ZassertNotEqualr r: r3 ) �procr? rN rO rX 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)rN r] rk z::1T)�deleter1 r# r r� r� rq r� r� r3 r� r� r� )r# r r� r� rq )r# r r� r� r� )r# r r� r� r� )r# r r� r� r� ) r �textwrap�dedentrY �path�basenamerj �getcwd�format�intr r �pyrun�evalr r r% r& �childrenr' r- rJ r3 r r` r re �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 � ��� tdd�}| jt|�t� r(dnd� x,|D ]$}| j|jttf� | j|j t � q4W tdd�}| jt|�d� | j|d jt� | j|d j t � t� r�tdd�}| jt|�d� | j|d jt� | j|d j t � tdd�}| jt|�t� r�dnd� x.|D ]&}| j|jttf� | j|j t � �qW td d�}| jt|�d� | j|d jt� | j|d j t � t� �r�td d�}| jt|�d� | j|d jt� | j|d j t � td d�}| jt|�t� �r�d nd� x2|D ]*}| j|jttf� | j|j t t f� �q�W t� �rbtd d�}| jt|�d� x.|D ]&}| j|jt� | j|j t t f� �q8W t �r�t �ppt �r�tdd�}| jt|�d� x.|D ]&}| j|jt� | j|j t t f� �q�W W d Q R X d S )Nr� )r" rM r1 rq r r� r� r� r� r � r� r$ � )r r+ r- rJ r r{ rN r r rO 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* rz s rz z requires rootc @ s& e Zd ZdZdd� Ze� dd� �ZdS )�TestSystemWideConnectionsz Tests for net_connections().c s� � fdd�}t � �n ddlm} xZ|j� D ]N\}}|dkrBt rBq*|\}}tj|�}� jt|�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)rw )r{ rN r rO r )r( r| �types_r? )r. r) r* r~ � s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmapr$ ) r �psutil._commonr� �itemsr r% r7 r- rJ �set)r. r~ r� r" �groupsr| 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 ]:}| j� }|j|� tjd| �}| j|�}�j|j� q0W x|D ] }t |� qtW �fdd�t j dd�D �} xJ�D ]B� | j t� fdd�| D ��|� t j � �} | j t| jd��|� 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) )r3 )r4 �x)�pidsr) r* r6 s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r# )r" c s g | ]}|j � kr|�qS r) )r3 )r4 r� )r3 r) r* r6 # s )r rJ �rangerj �appendr� r� r� r3 r r% r7 r- r& r' ) r. ZsocksZexpected�times�fnames�_�fname�srcZsprocZsysconsr� r) )r3 r� r* �test_multi_sockets_procs� s( z2TestSystemWideConnections.test_multi_sockets_procsN)r; r< r= rn r� r r� r) r) r) r* r� � s r� c @ s e Zd Zdd� ZdS )�TestMiscc C s� g }g }xht t�D ]\}|jd�rtt|�}t|�}|j� sBt|��| jt|� | j||� |j|� |j|� qW t r�tj tj t r�tj d S )NZCONN_)r� r% � startswithr� �str�isupperrx 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)9rn rY rQ r� rD � 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+ r, r> rF ro rz rE r� r� r; Zpsutil.tests.runnerr� �__file__r) r) r) r* �<module> sj _ 7 gA
💾 Save Changes
❌ Cancel