Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/psutil/tests/test_connections.py @ 2:6af9afd405e9 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400
parents 26e78fe6e8c4
children
comparison
equal deleted inserted replaced
1:75ca89e9b81c 2:6af9afd405e9
1 #!/usr/bin/env python3
2
3 # Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.
4 # Use of this source code is governed by a BSD-style license that can be
5 # found in the LICENSE file.
6
7 """Tests for net_connections() and Process.connections() APIs."""
8
9 import contextlib
10 import errno
11 import os
12 import socket
13 import textwrap
14 from contextlib import closing
15 from socket import AF_INET
16 from socket import AF_INET6
17 from socket import SOCK_DGRAM
18 from socket import SOCK_STREAM
19
20 import psutil
21 from psutil import FREEBSD
22 from psutil import LINUX
23 from psutil import MACOS
24 from psutil import NETBSD
25 from psutil import OPENBSD
26 from psutil import POSIX
27 from psutil import SUNOS
28 from psutil import WINDOWS
29 from psutil._common import supports_ipv6
30 from psutil._compat import PY3
31 from psutil.tests import AF_UNIX
32 from psutil.tests import bind_socket
33 from psutil.tests import bind_unix_socket
34 from psutil.tests import check_net_address
35 from psutil.tests import CIRRUS
36 from psutil.tests import create_sockets
37 from psutil.tests import enum
38 from psutil.tests import get_free_port
39 from psutil.tests import HAS_CONNECTIONS_UNIX
40 from psutil.tests import pyrun
41 from psutil.tests import reap_children
42 from psutil.tests import safe_rmpath
43 from psutil.tests import skip_on_access_denied
44 from psutil.tests import SKIP_SYSCONS
45 from psutil.tests import tcp_socketpair
46 from psutil.tests import TESTFN
47 from psutil.tests import TRAVIS
48 from psutil.tests import unittest
49 from psutil.tests import unix_socket_path
50 from psutil.tests import unix_socketpair
51 from psutil.tests import wait_for_file
52
53
54 thisproc = psutil.Process()
55 SOCK_SEQPACKET = getattr(socket, "SOCK_SEQPACKET", object())
56
57
58 class Base(object):
59
60 def setUp(self):
61 safe_rmpath(TESTFN)
62 if not (NETBSD or FREEBSD):
63 # process opens a UNIX socket to /var/log/run.
64 cons = thisproc.connections(kind='all')
65 assert not cons, cons
66
67 def tearDown(self):
68 safe_rmpath(TESTFN)
69 reap_children()
70 if not (FREEBSD or NETBSD):
71 # Make sure we closed all resources.
72 # NetBSD opens a UNIX socket to /var/log/run.
73 cons = thisproc.connections(kind='all')
74 assert not cons, cons
75
76 def compare_procsys_connections(self, pid, proc_cons, kind='all'):
77 """Given a process PID and its list of connections compare
78 those against system-wide connections retrieved via
79 psutil.net_connections.
80 """
81 try:
82 sys_cons = psutil.net_connections(kind=kind)
83 except psutil.AccessDenied:
84 # On MACOS, system-wide connections are retrieved by iterating
85 # over all processes
86 if MACOS:
87 return
88 else:
89 raise
90 # Filter for this proc PID and exlucde PIDs from the tuple.
91 sys_cons = [c[:-1] for c in sys_cons if c.pid == pid]
92 sys_cons.sort()
93 proc_cons.sort()
94 self.assertEqual(proc_cons, sys_cons)
95
96 def check_connection_ntuple(self, conn):
97 """Check validity of a connection namedtuple."""
98 def check_ntuple(conn):
99 has_pid = len(conn) == 7
100 self.assertIn(len(conn), (6, 7))
101 self.assertEqual(conn[0], conn.fd)
102 self.assertEqual(conn[1], conn.family)
103 self.assertEqual(conn[2], conn.type)
104 self.assertEqual(conn[3], conn.laddr)
105 self.assertEqual(conn[4], conn.raddr)
106 self.assertEqual(conn[5], conn.status)
107 if has_pid:
108 self.assertEqual(conn[6], conn.pid)
109
110 def check_family(conn):
111 self.assertIn(conn.family, (AF_INET, AF_INET6, AF_UNIX))
112 if enum is not None:
113 assert isinstance(conn.family, enum.IntEnum), conn
114 else:
115 assert isinstance(conn.family, int), conn
116 if conn.family == AF_INET:
117 # actually try to bind the local socket; ignore IPv6
118 # sockets as their address might be represented as
119 # an IPv4-mapped-address (e.g. "::127.0.0.1")
120 # and that's rejected by bind()
121 s = socket.socket(conn.family, conn.type)
122 with contextlib.closing(s):
123 try:
124 s.bind((conn.laddr[0], 0))
125 except socket.error as err:
126 if err.errno != errno.EADDRNOTAVAIL:
127 raise
128 elif conn.family == AF_UNIX:
129 self.assertEqual(conn.status, psutil.CONN_NONE)
130
131 def check_type(conn):
132 # SOCK_SEQPACKET may happen in case of AF_UNIX socks
133 self.assertIn(conn.type, (SOCK_STREAM, SOCK_DGRAM, SOCK_SEQPACKET))
134 if enum is not None:
135 assert isinstance(conn.type, enum.IntEnum), conn
136 else:
137 assert isinstance(conn.type, int), conn
138 if conn.type == SOCK_DGRAM:
139 self.assertEqual(conn.status, psutil.CONN_NONE)
140
141 def check_addrs(conn):
142 # check IP address and port sanity
143 for addr in (conn.laddr, conn.raddr):
144 if conn.family in (AF_INET, AF_INET6):
145 self.assertIsInstance(addr, tuple)
146 if not addr:
147 continue
148 self.assertIsInstance(addr.port, int)
149 assert 0 <= addr.port <= 65535, addr.port
150 check_net_address(addr.ip, conn.family)
151 elif conn.family == AF_UNIX:
152 self.assertIsInstance(addr, str)
153
154 def check_status(conn):
155 self.assertIsInstance(conn.status, str)
156 valids = [getattr(psutil, x) for x in dir(psutil)
157 if x.startswith('CONN_')]
158 self.assertIn(conn.status, valids)
159 if conn.family in (AF_INET, AF_INET6) and conn.type == SOCK_STREAM:
160 self.assertNotEqual(conn.status, psutil.CONN_NONE)
161 else:
162 self.assertEqual(conn.status, psutil.CONN_NONE)
163
164 check_ntuple(conn)
165 check_family(conn)
166 check_type(conn)
167 check_addrs(conn)
168 check_status(conn)
169
170
171 class TestBase(Base, unittest.TestCase):
172
173 @unittest.skipIf(SKIP_SYSCONS, "requires root")
174 def test_system(self):
175 with create_sockets():
176 for conn in psutil.net_connections(kind='all'):
177 self.check_connection_ntuple(conn)
178
179 def test_process(self):
180 with create_sockets():
181 for conn in psutil.Process().connections(kind='all'):
182 self.check_connection_ntuple(conn)
183
184 def test_invalid_kind(self):
185 self.assertRaises(ValueError, thisproc.connections, kind='???')
186 self.assertRaises(ValueError, psutil.net_connections, kind='???')
187
188
189 class TestUnconnectedSockets(Base, unittest.TestCase):
190 """Tests sockets which are open but not connected to anything."""
191
192 def get_conn_from_sock(self, sock):
193 cons = thisproc.connections(kind='all')
194 smap = dict([(c.fd, c) for c in cons])
195 if NETBSD or FREEBSD:
196 # NetBSD opens a UNIX socket to /var/log/run
197 # so there may be more connections.
198 return smap[sock.fileno()]
199 else:
200 self.assertEqual(len(cons), 1)
201 if cons[0].fd != -1:
202 self.assertEqual(smap[sock.fileno()].fd, sock.fileno())
203 return cons[0]
204
205 def check_socket(self, sock):
206 """Given a socket, makes sure it matches the one obtained
207 via psutil. It assumes this process created one connection
208 only (the one supposed to be checked).
209 """
210 conn = self.get_conn_from_sock(sock)
211 self.check_connection_ntuple(conn)
212
213 # fd, family, type
214 if conn.fd != -1:
215 self.assertEqual(conn.fd, sock.fileno())
216 self.assertEqual(conn.family, sock.family)
217 # see: http://bugs.python.org/issue30204
218 self.assertEqual(
219 conn.type, sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE))
220
221 # local address
222 laddr = sock.getsockname()
223 if not laddr and PY3 and isinstance(laddr, bytes):
224 # See: http://bugs.python.org/issue30205
225 laddr = laddr.decode()
226 if sock.family == AF_INET6:
227 laddr = laddr[:2]
228 if sock.family == AF_UNIX and OPENBSD:
229 # No addresses are set for UNIX sockets on OpenBSD.
230 pass
231 else:
232 self.assertEqual(conn.laddr, laddr)
233
234 # XXX Solaris can't retrieve system-wide UNIX sockets
235 if sock.family == AF_UNIX and HAS_CONNECTIONS_UNIX:
236 cons = thisproc.connections(kind='all')
237 self.compare_procsys_connections(os.getpid(), cons, kind='all')
238 return conn
239
240 def test_tcp_v4(self):
241 addr = ("127.0.0.1", get_free_port())
242 with closing(bind_socket(AF_INET, SOCK_STREAM, addr=addr)) as sock:
243 conn = self.check_socket(sock)
244 assert not conn.raddr
245 self.assertEqual(conn.status, psutil.CONN_LISTEN)
246
247 @unittest.skipIf(not supports_ipv6(), "IPv6 not supported")
248 def test_tcp_v6(self):
249 addr = ("::1", get_free_port())
250 with closing(bind_socket(AF_INET6, SOCK_STREAM, addr=addr)) as sock:
251 conn = self.check_socket(sock)
252 assert not conn.raddr
253 self.assertEqual(conn.status, psutil.CONN_LISTEN)
254
255 def test_udp_v4(self):
256 addr = ("127.0.0.1", get_free_port())
257 with closing(bind_socket(AF_INET, SOCK_DGRAM, addr=addr)) as sock:
258 conn = self.check_socket(sock)
259 assert not conn.raddr
260 self.assertEqual(conn.status, psutil.CONN_NONE)
261
262 @unittest.skipIf(not supports_ipv6(), "IPv6 not supported")
263 def test_udp_v6(self):
264 addr = ("::1", get_free_port())
265 with closing(bind_socket(AF_INET6, SOCK_DGRAM, addr=addr)) as sock:
266 conn = self.check_socket(sock)
267 assert not conn.raddr
268 self.assertEqual(conn.status, psutil.CONN_NONE)
269
270 @unittest.skipIf(not POSIX, 'POSIX only')
271 def test_unix_tcp(self):
272 with unix_socket_path() as name:
273 with closing(bind_unix_socket(name, type=SOCK_STREAM)) as sock:
274 conn = self.check_socket(sock)
275 assert not conn.raddr
276 self.assertEqual(conn.status, psutil.CONN_NONE)
277
278 @unittest.skipIf(not POSIX, 'POSIX only')
279 def test_unix_udp(self):
280 with unix_socket_path() as name:
281 with closing(bind_unix_socket(name, type=SOCK_STREAM)) as sock:
282 conn = self.check_socket(sock)
283 assert not conn.raddr
284 self.assertEqual(conn.status, psutil.CONN_NONE)
285
286
287 class TestConnectedSocket(Base, unittest.TestCase):
288 """Test socket pairs which are are actually connected to
289 each other.
290 """
291
292 # On SunOS, even after we close() it, the server socket stays around
293 # in TIME_WAIT state.
294 @unittest.skipIf(SUNOS, "unreliable on SUONS")
295 def test_tcp(self):
296 addr = ("127.0.0.1", get_free_port())
297 assert not thisproc.connections(kind='tcp4')
298 server, client = tcp_socketpair(AF_INET, addr=addr)
299 try:
300 cons = thisproc.connections(kind='tcp4')
301 self.assertEqual(len(cons), 2)
302 self.assertEqual(cons[0].status, psutil.CONN_ESTABLISHED)
303 self.assertEqual(cons[1].status, psutil.CONN_ESTABLISHED)
304 # May not be fast enough to change state so it stays
305 # commenteed.
306 # client.close()
307 # cons = thisproc.connections(kind='all')
308 # self.assertEqual(len(cons), 1)
309 # self.assertEqual(cons[0].status, psutil.CONN_CLOSE_WAIT)
310 finally:
311 server.close()
312 client.close()
313
314 @unittest.skipIf(not POSIX, 'POSIX only')
315 def test_unix(self):
316 with unix_socket_path() as name:
317 server, client = unix_socketpair(name)
318 try:
319 cons = thisproc.connections(kind='unix')
320 assert not (cons[0].laddr and cons[0].raddr)
321 assert not (cons[1].laddr and cons[1].raddr)
322 if NETBSD or FREEBSD:
323 # On NetBSD creating a UNIX socket will cause
324 # a UNIX connection to /var/run/log.
325 cons = [c for c in cons if c.raddr != '/var/run/log']
326 if CIRRUS:
327 cons = [c for c in cons if c.fd in
328 (server.fileno(), client.fileno())]
329 self.assertEqual(len(cons), 2, msg=cons)
330 if LINUX or FREEBSD or SUNOS:
331 # remote path is never set
332 self.assertEqual(cons[0].raddr, "")
333 self.assertEqual(cons[1].raddr, "")
334 # one local address should though
335 self.assertEqual(name, cons[0].laddr or cons[1].laddr)
336 elif OPENBSD:
337 # No addresses whatsoever here.
338 for addr in (cons[0].laddr, cons[0].raddr,
339 cons[1].laddr, cons[1].raddr):
340 self.assertEqual(addr, "")
341 else:
342 # On other systems either the laddr or raddr
343 # of both peers are set.
344 self.assertEqual(cons[0].laddr or cons[1].laddr, name)
345 self.assertEqual(cons[0].raddr or cons[1].raddr, name)
346 finally:
347 server.close()
348 client.close()
349
350
351 class TestFilters(Base, unittest.TestCase):
352
353 def test_filters(self):
354 def check(kind, families, types):
355 for conn in thisproc.connections(kind=kind):
356 self.assertIn(conn.family, families)
357 self.assertIn(conn.type, types)
358 if not SKIP_SYSCONS:
359 for conn in psutil.net_connections(kind=kind):
360 self.assertIn(conn.family, families)
361 self.assertIn(conn.type, types)
362
363 with create_sockets():
364 check('all',
365 [AF_INET, AF_INET6, AF_UNIX],
366 [SOCK_STREAM, SOCK_DGRAM, SOCK_SEQPACKET])
367 check('inet',
368 [AF_INET, AF_INET6],
369 [SOCK_STREAM, SOCK_DGRAM])
370 check('inet4',
371 [AF_INET],
372 [SOCK_STREAM, SOCK_DGRAM])
373 check('tcp',
374 [AF_INET, AF_INET6],
375 [SOCK_STREAM])
376 check('tcp4',
377 [AF_INET],
378 [SOCK_STREAM])
379 check('tcp6',
380 [AF_INET6],
381 [SOCK_STREAM])
382 check('udp',
383 [AF_INET, AF_INET6],
384 [SOCK_DGRAM])
385 check('udp4',
386 [AF_INET],
387 [SOCK_DGRAM])
388 check('udp6',
389 [AF_INET6],
390 [SOCK_DGRAM])
391 if HAS_CONNECTIONS_UNIX:
392 check('unix',
393 [AF_UNIX],
394 [SOCK_STREAM, SOCK_DGRAM, SOCK_SEQPACKET])
395
396 @skip_on_access_denied(only_if=MACOS)
397 def test_combos(self):
398 def check_conn(proc, conn, family, type, laddr, raddr, status, kinds):
399 all_kinds = ("all", "inet", "inet4", "inet6", "tcp", "tcp4",
400 "tcp6", "udp", "udp4", "udp6")
401 self.check_connection_ntuple(conn)
402 self.assertEqual(conn.family, family)
403 self.assertEqual(conn.type, type)
404 self.assertEqual(conn.laddr, laddr)
405 self.assertEqual(conn.raddr, raddr)
406 self.assertEqual(conn.status, status)
407 for kind in all_kinds:
408 cons = proc.connections(kind=kind)
409 if kind in kinds:
410 assert cons
411 else:
412 assert not cons, cons
413 # compare against system-wide connections
414 # XXX Solaris can't retrieve system-wide UNIX
415 # sockets.
416 if HAS_CONNECTIONS_UNIX:
417 self.compare_procsys_connections(proc.pid, [conn])
418
419 tcp_template = textwrap.dedent("""
420 import socket, time
421 s = socket.socket($family, socket.SOCK_STREAM)
422 s.bind(('$addr', 0))
423 s.listen(5)
424 with open('$testfn', 'w') as f:
425 f.write(str(s.getsockname()[:2]))
426 time.sleep(60)
427 """)
428
429 udp_template = textwrap.dedent("""
430 import socket, time
431 s = socket.socket($family, socket.SOCK_DGRAM)
432 s.bind(('$addr', 0))
433 with open('$testfn', 'w') as f:
434 f.write(str(s.getsockname()[:2]))
435 time.sleep(60)
436 """)
437
438 from string import Template
439 testfile = os.path.basename(TESTFN)
440 tcp4_template = Template(tcp_template).substitute(
441 family=int(AF_INET), addr="127.0.0.1", testfn=testfile)
442 udp4_template = Template(udp_template).substitute(
443 family=int(AF_INET), addr="127.0.0.1", testfn=testfile)
444 tcp6_template = Template(tcp_template).substitute(
445 family=int(AF_INET6), addr="::1", testfn=testfile)
446 udp6_template = Template(udp_template).substitute(
447 family=int(AF_INET6), addr="::1", testfn=testfile)
448
449 # launch various subprocess instantiating a socket of various
450 # families and types to enrich psutil results
451 tcp4_proc = pyrun(tcp4_template)
452 tcp4_addr = eval(wait_for_file(testfile))
453 udp4_proc = pyrun(udp4_template)
454 udp4_addr = eval(wait_for_file(testfile))
455 if supports_ipv6():
456 tcp6_proc = pyrun(tcp6_template)
457 tcp6_addr = eval(wait_for_file(testfile))
458 udp6_proc = pyrun(udp6_template)
459 udp6_addr = eval(wait_for_file(testfile))
460 else:
461 tcp6_proc = None
462 udp6_proc = None
463 tcp6_addr = None
464 udp6_addr = None
465
466 for p in thisproc.children():
467 cons = p.connections()
468 self.assertEqual(len(cons), 1)
469 for conn in cons:
470 # TCP v4
471 if p.pid == tcp4_proc.pid:
472 check_conn(p, conn, AF_INET, SOCK_STREAM, tcp4_addr, (),
473 psutil.CONN_LISTEN,
474 ("all", "inet", "inet4", "tcp", "tcp4"))
475 # UDP v4
476 elif p.pid == udp4_proc.pid:
477 check_conn(p, conn, AF_INET, SOCK_DGRAM, udp4_addr, (),
478 psutil.CONN_NONE,
479 ("all", "inet", "inet4", "udp", "udp4"))
480 # TCP v6
481 elif p.pid == getattr(tcp6_proc, "pid", None):
482 check_conn(p, conn, AF_INET6, SOCK_STREAM, tcp6_addr, (),
483 psutil.CONN_LISTEN,
484 ("all", "inet", "inet6", "tcp", "tcp6"))
485 # UDP v6
486 elif p.pid == getattr(udp6_proc, "pid", None):
487 check_conn(p, conn, AF_INET6, SOCK_DGRAM, udp6_addr, (),
488 psutil.CONN_NONE,
489 ("all", "inet", "inet6", "udp", "udp6"))
490
491 def test_count(self):
492 with create_sockets():
493 # tcp
494 cons = thisproc.connections(kind='tcp')
495 self.assertEqual(len(cons), 2 if supports_ipv6() else 1)
496 for conn in cons:
497 self.assertIn(conn.family, (AF_INET, AF_INET6))
498 self.assertEqual(conn.type, SOCK_STREAM)
499 # tcp4
500 cons = thisproc.connections(kind='tcp4')
501 self.assertEqual(len(cons), 1)
502 self.assertEqual(cons[0].family, AF_INET)
503 self.assertEqual(cons[0].type, SOCK_STREAM)
504 # tcp6
505 if supports_ipv6():
506 cons = thisproc.connections(kind='tcp6')
507 self.assertEqual(len(cons), 1)
508 self.assertEqual(cons[0].family, AF_INET6)
509 self.assertEqual(cons[0].type, SOCK_STREAM)
510 # udp
511 cons = thisproc.connections(kind='udp')
512 self.assertEqual(len(cons), 2 if supports_ipv6() else 1)
513 for conn in cons:
514 self.assertIn(conn.family, (AF_INET, AF_INET6))
515 self.assertEqual(conn.type, SOCK_DGRAM)
516 # udp4
517 cons = thisproc.connections(kind='udp4')
518 self.assertEqual(len(cons), 1)
519 self.assertEqual(cons[0].family, AF_INET)
520 self.assertEqual(cons[0].type, SOCK_DGRAM)
521 # udp6
522 if supports_ipv6():
523 cons = thisproc.connections(kind='udp6')
524 self.assertEqual(len(cons), 1)
525 self.assertEqual(cons[0].family, AF_INET6)
526 self.assertEqual(cons[0].type, SOCK_DGRAM)
527 # inet
528 cons = thisproc.connections(kind='inet')
529 self.assertEqual(len(cons), 4 if supports_ipv6() else 2)
530 for conn in cons:
531 self.assertIn(conn.family, (AF_INET, AF_INET6))
532 self.assertIn(conn.type, (SOCK_STREAM, SOCK_DGRAM))
533 # inet6
534 if supports_ipv6():
535 cons = thisproc.connections(kind='inet6')
536 self.assertEqual(len(cons), 2)
537 for conn in cons:
538 self.assertEqual(conn.family, AF_INET6)
539 self.assertIn(conn.type, (SOCK_STREAM, SOCK_DGRAM))
540 # Skipped on BSD becayse by default the Python process
541 # creates a UNIX socket to '/var/run/log'.
542 if HAS_CONNECTIONS_UNIX and not (FREEBSD or NETBSD):
543 cons = thisproc.connections(kind='unix')
544 self.assertEqual(len(cons), 3)
545 for conn in cons:
546 self.assertEqual(conn.family, AF_UNIX)
547 self.assertIn(conn.type, (SOCK_STREAM, SOCK_DGRAM))
548
549
550 @unittest.skipIf(SKIP_SYSCONS, "requires root")
551 class TestSystemWideConnections(Base, unittest.TestCase):
552 """Tests for net_connections()."""
553
554 def test_it(self):
555 def check(cons, families, types_):
556 for conn in cons:
557 self.assertIn(conn.family, families, msg=conn)
558 if conn.family != AF_UNIX:
559 self.assertIn(conn.type, types_, msg=conn)
560 self.check_connection_ntuple(conn)
561
562 with create_sockets():
563 from psutil._common import conn_tmap
564 for kind, groups in conn_tmap.items():
565 # XXX: SunOS does not retrieve UNIX sockets.
566 if kind == 'unix' and not HAS_CONNECTIONS_UNIX:
567 continue
568 families, types_ = groups
569 cons = psutil.net_connections(kind)
570 self.assertEqual(len(cons), len(set(cons)))
571 check(cons, families, types_)
572
573 # See: https://travis-ci.org/giampaolo/psutil/jobs/237566297
574 @unittest.skipIf(MACOS and TRAVIS, "unreliable on MACOS + TRAVIS")
575 def test_multi_sockets_procs(self):
576 # Creates multiple sub processes, each creating different
577 # sockets. For each process check that proc.connections()
578 # and net_connections() return the same results.
579 # This is done mainly to check whether net_connections()'s
580 # pid is properly set, see:
581 # https://github.com/giampaolo/psutil/issues/1013
582 with create_sockets() as socks:
583 expected = len(socks)
584 pids = []
585 times = 10
586 for i in range(times):
587 fname = os.path.realpath(TESTFN) + str(i)
588 src = textwrap.dedent("""\
589 import time, os
590 from psutil.tests import create_sockets
591 with create_sockets():
592 with open(r'%s', 'w') as f:
593 f.write(str(os.getpid()))
594 time.sleep(60)
595 """ % fname)
596 sproc = pyrun(src)
597 pids.append(sproc.pid)
598 self.addCleanup(safe_rmpath, fname)
599
600 # sync
601 for i in range(times):
602 fname = TESTFN + str(i)
603 wait_for_file(fname)
604
605 syscons = [x for x in psutil.net_connections(kind='all') if x.pid
606 in pids]
607 for pid in pids:
608 self.assertEqual(len([x for x in syscons if x.pid == pid]),
609 expected)
610 p = psutil.Process(pid)
611 self.assertEqual(len(p.connections('all')), expected)
612
613
614 class TestMisc(unittest.TestCase):
615
616 def test_connection_constants(self):
617 ints = []
618 strs = []
619 for name in dir(psutil):
620 if name.startswith('CONN_'):
621 num = getattr(psutil, name)
622 str_ = str(num)
623 assert str_.isupper(), str_
624 self.assertNotIn(str, strs)
625 self.assertNotIn(num, ints)
626 ints.append(num)
627 strs.append(str_)
628 if SUNOS:
629 psutil.CONN_IDLE
630 psutil.CONN_BOUND
631 if WINDOWS:
632 psutil.CONN_DELETE_TCB
633
634
635 if __name__ == '__main__':
636 from psutil.tests.runner import run
637 run(__file__)