🔐 Sid Gifari File Manager Pro
v8.0.5 | 2026-08-08 00:57:52 | PHP 8.2.32
📂
/ (Root)
/
opt
/
alt
/
python38
/
lib64
/
python3.8
/
site-packages
/
psutil
/
tests
/
__pycache__
📍 /opt/alt/python38/lib64/python3.8/site-packages/psutil/tests/__pycache__
🔄 Refresh
✏️
Editing: test_connections.cpython-38.pyc
Read Only
U �|�fmU � @ 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-�Z1e�2ed/�G d0d1� d1e-��Z3G d2d3� d3e �Z4e5d4k�r\dd5l6m7Z7 e7e8� dS )6zFTests 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)� reap_children)�retry_on_failure)� serialrun)�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* �P/opt/alt/python38/lib64/python3.8/site-packages/psutil/tests/test_connections.py�this_proc_net_connections3 s r, c @ s& e Zd Zdd� Zdd� Zd dd�ZdS ) �ConnectionTestCasec C s | � tdd�g � d S �Nr$ r" �� assertEqualr, ��selfr* r* r+ �setUp< s zConnectionTestCase.setUpc C s | � tdd�g � d S r. r/ r1 r* r* r+ �tearDown? s zConnectionTestCase.tearDownr$ c sf zt j|d�}W n" t jk r2 tr,Y 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�cr6 r* r+ � <listcomp>R s zBConnectionTestCase.compare_procsys_connections.<locals>.<listcomp>)r& r( Z AccessDeniedr �sortr0 )r2 r7 Z proc_consr# Zsys_consr* r6 r+ �compare_procsys_connectionsC s z.ConnectionTestCase.compare_procsys_connectionsN)r$ )�__name__� __module__� __qualname__r3 r4 r= r* r* r* r+ r- : s r- c @ s0 e Zd Ze�ed�dd� �Zdd� Zdd� ZdS ) �TestBasicOperations� requires rootc C s0 t � � tjdd�D ] }t|� qW 5 Q R X d S r. )r r& r( r �r2 �connr* r* r+ � test_systemY s z TestBasicOperations.test_systemc C s. t � � tdd�D ] }t|� qW 5 Q R X d S r. )r r, r rC r* r* r+ � test_process_ s z TestBasicOperations.test_processc C s& | j ttdd� | j ttjdd� d S )Nz???r" )� assertRaises� ValueErrorr, r& r( r1 r* r* r+ �test_invalid_kindd s z%TestBasicOperations.test_invalid_kindN) r> r? r@ �unittest�skipIfr rE rF rI r* r* r* r+ rA X s rA 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 st t dd�}tdd� |D ��}ts$tr0||�� S | �t|�d� |d jdkrh| �||�� j|�� � |d S d S )Nr$ r" c S s g | ]}|j |f�qS r* )�fdr8 r* r* r+ r; o s z=TestUnconnectedSockets.get_conn_from_sock.<locals>.<listcomp>� r r5 )r, �dictr r �filenor0 �lenrM )r2 �sockr) Zsmapr* r* r+ �get_conn_from_sockm s z)TestUnconnectedSockets.get_conn_from_sockc C s� | � |�}t|� |jdkr.| �|j|�� � | �|j|j� | �|j|�tj tj �� |� � }|szt rzt |t�rz|�� }|jtkr�|dd� }| �|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). r5 N� r$ r" )rS r rM r0 rP �family�type� getsockopt�socket� SOL_SOCKET�SO_TYPE� getsocknamer � isinstance�bytes�decoder �laddrr r r, r= �os�getpid)r2 rR rD r_ r) r* r* r+ � check_socketz s& � z#TestUnconnectedSockets.check_socketc C sN d}t ttt|d���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S �N�� 127.0.0.1r ��addrr* ) r r r r rb r0 �raddr�statusr& � CONN_LISTEN�r2 rg rR rD r* r* r+ � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedc C sN d}t ttt|d���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S �N)�::1r rf r* ) r r r r rb r0 rh ri r& rj rk r* r* r+ � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sN d}t ttt|d���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S rc ) r r r r rb r0 rh ri r& � CONN_NONErk r* r* r+ � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sN d}t ttt|d���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S rm ) r r r r rb r0 rh ri r& rp rk r* r* r+ � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6� POSIX onlyc C sP | � � }tt|td���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S �N)rV � � � get_testfnr r r rb r0 rh ri r& rp �r2 �testfnrR rD r* r* r+ � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sP | � � }tt|td���.}| �|�}| �|jd� | �|jtj � W 5 Q R X d S rt rv rx r* r* r+ � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)r> r? r@ �__doc__rS rb rl rJ rK r ro rq rr r rz r{ r* r* r* r+ rL i s rL c @ s: e Zd ZdZe�ed�dd� �Ze�e d�dd� �Z dS ) �TestConnectedSocketzFTest socket pairs which are actually connected to each other. zunreliable on SUONSc C s� d}| � tdd�g � tt|d�\}}zFtdd�}| � t|�d� | � |d jtj� | � |d jtj� W 5 |�� |�� X d S )Nrd �tcp4r" rf rT r rN ) r0 r, r r �closerQ ri r& ZCONN_ESTABLISHED)r2 rg �server�clientr) r* r* r+ �test_tcp� s z TestConnectedSocket.test_tcprs c C s | � � }t|�\}}z�tdd�}|d jr<|d jr<t|��|d jrX|d jrXt|��ts`trndd� |D �}| j t |�d|d� t s�ts�t s�t r�| � |d jd � | � |d jd � | � ||d jp�|d j� n | � |d jp�|d j|� W 5 |�� |�� X d S ) Nr% r" r rN c S s g | ]}|j d kr|�qS )z /var/run/log)rh r8 r* r* r+ r; � s z1TestConnectedSocket.test_unix.<locals>.<listcomp>rT ��msgru )rw r r r, r_ rh �AssertionErrorr r r0 rQ r r r )r2 ry r� r� r) r* r* r+ � test_unix� s z TestConnectedSocket.test_unixN) r> r? r@ r| rJ rK r r� r r� r* r* r* r+ r} � s r} 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 5 Q R X d S ) Nc sb t | d�D ] }� �|j|� � �|j|� q ts^tj| d�D ] }� �|j|� � �|j|� q<d S )Nr" )r, �assertInrU rV r r& r( )r# �families�typesrD r1 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 )r2 r� r* r1 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 � |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:�qd S )Nc s� d}t |� � �|j|� � �|j|� � �|j|� � �|j|� � �|j|� |D ]2} | j| d�} | |kr|� �| g � qV� �| g � qVt r�� � | j |g� d S )N) r$ r� r� �inet6r� r~ r� r� r� r� r" ) r r0 rU rV r_ rh ri r( ZassertNotEqualr r= r7 ) �procrD rU rV r_ rh ri �kindsZ all_kindsr# r) r1 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)] )�dirre )rU rg ry rn T)�deleterN r* )r$ r� r� r� r~ )r$ r� r� r� r� r7 )r$ r� r� r� r� )r$ r� r� r� r� ) r �textwrap�dedentr` �path�basenamerw �getcwd�format�intr r �pyrun�evalr r r& r' �childrenr( r0 rQ r7 r rj r rp �getattr)r2 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) rD r* r1 r+ � test_combos! s� � � � � � � � �zTestFilters.test_combosc C s� t � ��� tdd�}| �t|�t� r(dnd� |D ]$}| �|jttf� | �|j t � q2tdd�}| �t|�d� | �|d jt� | �|d j t � t� r�tdd�}| �t|�d� | �|d jt� | �|d j t � tdd�}| �t|�t� r�dnd� |D ]&}| �|jttf� | �|j t � �qtd d�}| �t|�d� | �|d jt� | �|d j t � t� �r�td d�}| �t|�d� | �|d jt� | �|d j t � td d�}| �t|�t� �r�d nd� |D ]*}| �|jttf� | �|j t t f� �q�t� �rRtd d�}| �t|�d� |D ]&}| �|jt� | �|j t t f� �q*t �r�t �s�t�s�tdd�}| �t|�d� |D ]&}| �|jt� | �|j t t f� �q�W 5 Q R X d S )Nr� r" rT rN r~ r r� r� r� r� r� � r� r% � )r r, r0 rQ r r� rU r r rV r r r r r r )r2 r) rD r* r* r+ � test_count� s\ zTestFilters.test_countN)r> r? r@ r� r r r� r� r* r* r* r+ r� s r� rB c @ s& e Zd ZdZdd� Ze� dd� �ZdS )�TestSystemWideConnectionsz Tests for net_connections().c s� � fdd�}t � �h ddlm} |�� D ]L\}}|dkr>ts>q(|\}}t�|�}� �t|�tt |��� ||||� q(W 5 Q R X d S )Nc sD | D ]:}� j |j||d� |jtkr6� j |j||d� t|� qd S )Nr� )r� rU r rV r )r) r� �types_rD r1 r* r+ r� � s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmapr% ) r �psutil._commonr� �itemsr r& r( r0 rQ �set)r2 r� r� r# �groupsr� r� r) r* r1 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 ��|� t � � �} | � t| � d��|� 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* r6 �r9 �x)�pidsr* r+ r; s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r$ r" c s g | ]}|j � kr|�qS r* r6 r� r6 r* r+ r; # s )r rQ �rangerw �appendr� r� r� r7 r r& r( r0 r' ) r2 �socks�expected�times�fnames�_�fname�srcZsprocZsysconsr� r* )r7 r� r+ �test_multi_sockets_procs� s2 � � � z2TestSystemWideConnections.test_multi_sockets_procsN)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 }t t�D ]\}|�d�rtt|�}t|�}|�� s@t|��| �t|� | �||� |�|� |�|� qt r~tj tj t r�tj d S )NZCONN_)r� r& � startswithr� �str�isupperr� Z assertNotInr� r Z CONN_IDLEZ CONN_BOUNDr ZCONN_DELETE_TCB)r2 Zints�strs�name�numZstr_r* r* r+ � test_net_connection_constants* s z&TestMisc.test_net_connection_constantsN)r> r? r@ r� r* r* r* r+ r� ) s r� �__main__)� run_from_name)9r| r` rX r� rJ � 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- rA rL r} r� rK r� r� r> Zpsutil.tests.runnerr� �__file__r* r* r* r+ �<module> sj _7 g A
💾 Save Changes
❌ Cancel