📁 File Manager Pro
v10.0.2 | PHP: 8.1.34
Server: LiteSpeed
2026-06-27 21:15:42
📂
/
/
usr
/
lib64
/
python3.6
/
site-packages
/
psutil
/
tests
/
__pycache__
✏️
Editing: test_connections.cpython-36.pyc
3 ��JZ�P � @ s8 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' dd"lm(Z( dd#lm)Z) dd$lm*Z* e j+� Z,G d%d&� d&e-�Z.G d'd(� d(e.e'j/�Z0G d)d*� d*e.e'j/�Z1G d+d,� d,e.e'j/�Z2G d-d.� d.e'j/�Z3e4d/k�r4e!e5� dS )0z;Tests for net_connections() and Process.connections() APIs.� N)�closing)�AF_INET)�AF_INET6)� SOCK_DGRAM)� SOCK_STREAM)�FREEBSD)�LINUX)�NETBSD)�OPENBSD)�OSX)�POSIX)�SUNOS)�WINDOWS)� supports_ipv6)�PY3)�AF_UNIX)� bind_socket)�bind_unix_socket)�check_connection_ntuple)�create_sockets)� get_free_port)�HAS_CONNECTIONS_UNIX)�pyrun)� reap_children)�run_test_module_by_name)� safe_rmpath)�skip_on_access_denied)�tcp_socketpair)�TESTFN)�TRAVIS)�unittest)�unix_socket_path)�unix_socketpair)� wait_for_filec @ s8 e Zd Zdd� Zdd� Zdd� Zd dd �Zdd d �ZdS )�Basec C s" t s tjdd�}| s t|��d S )N�all)�kind)r �thisproc� connections�AssertionError)�self�cons� r, �(/usr/lib64/python3.6/test_connections.py�setUp6 s z Base.setUpc C s0 t t� t� ts,tjdd�}| s,t|��d S )Nr% )r& )r r r r r' r( r) )r* r+ r, r, r- �tearDown< s z Base.tearDownc C sr t jdd�}tdd� |D ��}tr.||j� S | jt|�d� |d jdkrf| j||j� j|j� � |d S d S )Nr% )r& c S s g | ]}|j |f�qS r, )�fd)�.0�cr, r, r- � <listcomp>G s z+Base.get_conn_from_sock.<locals>.<listcomp>� r ���)r' r( �dictr �fileno� assertEqual�lenr0 )r* �sockr+ Zsmapr, r, r- �get_conn_from_sockE s zBase.get_conn_from_sockNc C s� |dkr| j |�}t|� |jdkr6| j|j|j� � | j|j|j� | j|j|jtj tj �� |j � }| r�t r�t |t�r�|j� }|jtkr�|dd� }|jtkr�tr�n| j|j|� |jtkr�tr�tjdd�}| jtj� |� |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). Nr4 � r% )r& r5 )r; r r0 r8 r7 �family�typeZ getsockopt�socketZ SOL_SOCKETZSO_TYPEZ getsocknamer � isinstance�bytes�decoder r r �laddrr r' r( �compare_procsys_connections�os�getpid)r* r: �connrC r+ r, r, r- � check_socketR s( zBase.check_socketr% 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 )Nr4 r5 )�pid)r1 r2 )rI r, r- r3 � s z4Base.compare_procsys_connections.<locals>.<listcomp>)�psutil�net_connectionsZ AccessDeniedr �sortr8 )r* rI Z proc_consr& Zsys_consr, )rI r- rD v s z Base.compare_procsys_connections)N)r% )�__name__� __module__� __qualname__r. r/ r; rH rD r, r, r, r- r$ 4 s $r$ c @ s| e Zd ZdZdd� Zeje� d�dd� �Zdd� Z eje� d�d d � �Z eje d �d d � �Z eje d �dd� �Z dS )�TestUnconnectedSocketsz;Tests sockets which are open but not connected to anything.c C sR dt � f}tttt|d���,}| j|�}|j s4t�| j|j t j � W d Q R X d S )Nz 127.0.0.1)�addr) r r r r r rH �raddrr) r8 �statusrJ � CONN_LISTEN)r* rQ r: rG r, r, r- � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedc C sR dt � f}tttt|d���,}| j|�}|j s4t�| j|j t j � W d Q R X d S )Nz::1)rQ ) r r r r r rH rR r) r8 rS rJ rT )r* rQ r: rG r, r, r- � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sR dt � f}tttt|d���,}| j|�}|j s4t�| j|j t j � W d Q R X d S )Nz 127.0.0.1)rQ ) r r r r r rH rR r) r8 rS rJ � CONN_NONE)r* rQ r: rG r, r, r- � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sR dt � f}tttt|d���,}| j|�}|j s4t�| j|j t j � W d Q R X d S )Nz::1)rQ ) r r r r r rH rR r) r8 rS rJ rW )r* rQ r: rG r, r, r- � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6z POSIX onlyc C sX t � �H}tt|td���,}| j|�}|j s0t�| j|jt j � W d Q R X W d Q R X d S )N)r> ) r! r r r rH rR r) r8 rS rJ rW )r* �namer: rG r, r, r- � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sX t � �H}tt|td���,}| j|�}|j s0t�| j|jt j � W d Q R X W d Q R X d S )N)r> ) r! r r r rH rR r) r8 rS rJ rW )r* rZ r: rG r, r, r- � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)rM rN rO �__doc__rU r �skipIfr rV rX rY r r[ r\ r, r, r, r- rP � s rP c @ sT e Zd ZdZejed�dd� �Zeje d�dd� �Z e e d�d d � �Z d d � Z d S )�TestConnectedSocketPairszJTest socket pairs which are are actually connected to each other. zunreliable on SUONSc C s� dt � f}tjdd� s t�tt|d�\}}zHtjdd�}| jt|�d� | j|d jt j � | j|d jt j � W d |j � |j � X d S )Nz 127.0.0.1�tcp4)r& )rQ r< r r4 ) r r' r( r) r r r8 r9 rS rJ ZCONN_ESTABLISHED�close)r* rQ �server�clientr+ r, r, r- �test_tcp� s z!TestConnectedSocketPairs.test_tcpz POSIX onlyc C sl t � ��Z}t|�\}}�z2tjdd�}|d jo8|d j s@t�|d joR|d j sZt�trldd� |D �}| jt |�d� t s�t s�t r�| j|d jd� | j|d jd� | j||d jp�|d j� n~t �r xv|d j|d j|d j|d jfD ]}| j|d� q�W n<| j|d j�p$|d j|� | j|d j�pB|d j|� W d |j� |j� X W d Q R X d S ) N�unix)r& r r4 c S s g | ]}|j d kr|�qS )z /var/run/log)rR )r1 r2 r, r, r- r3 � s z6TestConnectedSocketPairs.test_unix.<locals>.<listcomp>r< � )r! r" r' r( rC rR r) r r8 r9 r r r r ra )r* rZ rb rc r+ rQ r, r, r- � test_unix� s* "z"TestConnectedSocketPairs.test_unix)Zonly_ifc s � fdd�}t jd�}t jd�}ddlm} tjjt�}||�jt t �d|d�}||�jt t �d|d�}||�jt t �d |d�}||�jt t �d |d�} t |�} t t|��} t |�} t t|��} t� r�t |�}t t|��}t | �}t t|��}nd }d }d }d }� x�tj� D � ]�}|j� }� jt|�d � x�|D ]�}|j| jk�r\|||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�r0|||t t|f tjd� �q0W �q W � j t |jdd� d S ) Nc s� d }t |� � j|j|� � j|j|� � j|j|� � j|j|� � j|j|� x8|D ]0} | j| d �} | |krz| s�t�qX| sXt| ��qXW t r�� j | j |g� d S ) Nr% �inet�inet4�inet6�tcpr` �tcp6�udp�udp4�udp6)r& ) r% rh ri rj rk r` rl rm rn ro ) r r8 r= r> rC rR rS r( r) r rD rI ) �procrG r= r> rC rR rS ZkindsZ all_kindsr& r+ )r* r, r- � check_conn s z8TestConnectedSocketPairs.test_combos.<locals>.check_conna import socket, time s = socket.socket($family, socket.SOCK_STREAM) s.bind(('$addr', 0)) s.listen(1) with open('$testfn', 'w') as f: f.write(str(s.getsockname()[:2])) time.sleep(60) z� 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) r )�Templatez 127.0.0.1)r= rQ Ztestfnz::1r4 r% rh ri rk r` rm rn rI rj rl ro z???)r& )r% rh ri rk r` )r% rh ri rm rn )r% rh rj rk rl )r% rh rj rm ro ) �textwrap�dedent�stringrr rE �path�basenamer Z substitute�intr r r �evalr# r r' Zchildrenr( r8 r9 rI r rJ rT r rW �getattr� assertRaises� ValueError)r* rq Z tcp_templateZ udp_templaterr ZtestfileZ 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+ rG r, )r* r- � test_combos sb z$TestConnectedSocketPairs.test_combosc C s� t � ���}tjdd�}| jt|�t|�� tjdd�}| jt|�t� rJdnd� x,|D ]$}| j|jtt f� | j|j t � qVW tjdd�}| jt|�d� | j|d jt� | j|d j t � t� �rtjdd�}| jt|�d� | j|d jt � | j|d j t � tjd d�}| jt|�t� �r(dnd� x.|D ]&}| j|jtt f� | j|j t � �q4W tjd d�}| jt|�d� | j|d jt� | j|d j t � t� �r�tjd d�}| jt|�d� | j|d jt � | j|d j t � tjd d�}| jt|�t� �rd nd� x2|D ]*}| j|jtt f� | j|j t t f� �qW t� �r�tjdd�}| jt|�d� x.|D ]&}| j|jt � | j|j t t f� �qlW t �r�tjdd�}| jt|�d� x.|D ]&}| j|jt� | j|j t t f� �q�W W d Q R X d S )Nr% )r& rk r< r4 r` r rl rm rn ro rh � rj re � )r r' r( r8 r9 r �assertInr= r r r> r r r r )r* �socksr+ rG r, r, r- � test_multi_sockets_filteringe s` z5TestConnectedSocketPairs.test_multi_sockets_filteringN)rM rN rO r] r r^ r rd r rg r r r~ r� r, r, r, r- r_ � s !br_ c @ sJ e Zd ZdZe� dd� �Ze� dd� �Ze� eje o6e d�dd� ��Z d S ) �TestSystemWideConnectionsz Tests for net_connections().c s� � fdd�}t � �� ddlm} xZ|j� D ]N\}}|dkrBt rBq*|\}}tj|�}� jt|�tt |��� ||||� q*W � j t tjdd� W d Q R X d S )Nc sV t tdt� �}xB| D ]:}� j|j||d� |j|krF� j|j||d� t|� qW d S )Nr )�msg)rz r? �objectr� r= r> r )r+ �families�types_r rG )r* r, r- �check� s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmapre z???)r& ) r �psutil._commonr� �itemsr rJ rK r8 r9 �setr{ r| )r* r� r� r& �groupsr� r� r+ r, )r* r- �test_it� s z!TestSystemWideConnections.test_itc C s@ t � �0}dd� tjdd�D �}| jt|�t|�� W d Q R X d S )Nc S s g | ]}|j tj� kr|�qS r, )rI rE rF )r1 �xr, r, r- r3 � s z>TestSystemWideConnections.test_multi_socks.<locals>.<listcomp>r% )r& )r rJ rK r8 r9 )r* r� r+ r, r, r- �test_multi_socks� s z*TestSystemWideConnections.test_multi_sockszunreliable on OSX + TRAVISc s t � �}t|�}W d Q R X g �d}xRt|�D ]F}tjjt�t|� }tj d| �}t |�}�j |j � | j t|� q,W x$t|�D ]}tt|� }t|� q�W �fdd�tjdd�D �}xJ�D ]B� | jt� fdd�|D ��|� tj� �} | jt| jd��|� q�W d S )N� a import time, os from psutil.tests import create_sockets with create_sockets(): with open('%s', 'w') as f: f.write(str(os.getpid())) time.sleep(60) c s g | ]}|j � kr|�qS r, )rI )r1 r� )�pidsr, r- r3 � s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r% )r& c s g | ]}|j � kr|�qS r, )rI )r1 r� )rI r, r- r3 � s )r r9 �rangerE rv �realpathr �strrs rt r �appendrI Z addCleanupr r# rJ rK r8 �Processr( ) r* r� Zexpected�times�iZfname�srcZsprocZsysconsr} r, )rI r� r- �test_multi_sockets_procs� s( z2TestSystemWideConnections.test_multi_sockets_procsN) rM rN rO r] r r� r� r r^ 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_)�dirrJ � startswithrz r� �isupperr) Z assertNotInr� r Z CONN_IDLEZ CONN_BOUNDr ZCONN_DELETE_TCB)r* ZintsZstrsrZ ZnumZstr_r, r, r- �test_connection_constants� s z"TestMisc.test_connection_constantsN)rM rN rO r� r, r, r, r- r� � s r� �__main__)6r] rE r? rs � contextlibr r r r r rJ 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! r" r# r� r' r� r$ ZTestCaserP r_ r� r� rM �__file__r, r, r, r- �<module> s^ \7 `P
💾 Save Changes
❌ Cancel