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-Z1e2ed/G d0d1 d1e-Z3G d2d3 d3e Z4e5d4kr\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_fileSOCK_SEQPACKETc C s$ t j| d}| dkr t|S |S )Nkind)allunix)psutilProcessnet_connectionsr )r# cons r* P/opt/alt/python38/lib64/python3.8/site-packages/psutil/tests/test_connections.pythis_proc_net_connections3 s r, c @ s& e Zd Zdd Zdd Zd ddZdS )
ConnectionTestCasec C s | tddg d S Nr$ r" assertEqualr, selfr* r* r+ setUp< s zConnectionTestCase.setUpc C s | tddg 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 )zGiven 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 )Npid.0cr6 r* r+
R s
zBConnectionTestCase.compare_procsys_connections..)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 Zeeddd Zdd Zdd ZdS ) TestBasicOperations
requires rootc C s0 t tjddD ] }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 tddD ] }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@ unittestskipIfr rE rF rI r* r* r* r+ rA X s
rA c @ s e Zd ZdZdd Zdd Zdd Zee dd d
Z
d d Z ee dd
d Z ee
ddd Zee
ddd 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 |fqS r* )fdr8 r* r* r+ r; o s z=TestUnconnectedSockets.get_conn_from_sock.. 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
|trz| }|jtkr|dd }| |j| |jtkrtrtdd}| jt |dd |S )zGiven 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 familytype
getsockoptsocket
SOL_SOCKETSO_TYPE getsocknamer
isinstancebytesdecoder laddrr r r, r= osgetpid)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 raddrstatusr& CONN_LISTENr2 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 ZdZeeddd Zee ddd Z dS ) TestConnectedSocketzFTest socket pairs which are actually connected to
each other.
zunreliable on SUONSc C s d}| tddg 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 serverclientr) r* r* r+ test_tcp s
z TestConnectedSocket.test_tcprs c C s
| }t|\}}ztdd}|d jr<|d jr.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 Zeeddd 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 | dD ] } |j| |j| q
ts^tj| dD ] } |j| |j| q.checkr$ inetinet4tcpr~ tcp6udpudp4udp6r% )r r r r r r r! r )r2 r r* r1 r+ test_filters s* zTestFilters.test_filters)Zonly_ifc s t fdd}td}td}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| jkrf|||t
t|
d tjd
n|j| jkr|||t
t| d tjd nZ|jt|
dd kr|||t t|d tjd n,|jt|dd kr:|||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..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 textwrapdedentr` pathbasenamerw getcwdformatintr r pyrunevalr 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_addrpr) 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 rtdd}| t|d | |d jt | |d j t
tdd}| t|t rdnd |D ]&}| |jttf | |j t qtd d}| t|d | |d jt | |d j t t rtd
d}| t|d | |d jt | |d j t td d}| t|t rd nd |D ]*}| |jttf | |j t
t f qt rRtd
d}| t|d |D ]&}| |jt | |j t
t f q*t rt
ststdd}| t|d |D ]&}| |jt | |j t
t f qW 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..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 ]:}| }|| td| }| |}|j q.|D ] }t | qnfddt
j ddD } D ]B | t fdd| D | t
}
| t|
d| qd 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..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 socksexpectedtimesfnames_fnamesrcZsprocZsysconsr 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 tD ]\}|drtt|}t|}| s@t|| t| | || || || qt r~tj
tj t rtj
d S )NZCONN_)r r&
startswithr strisupperr Z assertNotInr r
Z CONN_IDLEZ
CONN_BOUNDr ZCONN_DELETE_TCB)r2 ZintsstrsnamenumZstr_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+ sj _7 g
A