File: //usr/local/lib64/python3.9/site-packages/psutil/tests/__pycache__/test_connections.cpython-39.pyc
a
��?hS � @ 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�}| dv r t|�S |S )N��kind)�all�unix)�psutil�Process�net_connectionsr )r# �cons� r* �I/usr/local/lib64/python3.9/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 ksJ �d S �Nr$ r" �r, ��selfr* r* r+ �setUp; s zConnectionTestCase.setUpc C s t dd�g ksJ �d S r0 r1 r2 r* r* r+ �tearDown> s zConnectionTestCase.tearDownr$ c sd zt j|d�}W n t jy0 tr*Y dS � Y n0 � fdd�|D �}|�� |�� ||ks`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�cr7 r* r+ �
<listcomp>Q � zBConnectionTestCase.compare_procsys_connections.<locals>.<listcomp>)r& r( Z AccessDeniedr �sort)r3 r8 Z proc_consr# Zsys_consr* r7 r+ �compare_procsys_connectionsB s z.ConnectionTestCase.compare_procsys_connectionsN)r$ )�__name__�
__module__� __qualname__r4 r5 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 sD t � �* tjdd�D ] }t|� qW d � n1 s60 Y d S r0 )r r& r( r �r3 �connr* r* r+ � test_systemX s z TestBasicOperations.test_systemc C sB t � �( tdd�D ] }t|� qW d � n1 s40 Y d S r0 )r r, r rG r* r* r+ � test_process^ s z TestBasicOperations.test_processc C sn t �t�� tdd� W d � n1 s*0 Y t �t�� tjdd� W d � n1 s`0 Y d S )Nz???r" )r Zraises�
ValueErrorr, r& r( r2 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@J �|d jdkrh||�� j|�� kshJ �|d S d S )Nr$ r" c S s g | ]}|j |f�qS r* )�fdr9 r* r* r+ r< p r= z=TestUnconnectedSockets.get_conn_from_sock.<locals>.<listcomp>� r r6 )r, �dictr
r �fileno�lenrP )r3 �sockr) Zsmapr* r* r+ �get_conn_from_sockn s
z)TestUnconnectedSockets.get_conn_from_sockc C s� | � |�}t|� |jdkr.|j|�� ks.J �|j|jks>J �|j|�tjtj �ksXJ �|�
� }|szt rzt |t
�rz|�� }|jtkr�|dd� }|j|ks�J �|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).
r6 N� r$ r" )rV r rP rS �family�type�
getsockopt�socket�
SOL_SOCKET�SO_TYPE� getsocknamer �
isinstance�bytes�decoder �laddrr r r, r? �os�getpid)r3 rU rH rb r) r* r* r+ � check_socket{ s
z#TestUnconnectedSockets.check_socketc C sb d}t ttt|d���8}| �|�}|jdks0J �|jtjks@J �W d � n1 sT0 Y d S �N�� 127.0.0.1r ��addrr* ) r r r r re �raddr�statusr& � CONN_LISTEN�r3 rj rU rH r* r* r+ � test_tcp_v4� s
z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedrE c C sb d}t ttt|d���8}| �|�}|jdks0J �|jtjks@J �W d � n1 sT0 Y d S �N)�::1r ri r* ) r r r r re rk rl r& rm rn r* r* r+ � test_tcp_v6� s
z"TestUnconnectedSockets.test_tcp_v6c C sb d}t ttt|d���8}| �|�}|jdks0J �|jtjks@J �W d � n1 sT0 Y d S rf ) r r r r re rk rl r& � CONN_NONErn r* r* r+ � test_udp_v4� s
z"TestUnconnectedSockets.test_udp_v4c C sb d}t ttt|d���8}| �|�}|jdks0J �|jtjks@J �W d � n1 sT0 Y d S rp ) r r r r re rk rl r& rs rn r* r* r+ � test_udp_v6� s
z"TestUnconnectedSockets.test_udp_v6�
POSIX onlyc C sd | � � }tt|td���8}| �|�}|jdks2J �|jtjksBJ �W d � n1 sV0 Y d S �N)rY � � �
get_testfnr r r re rk rl r& rs �r3 �testfnrU rH r* r* r+ �
test_unix_tcp� s
z$TestUnconnectedSockets.test_unix_tcpc C sd | � � }tt|td���8}| �|�}|jdks2J �|jtjksBJ �W d � n1 sV0 Y 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 ksJ �tt|d�\}}zVt dd�}t|�dksBJ �|d jtjksVJ �|d jtjksjJ �W |�� |�� n|�� |�� 0 d S )Nrg �tcp4r" ri rW r rQ )r, r r rT rl r& ZCONN_ESTABLISHED�close)r3 rj �server�clientr) r* r* r+ �test_tcp� s
�z TestConnectedSocket.test_tcprv c C s | � � }t|�\}}z�tdd�}|d jr<|d jr<J |��|d jrX|d jrXJ |��ts`trndd� |D �}t|�dks~J �ts�ts�t s�t
r�|d jdks�J �|d jdks�J �||d jp�|d jks�J �n |d jp�|d j|ks�J �W |� � |� � n|� � |� � 0 d S ) Nr% r" r rQ c S s g | ]}|j d kr|�qS )z /var/run/log)rk r9 r* r* r+ r< � r= z1TestConnectedSocket.test_unix.<locals>.<listcomp>rW rx ) rz r r, rb rk r
r rT r r
r r� )r3 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 d � n1 s�0 Y d S )
Nc S sb t | d�D ] }|j|v s J �|j|v s
J �q
ts^tj| d�D ] }|j|v sNJ �|j|v s<J �q<d S )Nr" )r, rX 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 )r3 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�s8J �|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|ksJ �|j|ks(J �|j|ks6J �|j|ksDJ �|j|ksRJ �|D ]2} | j| d�}
| |v r||
g ks�J �qV|
g ksVJ �qVtr�� �| j |g� d S )N)
r$ r� r� �inet6r� r� r� r� r� r� r" )
r rX rY rb rk rl r( r r? r8 ) �procrH rX rY rb rk rl �kindsZ all_kindsr# r) r2 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� r8 )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 r8 r rm r rs �getattr)r3 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* r2 r+ � test_combos s�
�
�
�
�
� � � �zTestFilters.test_combosc C s� t � ��� tdd�}t|�t� r$dndks.J �|D ]$}|jttfv sHJ �|jtks2J �q2tdd�}t|�dksrJ �|d jtks�J �|d jtks�J �t� r�tdd�}t|�dks�J �|d jtks�J �|d jtks�J �tdd�}t|�t� r�dndks�J �|D ]*}|jttfv �sJ �|jt k�sJ ��qtd d�}t|�dk�sJJ �|d jtk�s^J �|d jt k�srJ �t� �r�td
d�}t|�dk�s�J �|d jtk�s�J �|d jt k�s�J �td d�}t|�t� �r�d ndk�s�J �|D ].}|jttfv �sJ �|jtt fv �s�J ��q�t� �rntd
d�}t|�dk�s>J �|D ]*}|jtk�sVJ �|jtt fv �sBJ ��qBt
�r�t �s�t �s�tdd�}t|�dk�s�J �|D ]*}|jt
k�s�J �|jtt fv �s�J ��q�W d � n1 �s�0 Y d S )Nr� r" rW rQ r� r r� r� r� r� r� � r� r% � )r r, rT r rX r r rY r r r r r
r )r3 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 � �r ddlm} |�� D ]L\}}|dkr:ts:q$|\}}t�|�}t|�tt|��ksdJ �||||� q$W d � n1 s�0 Y d S )Nc S s< | D ]2}|j |v sJ �|j tkr.|j|v s.J �t|� qd S )N)rX 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 �set)r3 r� r� r# �groupsr� r� r) r* r* r+ �test_it� s
z!TestSystemWideConnections.test_itc s� t � �}t|�}W d � n1 s$0 Y g �d}g }t|�D ]:}| �� }|�|� t�d| �}| �|�}��|j� qB|D ] }t |� q��fdd�t
j dd�D �} �D ]B� t� fdd�| D ��|ks�J �t
� � �}
t|
� d��|ks�J �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 � v r|�qS r* r7 �r: �x)�pidsr* r+ r< s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r$ r" c s g | ]}|j � kr|�qS r* r7 r� r7 r* r+ r< ! r= )
r rT �rangerz �appendr� r� r� r8 r r& r( r' ) r3 �socks�expected�times�fnames�_�fname�srcZsprocZsysconsr� r* )r8 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@J |��t|vsLJ �||vsXJ �|�|� |�|� qtr~tj tj t
r�tj d S )NZCONN_) r� r& �
startswithr� �str�isupperr� r
Z CONN_IDLEZ
CONN_BOUNDr ZCONN_DELETE_TCB)r3 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?