<s> """<STR_LIT>""" <EOL> from __future__ import division <EOL> import unittest <EOL> import os <EOL> import sys <EOL> import subprocess <EOL> import time <EOL> import signal <EOL> import types <EOL> import traceback <EOL> import socket <EOL> import warnings <EOL> import atexit <EOL> import errno <EOL> import threading <EOL> import tempfile <EOL> import stat <EOL> import collections <EOL> import datetime <EOL> import psutil <EOL> from psutil . _compat import PY3 , callable , long <EOL> PYTHON = os . path . realpath ( sys . executable ) <EOL> DEVNULL = open ( os . devnull , '<STR_LIT>' ) <EOL> TESTFN = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> POSIX = os . name == '<STR_LIT>' <EOL> LINUX = sys . platform . startswith ( "<STR_LIT>" ) <EOL> WINDOWS = sys . platform . startswith ( "<STR_LIT:win32>" ) <EOL> OSX = sys . platform . startswith ( "<STR_LIT>" ) <EOL> BSD = sys . platform . startswith ( "<STR_LIT>" ) <EOL> _subprocesses_started = set ( ) <EOL> def get_test_subprocess ( cmd = None , stdout = DEVNULL , stderr = DEVNULL , stdin = None , <EOL> wait = False ) : <EOL> """<STR_LIT>""" <EOL> if cmd is None : <EOL> pyline = "<STR_LIT>" <EOL> if wait : <EOL> pyline += "<STR_LIT>" % TESTFN <EOL> pyline += "<STR_LIT>" <EOL> cmd_ = [ PYTHON , "<STR_LIT:-c>" , pyline ] <EOL> else : <EOL> cmd_ = cmd <EOL> sproc = subprocess . Popen ( cmd_ , stdout = stdout , stderr = stderr , stdin = stdin ) <EOL> if wait : <EOL> if cmd is None : <EOL> stop_at = time . time ( ) + <NUM_LIT:3> <EOL> while time . time ( ) > stop_at : <EOL> if os . path . exists ( TESTFN ) : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> else : <EOL> warn ( "<STR_LIT>" ) <EOL> else : <EOL> wait_for_pid ( sproc . pid ) <EOL> _subprocesses_started . add ( sproc . pid ) <EOL> return sproc <EOL> def warn ( msg , warntype = RuntimeWarning ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( msg , warntype ) <EOL> def sh ( cmdline , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) : <EOL> """<STR_LIT>""" <EOL> p = subprocess . Popen ( cmdline , shell = True , stdout = stdout , stderr = stderr ) <EOL> stdout , stderr = p . communicate ( ) <EOL> if p . returncode != <NUM_LIT:0> : <EOL> raise RuntimeError ( stderr ) <EOL> if stderr : <EOL> warn ( stderr ) <EOL> if PY3 : <EOL> stdout = str ( stdout , sys . stdout . encoding ) <EOL> return stdout . strip ( ) <EOL> def which ( program ) : <EOL> """<STR_LIT>""" <EOL> def is_exe ( fpath ) : <EOL> return os . path . isfile ( fpath ) and os . access ( fpath , os . X_OK ) <EOL> fpath , fname = os . path . split ( program ) <EOL> if fpath : <EOL> if is_exe ( program ) : <EOL> return program <EOL> else : <EOL> for path in os . environ [ "<STR_LIT>" ] . split ( os . pathsep ) : <EOL> exe_file = os . path . join ( path , program ) <EOL> if is_exe ( exe_file ) : <EOL> return exe_file <EOL> return None <EOL> def wait_for_pid ( pid , timeout = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> raise_at = time . time ( ) + timeout <EOL> while <NUM_LIT:1> : <EOL> if pid in psutil . get_pid_list ( ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> return <EOL> time . sleep ( <NUM_LIT> ) <EOL> if time . time ( ) >= raise_at : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def reap_children ( search_all = False ) : <EOL> """<STR_LIT>""" <EOL> pids = _subprocesses_started <EOL> if search_all : <EOL> this_process = psutil . Process ( os . getpid ( ) ) <EOL> for p in this_process . get_children ( recursive = True ) : <EOL> pids . add ( p . pid ) <EOL> while pids : <EOL> pid = pids . pop ( ) <EOL> try : <EOL> child = psutil . Process ( pid ) <EOL> child . kill ( ) <EOL> except psutil . NoSuchProcess : <EOL> pass <EOL> except psutil . AccessDenied : <EOL> warn ( "<STR_LIT>" % pid ) <EOL> else : <EOL> child . wait ( timeout = <NUM_LIT:3> ) <EOL> def check_ip_address ( addr , family ) : <EOL> """<STR_LIT>""" <EOL> if not addr : <EOL> return <EOL> ip , port = addr <EOL> assert isinstance ( port , int ) , port <EOL> if family == socket . AF_INET : <EOL> ip = list ( map ( int , ip . split ( '<STR_LIT:.>' ) ) ) <EOL> assert len ( ip ) == <NUM_LIT:4> , ip <EOL> for num in ip : <EOL> assert <NUM_LIT:0> <= num <= <NUM_LIT:255> , ip <EOL> assert <NUM_LIT:0> <= port <= <NUM_LIT> , port <EOL> def safe_remove ( fname ) : <EOL> try : <EOL> os . remove ( fname ) <EOL> except OSError : <EOL> pass <EOL> def call_until ( fun , expr , timeout = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> stop_at = time . time ( ) + timeout <EOL> while time . time ( ) < stop_at : <EOL> ret = fun ( ) <EOL> if eval ( expr ) : <EOL> return ret <EOL> time . sleep ( <NUM_LIT> ) <EOL> raise RuntimeError ( '<STR_LIT>' % ret ) <EOL> def skipIf ( condition , reason = "<STR_LIT>" , warn = False ) : <EOL> """<STR_LIT>""" <EOL> def outer ( fun , * args , ** kwargs ) : <EOL> def inner ( self ) : <EOL> if condition : <EOL> sys . stdout . write ( "<STR_LIT>" ) <EOL> sys . stdout . flush ( ) <EOL> if warn : <EOL> objname = "<STR_LIT>" % ( self . __class__ . __name__ , fun . __name__ ) <EOL> msg = "<STR_LIT>" % objname <EOL> if reason : <EOL> msg += "<STR_LIT>" + repr ( reason ) <EOL> atexit . register ( warn , msg ) <EOL> return <EOL> else : <EOL> return fun ( self , * args , ** kwargs ) <EOL> return inner <EOL> return outer <EOL> def skipUnless ( condition , reason = "<STR_LIT>" , warn = False ) : <EOL> """<STR_LIT>""" <EOL> if not condition : <EOL> return skipIf ( True , reason , warn ) <EOL> return skipIf ( False ) <EOL> def ignore_access_denied ( fun ) : <EOL> """<STR_LIT>""" <EOL> def outer ( fun , * args , ** kwargs ) : <EOL> def inner ( self ) : <EOL> try : <EOL> return fun ( self , * args , ** kwargs ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> return inner <EOL> return outer <EOL> def supports_ipv6 ( ) : <EOL> """<STR_LIT>""" <EOL> if not socket . has_ipv6 or not hasattr ( socket , "<STR_LIT>" ) : <EOL> return False <EOL> sock = None <EOL> try : <EOL> try : <EOL> sock = socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> sock . bind ( ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> except ( socket . error , socket . gaierror ) : <EOL> return False <EOL> else : <EOL> return True <EOL> finally : <EOL> if sock is not None : <EOL> sock . close ( ) <EOL> class ThreadTask ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . _running = False <EOL> self . _interval = None <EOL> self . _flag = threading . Event ( ) <EOL> def __repr__ ( self ) : <EOL> name = self . __class__ . __name__ <EOL> return '<STR_LIT>' % ( name , self . _running , id ( self ) ) <EOL> def start ( self , interval = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> if self . _running : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _interval = interval <EOL> threading . Thread . start ( self ) <EOL> self . _flag . wait ( ) <EOL> def run ( self ) : <EOL> self . _running = True <EOL> self . _flag . set ( ) <EOL> while self . _running : <EOL> time . sleep ( self . _interval ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _running : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _running = False <EOL> self . join ( ) <EOL> class TestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> safe_remove ( TESTFN ) <EOL> def tearDown ( self ) : <EOL> reap_children ( ) <EOL> if not hasattr ( unittest . TestCase , "<STR_LIT>" ) : <EOL> def assertIn ( self , member , container , msg = None ) : <EOL> if member not in container : <EOL> self . fail ( msg or '<STR_LIT>' % ( repr ( member ) , repr ( container ) ) ) <EOL> def assert_eq_w_tol ( self , first , second , tolerance ) : <EOL> difference = abs ( first - second ) <EOL> if difference <= tolerance : <EOL> return <EOL> msg = '<STR_LIT>' % ( first , second , tolerance , difference ) <EOL> raise AssertionError ( msg ) <EOL> def test_process_iter ( self ) : <EOL> assert os . getpid ( ) in [ x . pid for x in psutil . process_iter ( ) ] <EOL> sproc = get_test_subprocess ( ) <EOL> assert sproc . pid in [ x . pid for x in psutil . process_iter ( ) ] <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> assert sproc . pid not in [ x . pid for x in psutil . process_iter ( ) ] <EOL> def test_TOTAL_PHYMEM ( self ) : <EOL> x = psutil . TOTAL_PHYMEM <EOL> assert isinstance ( x , ( int , long ) ) <EOL> assert x > <NUM_LIT:0> <EOL> self . assertEqual ( x , psutil . virtual_memory ( ) . total ) <EOL> def test_BOOT_TIME ( self ) : <EOL> x = psutil . BOOT_TIME <EOL> assert isinstance ( x , float ) <EOL> assert x > <NUM_LIT:0> <EOL> assert x < time . time ( ) , x <EOL> def test_NUM_CPUS ( self ) : <EOL> self . assertEqual ( psutil . NUM_CPUS , len ( psutil . cpu_times ( percpu = True ) ) ) <EOL> assert psutil . NUM_CPUS >= <NUM_LIT:1> , psutil . NUM_CPUS <EOL> @ skipUnless ( POSIX ) <EOL> def test_PAGESIZE ( self ) : <EOL> import resource <EOL> self . assertEqual ( os . sysconf ( "<STR_LIT>" ) , resource . getpagesize ( ) ) <EOL> def test_deprecated_apis ( self ) : <EOL> warnings . filterwarnings ( "<STR_LIT:error>" ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> try : <EOL> self . assertRaises ( DeprecationWarning , psutil . virtmem_usage ) <EOL> self . assertRaises ( DeprecationWarning , psutil . used_phymem ) <EOL> self . assertRaises ( DeprecationWarning , psutil . avail_phymem ) <EOL> self . assertRaises ( DeprecationWarning , psutil . total_virtmem ) <EOL> self . assertRaises ( DeprecationWarning , psutil . used_virtmem ) <EOL> self . assertRaises ( DeprecationWarning , psutil . avail_virtmem ) <EOL> self . assertRaises ( DeprecationWarning , psutil . phymem_usage ) <EOL> self . assertRaises ( DeprecationWarning , psutil . get_process_list ) <EOL> self . assertRaises ( DeprecationWarning , psutil . get_process_list ) <EOL> if LINUX : <EOL> self . assertRaises ( DeprecationWarning , psutil . phymem_buffers ) <EOL> self . assertRaises ( DeprecationWarning , psutil . cached_phymem ) <EOL> try : <EOL> p . nice <EOL> except DeprecationWarning : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> finally : <EOL> warnings . resetwarnings ( ) <EOL> def test_deprecated_apis_retval ( self ) : <EOL> warnings . filterwarnings ( "<STR_LIT:ignore>" ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> try : <EOL> self . assertEqual ( psutil . total_virtmem ( ) , psutil . swap_memory ( ) . total ) <EOL> self . assertEqual ( psutil . get_process_list ( ) , list ( psutil . process_iter ( ) ) ) <EOL> self . assertEqual ( p . nice , p . get_nice ( ) ) <EOL> finally : <EOL> warnings . resetwarnings ( ) <EOL> def test_virtual_memory ( self ) : <EOL> mem = psutil . virtual_memory ( ) <EOL> assert mem . total > <NUM_LIT:0> , mem <EOL> assert mem . available > <NUM_LIT:0> , mem <EOL> assert <NUM_LIT:0> <= mem . percent <= <NUM_LIT:100> , mem <EOL> assert mem . used > <NUM_LIT:0> , mem <EOL> assert mem . free >= <NUM_LIT:0> , mem <EOL> for name in mem . _fields : <EOL> if name != '<STR_LIT>' : <EOL> value = getattr ( mem , name ) <EOL> if not value >= <NUM_LIT:0> : <EOL> self . fail ( "<STR_LIT>" % ( name , value ) ) <EOL> if value > mem . total : <EOL> self . fail ( "<STR_LIT>" % ( name , mem . total , name , value ) ) <EOL> def test_swap_memory ( self ) : <EOL> mem = psutil . swap_memory ( ) <EOL> assert mem . total > <NUM_LIT:0> , mem <EOL> assert mem . used >= <NUM_LIT:0> , mem <EOL> assert mem . free > <NUM_LIT:0> , mem <EOL> assert <NUM_LIT:0> <= mem . percent <= <NUM_LIT:100> , mem <EOL> assert mem . sin >= <NUM_LIT:0> , mem <EOL> assert mem . sout >= <NUM_LIT:0> , mem <EOL> def test_pid_exists ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> assert psutil . pid_exists ( sproc . pid ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> self . assertFalse ( psutil . pid_exists ( sproc . pid ) ) <EOL> self . assertFalse ( psutil . pid_exists ( - <NUM_LIT:1> ) ) <EOL> def test_pid_exists_2 ( self ) : <EOL> reap_children ( ) <EOL> pids = psutil . get_pid_list ( ) <EOL> for pid in pids : <EOL> try : <EOL> assert psutil . pid_exists ( pid ) <EOL> except AssertionError : <EOL> time . sleep ( <NUM_LIT> ) <EOL> if pid in psutil . get_pid_list ( ) : <EOL> self . fail ( pid ) <EOL> pids = range ( max ( pids ) + <NUM_LIT> , max ( pids ) + <NUM_LIT> ) <EOL> for pid in pids : <EOL> self . assertFalse ( psutil . pid_exists ( pid ) ) <EOL> def test_get_pid_list ( self ) : <EOL> plist = [ x . pid for x in psutil . process_iter ( ) ] <EOL> pidlist = psutil . get_pid_list ( ) <EOL> self . assertEqual ( plist . sort ( ) , pidlist . sort ( ) ) <EOL> self . assertEqual ( len ( pidlist ) , len ( set ( pidlist ) ) ) <EOL> def test_test ( self ) : <EOL> stdout = sys . stdout <EOL> sys . stdout = DEVNULL <EOL> try : <EOL> psutil . test ( ) <EOL> finally : <EOL> sys . stdout = stdout <EOL> def test_sys_cpu_times ( self ) : <EOL> total = <NUM_LIT:0> <EOL> times = psutil . cpu_times ( ) <EOL> sum ( times ) <EOL> for cp_time in times : <EOL> assert isinstance ( cp_time , float ) <EOL> assert cp_time >= <NUM_LIT:0.0> , cp_time <EOL> total += cp_time <EOL> self . assertEqual ( total , sum ( times ) ) <EOL> str ( times ) <EOL> def test_sys_cpu_times2 ( self ) : <EOL> t1 = sum ( psutil . cpu_times ( ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> t2 = sum ( psutil . cpu_times ( ) ) <EOL> difference = t2 - t1 <EOL> if not difference >= <NUM_LIT> : <EOL> self . fail ( "<STR_LIT>" % difference ) <EOL> def test_sys_per_cpu_times ( self ) : <EOL> for times in psutil . cpu_times ( percpu = True ) : <EOL> total = <NUM_LIT:0> <EOL> sum ( times ) <EOL> for cp_time in times : <EOL> assert isinstance ( cp_time , float ) <EOL> assert cp_time >= <NUM_LIT:0.0> , cp_time <EOL> total += cp_time <EOL> self . assertEqual ( total , sum ( times ) ) <EOL> str ( times ) <EOL> def test_sys_per_cpu_times2 ( self ) : <EOL> tot1 = psutil . cpu_times ( percpu = True ) <EOL> stop_at = time . time ( ) + <NUM_LIT:0.1> <EOL> while <NUM_LIT:1> : <EOL> if time . time ( ) >= stop_at : <EOL> break <EOL> tot2 = psutil . cpu_times ( percpu = True ) <EOL> for t1 , t2 in zip ( tot1 , tot2 ) : <EOL> t1 , t2 = sum ( t1 ) , sum ( t2 ) <EOL> difference = t2 - t1 <EOL> if difference >= <NUM_LIT> : <EOL> return <EOL> self . fail ( ) <EOL> def test_sys_cpu_percent ( self ) : <EOL> psutil . cpu_percent ( interval = <NUM_LIT> ) <EOL> psutil . cpu_percent ( interval = <NUM_LIT> ) <EOL> for x in range ( <NUM_LIT:1000> ) : <EOL> percent = psutil . cpu_percent ( interval = None ) <EOL> assert isinstance ( percent , float ) <EOL> assert percent >= <NUM_LIT:0.0> , percent <EOL> assert percent <= <NUM_LIT> , percent <EOL> def test_sys_per_cpu_percent ( self ) : <EOL> psutil . cpu_percent ( interval = <NUM_LIT> , percpu = True ) <EOL> psutil . cpu_percent ( interval = <NUM_LIT> , percpu = True ) <EOL> for x in range ( <NUM_LIT:1000> ) : <EOL> percents = psutil . cpu_percent ( interval = None , percpu = True ) <EOL> for percent in percents : <EOL> assert isinstance ( percent , float ) <EOL> assert percent >= <NUM_LIT:0.0> , percent <EOL> assert percent <= <NUM_LIT> , percent <EOL> def test_disk_usage ( self ) : <EOL> usage = psutil . disk_usage ( os . getcwd ( ) ) <EOL> assert usage . total > <NUM_LIT:0> , usage <EOL> assert usage . used > <NUM_LIT:0> , usage <EOL> assert usage . free > <NUM_LIT:0> , usage <EOL> assert usage . total > usage . used , usage <EOL> assert usage . total > usage . free , usage <EOL> assert <NUM_LIT:0> <= usage . percent <= <NUM_LIT:100> , usage . percent <EOL> fname = tempfile . mktemp ( ) <EOL> try : <EOL> psutil . disk_usage ( fname ) <EOL> except OSError : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . args [ <NUM_LIT:0> ] != errno . ENOENT : <EOL> raise <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_disk_partitions ( self ) : <EOL> for disk in psutil . disk_partitions ( all = False ) : <EOL> assert os . path . exists ( disk . device ) , disk <EOL> assert os . path . isdir ( disk . mountpoint ) , disk <EOL> assert disk . fstype , disk <EOL> assert isinstance ( disk . opts , str ) <EOL> for disk in psutil . disk_partitions ( all = True ) : <EOL> if not WINDOWS : <EOL> try : <EOL> os . stat ( disk . mountpoint ) <EOL> except OSError : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . errno not in ( errno . EPERM , errno . EACCES ) : <EOL> raise <EOL> else : <EOL> assert os . path . isdir ( disk . mountpoint ) , disk . mountpoint <EOL> assert isinstance ( disk . fstype , str ) <EOL> assert isinstance ( disk . opts , str ) <EOL> def find_mount_point ( path ) : <EOL> path = os . path . abspath ( path ) <EOL> while not os . path . ismount ( path ) : <EOL> path = os . path . dirname ( path ) <EOL> return path <EOL> mount = find_mount_point ( __file__ ) <EOL> mounts = [ x . mountpoint for x in psutil . disk_partitions ( all = True ) ] <EOL> self . assertIn ( mount , mounts ) <EOL> psutil . disk_usage ( mount ) <EOL> def test_network_io_counters ( self ) : <EOL> def check_ntuple ( nt ) : <EOL> self . assertEqual ( nt [ <NUM_LIT:0> ] , nt . bytes_sent ) <EOL> self . assertEqual ( nt [ <NUM_LIT:1> ] , nt . bytes_recv ) <EOL> self . assertEqual ( nt [ <NUM_LIT:2> ] , nt . packets_sent ) <EOL> self . assertEqual ( nt [ <NUM_LIT:3> ] , nt . packets_recv ) <EOL> self . assertEqual ( nt [ <NUM_LIT:4> ] , nt . errin ) <EOL> self . assertEqual ( nt [ <NUM_LIT:5> ] , nt . errout ) <EOL> self . assertEqual ( nt [ <NUM_LIT:6> ] , nt . dropin ) <EOL> self . assertEqual ( nt [ <NUM_LIT:7> ] , nt . dropout ) <EOL> assert nt . bytes_sent >= <NUM_LIT:0> , nt <EOL> assert nt . bytes_recv >= <NUM_LIT:0> , nt <EOL> assert nt . packets_sent >= <NUM_LIT:0> , nt <EOL> assert nt . packets_recv >= <NUM_LIT:0> , nt <EOL> assert nt . errin >= <NUM_LIT:0> , nt <EOL> assert nt . errout >= <NUM_LIT:0> , nt <EOL> assert nt . dropin >= <NUM_LIT:0> , nt <EOL> assert nt . dropout >= <NUM_LIT:0> , nt <EOL> ret = psutil . network_io_counters ( pernic = False ) <EOL> check_ntuple ( ret ) <EOL> ret = psutil . network_io_counters ( pernic = True ) <EOL> assert ret != [ ] <EOL> for key in ret : <EOL> assert key <EOL> check_ntuple ( ret [ key ] ) <EOL> def test_disk_io_counters ( self ) : <EOL> def check_ntuple ( nt ) : <EOL> self . assertEqual ( nt [ <NUM_LIT:0> ] , nt . read_count ) <EOL> self . assertEqual ( nt [ <NUM_LIT:1> ] , nt . write_count ) <EOL> self . assertEqual ( nt [ <NUM_LIT:2> ] , nt . read_bytes ) <EOL> self . assertEqual ( nt [ <NUM_LIT:3> ] , nt . write_bytes ) <EOL> self . assertEqual ( nt [ <NUM_LIT:4> ] , nt . read_time ) <EOL> self . assertEqual ( nt [ <NUM_LIT:5> ] , nt . write_time ) <EOL> assert nt . read_count >= <NUM_LIT:0> , nt <EOL> assert nt . write_count >= <NUM_LIT:0> , nt <EOL> assert nt . read_bytes >= <NUM_LIT:0> , nt <EOL> assert nt . write_bytes >= <NUM_LIT:0> , nt <EOL> assert nt . read_time >= <NUM_LIT:0> , nt <EOL> assert nt . write_time >= <NUM_LIT:0> , nt <EOL> ret = psutil . disk_io_counters ( perdisk = False ) <EOL> check_ntuple ( ret ) <EOL> ret = psutil . disk_io_counters ( perdisk = True ) <EOL> for key in ret : <EOL> assert key , key <EOL> check_ntuple ( ret [ key ] ) <EOL> def test_get_users ( self ) : <EOL> users = psutil . get_users ( ) <EOL> assert users <EOL> for user in users : <EOL> assert user . name , user <EOL> user . terminal <EOL> user . host <EOL> assert user . started > <NUM_LIT:0.0> , user <EOL> datetime . datetime . fromtimestamp ( user . started ) <EOL> def test_kill ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> test_pid = sproc . pid <EOL> p = psutil . Process ( test_pid ) <EOL> name = p . name <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> self . assertFalse ( psutil . pid_exists ( test_pid ) and name == PYTHON ) <EOL> def test_terminate ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> test_pid = sproc . pid <EOL> p = psutil . Process ( test_pid ) <EOL> name = p . name <EOL> p . terminate ( ) <EOL> p . wait ( ) <EOL> self . assertFalse ( psutil . pid_exists ( test_pid ) and name == PYTHON ) <EOL> def test_send_signal ( self ) : <EOL> if POSIX : <EOL> sig = signal . SIGKILL <EOL> else : <EOL> sig = signal . SIGTERM <EOL> sproc = get_test_subprocess ( ) <EOL> test_pid = sproc . pid <EOL> p = psutil . Process ( test_pid ) <EOL> name = p . name <EOL> p . send_signal ( sig ) <EOL> p . wait ( ) <EOL> self . assertFalse ( psutil . pid_exists ( test_pid ) and name == PYTHON ) <EOL> def test_wait ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . kill ( ) <EOL> code = p . wait ( ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertEqual ( code , signal . SIGKILL ) <EOL> else : <EOL> self . assertEqual ( code , <NUM_LIT:0> ) <EOL> self . assertFalse ( p . is_running ( ) ) <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . terminate ( ) <EOL> code = p . wait ( ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertEqual ( code , signal . SIGTERM ) <EOL> else : <EOL> self . assertEqual ( code , <NUM_LIT:0> ) <EOL> self . assertFalse ( p . is_running ( ) ) <EOL> code = "<STR_LIT>" <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , code ] ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:5> ) <EOL> self . assertFalse ( p . is_running ( ) ) <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , code ] ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:5> ) <EOL> self . assertIn ( p . wait ( ) , ( <NUM_LIT:5> , None ) ) <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . name <EOL> self . assertRaises ( psutil . TimeoutExpired , p . wait , <NUM_LIT> ) <EOL> self . assertRaises ( ValueError , p . wait , - <NUM_LIT:1> ) <EOL> @ skipUnless ( POSIX ) <EOL> def test_wait_non_children ( self ) : <EOL> code = "<STR_LIT>" <EOL> code += "<STR_LIT>" <EOL> code += "<STR_LIT>" % PYTHON <EOL> code += "<STR_LIT>" <EOL> code += "<STR_LIT>" <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , code ] , stdout = subprocess . PIPE ) <EOL> grandson_pid = int ( sproc . stdout . read ( ) ) <EOL> grandson_proc = psutil . Process ( grandson_pid ) <EOL> try : <EOL> self . assertRaises ( psutil . TimeoutExpired , grandson_proc . wait , <NUM_LIT> ) <EOL> grandson_proc . kill ( ) <EOL> ret = grandson_proc . wait ( ) <EOL> self . assertEqual ( ret , None ) <EOL> finally : <EOL> if grandson_proc . is_running ( ) : <EOL> grandson_proc . kill ( ) <EOL> grandson_proc . wait ( ) <EOL> def test_wait_timeout_0 ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertRaises ( psutil . TimeoutExpired , p . wait , <NUM_LIT:0> ) <EOL> p . kill ( ) <EOL> stop_at = time . time ( ) + <NUM_LIT:2> <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> code = p . wait ( <NUM_LIT:0> ) <EOL> except psutil . TimeoutExpired : <EOL> if time . time ( ) >= stop_at : <EOL> raise <EOL> else : <EOL> break <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertEqual ( code , signal . SIGKILL ) <EOL> else : <EOL> self . assertEqual ( code , <NUM_LIT:0> ) <EOL> self . assertFalse ( p . is_running ( ) ) <EOL> def test_cpu_percent ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> p . get_cpu_percent ( interval = <NUM_LIT> ) <EOL> p . get_cpu_percent ( interval = <NUM_LIT> ) <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> percent = p . get_cpu_percent ( interval = None ) <EOL> assert isinstance ( percent , float ) <EOL> assert percent >= <NUM_LIT:0.0> , percent <EOL> if os . name != '<STR_LIT>' : <EOL> assert percent <= <NUM_LIT> , percent <EOL> else : <EOL> assert percent >= <NUM_LIT:0.0> , percent <EOL> def test_cpu_times ( self ) : <EOL> times = psutil . Process ( os . getpid ( ) ) . get_cpu_times ( ) <EOL> assert ( times . user > <NUM_LIT:0.0> ) or ( times . system > <NUM_LIT:0.0> ) , times <EOL> time . strftime ( "<STR_LIT>" , time . localtime ( times . user ) ) <EOL> time . strftime ( "<STR_LIT>" , time . localtime ( times . system ) ) <EOL> @ skipUnless ( sys . version_info > ( <NUM_LIT:2> , <NUM_LIT:6> , <NUM_LIT:1> ) and not OSX ) <EOL> def test_cpu_times2 ( self ) : <EOL> user_time , kernel_time = psutil . Process ( os . getpid ( ) ) . get_cpu_times ( ) <EOL> utime , ktime = os . times ( ) [ : <NUM_LIT:2> ] <EOL> if ( max ( [ user_time , utime ] ) - min ( [ user_time , utime ] ) ) > <NUM_LIT:0.1> : <EOL> self . fail ( "<STR_LIT>" % ( utime , user_time ) ) <EOL> if ( max ( [ kernel_time , ktime ] ) - min ( [ kernel_time , ktime ] ) ) > <NUM_LIT:0.1> : <EOL> self . fail ( "<STR_LIT>" % ( ktime , kernel_time ) ) <EOL> def test_create_time ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> now = time . time ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> create_time = p . create_time <EOL> difference = abs ( create_time - now ) <EOL> if difference > <NUM_LIT:2> : <EOL> self . fail ( "<STR_LIT>" <EOL> % ( now , create_time , difference ) ) <EOL> time . strftime ( "<STR_LIT>" , time . localtime ( p . create_time ) ) <EOL> @ skipIf ( WINDOWS ) <EOL> def test_terminal ( self ) : <EOL> tty = sh ( '<STR_LIT>' ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertEqual ( p . terminal , tty ) <EOL> @ skipIf ( OSX , warn = False ) <EOL> def test_get_io_counters ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> io1 = p . get_io_counters ( ) <EOL> f = open ( PYTHON , '<STR_LIT:rb>' ) <EOL> f . read ( ) <EOL> f . close ( ) <EOL> io2 = p . get_io_counters ( ) <EOL> if not BSD : <EOL> assert io2 . read_count > io1 . read_count , ( io1 , io2 ) <EOL> self . assertEqual ( io2 . write_count , io1 . write_count ) <EOL> assert io2 . read_bytes >= io1 . read_bytes , ( io1 , io2 ) <EOL> assert io2 . write_bytes >= io1 . write_bytes , ( io1 , io2 ) <EOL> io1 = p . get_io_counters ( ) <EOL> f = tempfile . TemporaryFile ( ) <EOL> if PY3 : <EOL> f . write ( bytes ( "<STR_LIT:x>" * <NUM_LIT> , '<STR_LIT:ascii>' ) ) <EOL> else : <EOL> f . write ( "<STR_LIT:x>" * <NUM_LIT> ) <EOL> f . close ( ) <EOL> io2 = p . get_io_counters ( ) <EOL> assert io2 . write_count >= io1 . write_count , ( io1 , io2 ) <EOL> assert io2 . write_bytes >= io1 . write_bytes , ( io1 , io2 ) <EOL> assert io2 . read_count >= io1 . read_count , ( io1 , io2 ) <EOL> assert io2 . read_bytes >= io1 . read_bytes , ( io1 , io2 ) <EOL> @ skipUnless ( LINUX ) <EOL> def test_get_set_ionice ( self ) : <EOL> from psutil import ( IOPRIO_CLASS_NONE , IOPRIO_CLASS_RT , IOPRIO_CLASS_BE , <EOL> IOPRIO_CLASS_IDLE ) <EOL> self . assertEqual ( IOPRIO_CLASS_NONE , <NUM_LIT:0> ) <EOL> self . assertEqual ( IOPRIO_CLASS_RT , <NUM_LIT:1> ) <EOL> self . assertEqual ( IOPRIO_CLASS_BE , <NUM_LIT:2> ) <EOL> self . assertEqual ( IOPRIO_CLASS_IDLE , <NUM_LIT:3> ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> try : <EOL> p . set_ionice ( <NUM_LIT:2> ) <EOL> ioclass , value = p . get_ionice ( ) <EOL> self . assertEqual ( ioclass , <NUM_LIT:2> ) <EOL> self . assertEqual ( value , <NUM_LIT:4> ) <EOL> p . set_ionice ( <NUM_LIT:3> ) <EOL> ioclass , value = p . get_ionice ( ) <EOL> self . assertEqual ( ioclass , <NUM_LIT:3> ) <EOL> self . assertEqual ( value , <NUM_LIT:0> ) <EOL> p . set_ionice ( <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> ioclass , value = p . get_ionice ( ) <EOL> self . assertEqual ( ioclass , <NUM_LIT:2> ) <EOL> self . assertEqual ( value , <NUM_LIT:0> ) <EOL> p . set_ionice ( <NUM_LIT:2> , <NUM_LIT:7> ) <EOL> ioclass , value = p . get_ionice ( ) <EOL> self . assertEqual ( ioclass , <NUM_LIT:2> ) <EOL> self . assertEqual ( value , <NUM_LIT:7> ) <EOL> self . assertRaises ( ValueError , p . set_ionice , <NUM_LIT:2> , <NUM_LIT:10> ) <EOL> finally : <EOL> p . set_ionice ( IOPRIO_CLASS_NONE ) <EOL> def test_get_num_threads ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> step1 = p . get_num_threads ( ) <EOL> thread = ThreadTask ( ) <EOL> thread . start ( ) <EOL> try : <EOL> step2 = p . get_num_threads ( ) <EOL> self . assertEqual ( step2 , step1 + <NUM_LIT:1> ) <EOL> thread . stop ( ) <EOL> finally : <EOL> if thread . _running : <EOL> thread . stop ( ) <EOL> @ skipUnless ( WINDOWS ) <EOL> def test_get_num_handles ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> assert p . get_num_handles ( ) > <NUM_LIT:0> <EOL> def test_get_threads ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> step1 = p . get_threads ( ) <EOL> thread = ThreadTask ( ) <EOL> thread . start ( ) <EOL> try : <EOL> step2 = p . get_threads ( ) <EOL> self . assertEqual ( len ( step2 ) , len ( step1 ) + <NUM_LIT:1> ) <EOL> if LINUX : <EOL> self . assertEqual ( step2 [ <NUM_LIT:0> ] . id , os . getpid ( ) ) <EOL> athread = step2 [ <NUM_LIT:0> ] <EOL> self . assertEqual ( athread . id , athread [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( athread . user_time , athread [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( athread . system_time , athread [ <NUM_LIT:2> ] ) <EOL> thread . stop ( ) <EOL> finally : <EOL> if thread . _running : <EOL> thread . stop ( ) <EOL> def test_get_memory_info ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> rss1 , vms1 = p . get_memory_info ( ) <EOL> percent1 = p . get_memory_percent ( ) <EOL> assert rss1 > <NUM_LIT:0> <EOL> assert vms1 > <NUM_LIT:0> <EOL> memarr = [ None ] * <NUM_LIT> <EOL> rss2 , vms2 = p . get_memory_info ( ) <EOL> percent2 = p . get_memory_percent ( ) <EOL> assert rss2 > rss1 <EOL> assert vms2 >= vms1 <EOL> assert percent2 > percent1 <EOL> del memarr <EOL> def test_get_memory_maps ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> maps = p . get_memory_maps ( ) <EOL> paths = [ x for x in maps ] <EOL> self . assertEqual ( len ( paths ) , len ( set ( paths ) ) ) <EOL> ext_maps = p . get_memory_maps ( grouped = False ) <EOL> for nt in maps : <EOL> if not nt . path . startswith ( '<STR_LIT:[>' ) : <EOL> assert os . path . isabs ( nt . path ) , nt . path <EOL> if POSIX : <EOL> assert os . path . exists ( nt . path ) , nt . path <EOL> else : <EOL> if '<STR_LIT>' not in os . path . basename ( nt . path ) : <EOL> assert os . path . exists ( nt . path ) , nt . path <EOL> for nt in ext_maps : <EOL> for fname in nt . _fields : <EOL> value = getattr ( nt , fname ) <EOL> if fname == '<STR_LIT:path>' : <EOL> continue <EOL> elif fname in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> assert value , value <EOL> else : <EOL> assert isinstance ( value , ( int , long ) ) <EOL> assert value >= <NUM_LIT:0> , value <EOL> def test_get_memory_percent ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> assert p . get_memory_percent ( ) > <NUM_LIT:0.0> <EOL> def test_pid ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> self . assertEqual ( psutil . Process ( sproc . pid ) . pid , sproc . pid ) <EOL> def test_is_running ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> assert p . is_running ( ) <EOL> assert p . is_running ( ) <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> assert not p . is_running ( ) <EOL> assert not p . is_running ( ) <EOL> def test_exe ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> exe = psutil . Process ( sproc . pid ) . exe <EOL> try : <EOL> self . assertEqual ( exe , PYTHON ) <EOL> except AssertionError : <EOL> ver = "<STR_LIT>" % ( sys . version_info [ <NUM_LIT:0> ] , sys . version_info [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( exe . replace ( ver , '<STR_LIT>' ) , PYTHON . replace ( ver , '<STR_LIT>' ) ) <EOL> def test_cmdline ( self ) : <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT>" ] , wait = True ) <EOL> self . assertEqual ( psutil . Process ( sproc . pid ) . cmdline , [ PYTHON , "<STR_LIT>" ] ) <EOL> def test_name ( self ) : <EOL> sproc = get_test_subprocess ( PYTHON , wait = True ) <EOL> pyexe = os . path . basename ( os . path . realpath ( sys . executable ) ) . lower ( ) <EOL> self . assertEqual ( psutil . Process ( sproc . pid ) . name . lower ( ) , pyexe ) <EOL> if os . name == '<STR_LIT>' : <EOL> def test_uids ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> real , effective , saved = p . uids <EOL> self . assertEqual ( real , os . getuid ( ) ) <EOL> self . assertEqual ( effective , os . geteuid ( ) ) <EOL> if hasattr ( os , "<STR_LIT>" ) : <EOL> self . assertEqual ( saved , os . getresuid ( ) [ <NUM_LIT:2> ] ) <EOL> def test_gids ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> real , effective , saved = p . gids <EOL> self . assertEqual ( real , os . getgid ( ) ) <EOL> self . assertEqual ( effective , os . getegid ( ) ) <EOL> if hasattr ( os , "<STR_LIT>" ) : <EOL> self . assertEqual ( saved , os . getresgid ( ) [ <NUM_LIT:2> ] ) <EOL> def test_nice ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertRaises ( TypeError , p . set_nice , "<STR_LIT:str>" ) <EOL> try : <EOL> try : <EOL> first_nice = p . get_nice ( ) <EOL> p . set_nice ( <NUM_LIT:1> ) <EOL> self . assertEqual ( p . get_nice ( ) , <NUM_LIT:1> ) <EOL> if not OSX : <EOL> p . set_nice ( <NUM_LIT:0> ) <EOL> self . assertEqual ( p . get_nice ( ) , <NUM_LIT:0> ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> finally : <EOL> try : <EOL> p . set_nice ( first_nice ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> if os . name == '<STR_LIT>' : <EOL> def test_nice ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertRaises ( TypeError , p . set_nice , "<STR_LIT:str>" ) <EOL> try : <EOL> self . assertEqual ( p . get_nice ( ) , psutil . NORMAL_PRIORITY_CLASS ) <EOL> p . set_nice ( psutil . HIGH_PRIORITY_CLASS ) <EOL> self . assertEqual ( p . get_nice ( ) , psutil . HIGH_PRIORITY_CLASS ) <EOL> p . set_nice ( psutil . NORMAL_PRIORITY_CLASS ) <EOL> self . assertEqual ( p . get_nice ( ) , psutil . NORMAL_PRIORITY_CLASS ) <EOL> finally : <EOL> p . set_nice ( psutil . NORMAL_PRIORITY_CLASS ) <EOL> def test_status ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertEqual ( p . status , psutil . STATUS_RUNNING ) <EOL> self . assertEqual ( str ( p . status ) , "<STR_LIT>" ) <EOL> def test_status_constants ( self ) : <EOL> self . assertTrue ( psutil . STATUS_RUNNING == <NUM_LIT:0> ) <EOL> self . assertTrue ( psutil . STATUS_RUNNING == long ( <NUM_LIT:0> ) ) <EOL> self . assertTrue ( psutil . STATUS_RUNNING == '<STR_LIT>' ) <EOL> self . assertFalse ( psutil . STATUS_RUNNING == <NUM_LIT:1> ) <EOL> self . assertFalse ( psutil . STATUS_RUNNING == '<STR_LIT>' ) <EOL> self . assertFalse ( psutil . STATUS_RUNNING != <NUM_LIT:0> ) <EOL> self . assertFalse ( psutil . STATUS_RUNNING != '<STR_LIT>' ) <EOL> self . assertTrue ( psutil . STATUS_RUNNING != <NUM_LIT:1> ) <EOL> self . assertTrue ( psutil . STATUS_RUNNING != '<STR_LIT>' ) <EOL> def test_username ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> if POSIX : <EOL> import pwd <EOL> self . assertEqual ( p . username , pwd . getpwuid ( os . getuid ( ) ) . pw_name ) <EOL> elif WINDOWS : <EOL> expected_username = os . environ [ '<STR_LIT>' ] <EOL> expected_domain = os . environ [ '<STR_LIT>' ] <EOL> domain , username = p . username . split ( '<STR_LIT:\\>' ) <EOL> self . assertEqual ( domain , expected_domain ) <EOL> self . assertEqual ( username , expected_username ) <EOL> else : <EOL> p . username <EOL> @ skipIf ( not hasattr ( psutil . Process , "<STR_LIT>" ) ) <EOL> def test_getcwd ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertEqual ( p . getcwd ( ) , os . getcwd ( ) ) <EOL> @ skipIf ( not hasattr ( psutil . Process , "<STR_LIT>" ) ) <EOL> def test_getcwd_2 ( self ) : <EOL> cmd = [ PYTHON , "<STR_LIT:-c>" , "<STR_LIT>" ] <EOL> sproc = get_test_subprocess ( cmd , wait = True ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> call_until ( p . getcwd , "<STR_LIT>" , timeout = <NUM_LIT:1> ) <EOL> @ skipIf ( not hasattr ( psutil . Process , "<STR_LIT>" ) ) <EOL> def test_cpu_affinity ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> initial = p . get_cpu_affinity ( ) <EOL> all_cpus = list ( range ( len ( psutil . cpu_percent ( percpu = True ) ) ) ) <EOL> for n in all_cpus : <EOL> p . set_cpu_affinity ( [ n ] ) <EOL> self . assertEqual ( p . get_cpu_affinity ( ) , [ n ] ) <EOL> p . set_cpu_affinity ( all_cpus ) <EOL> self . assertEqual ( p . get_cpu_affinity ( ) , all_cpus ) <EOL> p . set_cpu_affinity ( initial ) <EOL> invalid_cpu = [ len ( psutil . cpu_times ( percpu = True ) ) + <NUM_LIT:10> ] <EOL> self . assertRaises ( ValueError , p . set_cpu_affinity , invalid_cpu ) <EOL> def test_get_open_files ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> files = p . get_open_files ( ) <EOL> self . assertFalse ( TESTFN in files ) <EOL> f = open ( TESTFN , '<STR_LIT:w>' ) <EOL> call_until ( p . get_open_files , "<STR_LIT>" % len ( files ) ) <EOL> filenames = [ x . path for x in p . get_open_files ( ) ] <EOL> self . assertIn ( TESTFN , filenames ) <EOL> f . close ( ) <EOL> for file in filenames : <EOL> assert os . path . isfile ( file ) , file <EOL> cmdline = "<STR_LIT>" % TESTFN <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , cmdline ] , wait = True ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> filenames = [ x . path for x in p . get_open_files ( ) ] <EOL> if TESTFN in filenames : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> else : <EOL> self . assertIn ( TESTFN , filenames ) <EOL> for file in filenames : <EOL> assert os . path . isfile ( file ) , file <EOL> def test_get_open_files2 ( self ) : <EOL> fileobj = open ( TESTFN , '<STR_LIT:w>' ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> for path , fd in p . get_open_files ( ) : <EOL> if path == fileobj . name or fd == fileobj . fileno ( ) : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" % repr ( p . get_open_files ( ) ) ) <EOL> self . assertEqual ( path , fileobj . name ) <EOL> if WINDOWS : <EOL> self . assertEqual ( fd , - <NUM_LIT:1> ) <EOL> else : <EOL> self . assertEqual ( fd , fileobj . fileno ( ) ) <EOL> ntuple = p . get_open_files ( ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( ntuple [ <NUM_LIT:0> ] , ntuple . path ) <EOL> self . assertEqual ( ntuple [ <NUM_LIT:1> ] , ntuple . fd ) <EOL> fileobj . close ( ) <EOL> self . assertTrue ( fileobj . name not in p . get_open_files ( ) ) <EOL> def test_get_connections ( self ) : <EOL> arg = "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" <EOL> sproc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , arg ] ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> cons = p . get_connections ( ) <EOL> if cons : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( len ( cons ) , <NUM_LIT:1> ) <EOL> con = cons [ <NUM_LIT:0> ] <EOL> self . assertEqual ( con . family , socket . AF_INET ) <EOL> self . assertEqual ( con . type , socket . SOCK_STREAM ) <EOL> self . assertEqual ( con . status , "<STR_LIT>" ) <EOL> ip , port = con . local_address <EOL> self . assertEqual ( ip , '<STR_LIT:127.0.0.1>' ) <EOL> self . assertEqual ( con . remote_address , ( ) ) <EOL> if WINDOWS : <EOL> self . assertEqual ( con . fd , - <NUM_LIT:1> ) <EOL> else : <EOL> assert con . fd > <NUM_LIT:0> , con <EOL> self . assertEqual ( con [ <NUM_LIT:0> ] , con . fd ) <EOL> self . assertEqual ( con [ <NUM_LIT:1> ] , con . family ) <EOL> self . assertEqual ( con [ <NUM_LIT:2> ] , con . type ) <EOL> self . assertEqual ( con [ <NUM_LIT:3> ] , con . local_address ) <EOL> self . assertEqual ( con [ <NUM_LIT:4> ] , con . remote_address ) <EOL> self . assertEqual ( con [ <NUM_LIT:5> ] , con . status ) <EOL> self . assertRaises ( ValueError , p . get_connections , '<STR_LIT:foo>' ) <EOL> @ skipUnless ( supports_ipv6 ( ) ) <EOL> def test_get_connections_ipv6 ( self ) : <EOL> s = socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> s . bind ( ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> s . listen ( <NUM_LIT:1> ) <EOL> cons = psutil . Process ( os . getpid ( ) ) . get_connections ( ) <EOL> s . close ( ) <EOL> self . assertEqual ( len ( cons ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( cons [ <NUM_LIT:0> ] . local_address [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> @ skipUnless ( hasattr ( socket , '<STR_LIT>' ) ) <EOL> def test_get_connections_unix ( self ) : <EOL> safe_remove ( TESTFN ) <EOL> sock = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> sock . bind ( TESTFN ) <EOL> conn = psutil . Process ( os . getpid ( ) ) . get_connections ( kind = '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( conn . fd , sock . fileno ( ) ) <EOL> self . assertEqual ( conn . family , socket . AF_UNIX ) <EOL> self . assertEqual ( conn . type , socket . SOCK_STREAM ) <EOL> self . assertEqual ( conn . local_address , TESTFN ) <EOL> self . assertTrue ( not conn . remote_address ) <EOL> self . assertTrue ( not conn . status ) <EOL> sock . close ( ) <EOL> safe_remove ( TESTFN ) <EOL> sock = socket . socket ( socket . AF_UNIX , socket . SOCK_DGRAM ) <EOL> sock . bind ( TESTFN ) <EOL> conn = psutil . Process ( os . getpid ( ) ) . get_connections ( kind = '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( conn . type , socket . SOCK_DGRAM ) <EOL> sock . close ( ) <EOL> @ skipUnless ( hasattr ( socket , "<STR_LIT>" ) and not WINDOWS ) <EOL> def test_connection_fromfd ( self ) : <EOL> sock = socket . socket ( ) <EOL> sock . bind ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) ) <EOL> sock . listen ( <NUM_LIT:1> ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> for conn in p . get_connections ( ) : <EOL> if conn . fd == sock . fileno ( ) : <EOL> break <EOL> else : <EOL> sock . close ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> dupsock = socket . fromfd ( conn . fd , conn . family , conn . type ) <EOL> try : <EOL> self . assertEqual ( dupsock . getsockname ( ) , conn . local_address ) <EOL> self . assertNotEqual ( sock . fileno ( ) , dupsock . fileno ( ) ) <EOL> finally : <EOL> sock . close ( ) <EOL> dupsock . close ( ) <EOL> def test_get_connections_all ( self ) : <EOL> tcp_template = "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" <EOL> udp_template = "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" "<STR_LIT>" <EOL> from string import Template <EOL> tcp4_template = Template ( tcp_template ) . substitute ( family = socket . AF_INET , <EOL> addr = "<STR_LIT:127.0.0.1>" ) <EOL> udp4_template = Template ( udp_template ) . substitute ( family = socket . AF_INET , <EOL> addr = "<STR_LIT:127.0.0.1>" ) <EOL> tcp6_template = Template ( tcp_template ) . substitute ( family = socket . AF_INET6 , <EOL> addr = "<STR_LIT>" ) <EOL> udp6_template = Template ( udp_template ) . substitute ( family = socket . AF_INET6 , <EOL> addr = "<STR_LIT>" ) <EOL> tcp4_proc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , tcp4_template ] ) <EOL> udp4_proc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , udp4_template ] ) <EOL> if supports_ipv6 ( ) : <EOL> tcp6_proc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , tcp6_template ] ) <EOL> udp6_proc = get_test_subprocess ( [ PYTHON , "<STR_LIT:-c>" , udp6_template ] ) <EOL> else : <EOL> tcp6_proc = None <EOL> udp6_proc = None <EOL> all_kinds = ( "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> for p in psutil . Process ( os . getpid ( ) ) . get_children ( ) : <EOL> for conn in p . get_connections ( ) : <EOL> if p . pid == tcp4_proc . pid : <EOL> self . assertEqual ( conn . family , socket . AF_INET ) <EOL> self . assertEqual ( conn . type , socket . SOCK_STREAM ) <EOL> self . assertEqual ( conn . local_address [ <NUM_LIT:0> ] , "<STR_LIT:127.0.0.1>" ) <EOL> self . assertEqual ( conn . remote_address , ( ) ) <EOL> self . assertEqual ( conn . status , "<STR_LIT>" ) <EOL> for kind in all_kinds : <EOL> cons = p . get_connections ( kind = kind ) <EOL> if kind in ( "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> assert cons != [ ] , cons <EOL> else : <EOL> self . assertEqual ( cons , [ ] , cons ) <EOL> elif p . pid == udp4_proc . pid : <EOL> self . assertEqual ( conn . family , socket . AF_INET ) <EOL> self . assertEqual ( conn . type , socket . SOCK_DGRAM ) <EOL> self . assertEqual ( conn . local_address [ <NUM_LIT:0> ] , "<STR_LIT:127.0.0.1>" ) <EOL> self . assertEqual ( conn . remote_address , ( ) ) <EOL> self . assertEqual ( conn . status , "<STR_LIT>" ) <EOL> for kind in all_kinds : <EOL> cons = p . get_connections ( kind = kind ) <EOL> if kind in ( "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> assert cons != [ ] , cons <EOL> else : <EOL> self . assertEqual ( cons , [ ] , cons ) <EOL> elif p . pid == getattr ( tcp6_proc , "<STR_LIT>" , None ) : <EOL> self . assertEqual ( conn . family , socket . AF_INET6 ) <EOL> self . assertEqual ( conn . type , socket . SOCK_STREAM ) <EOL> self . assertIn ( conn . local_address [ <NUM_LIT:0> ] , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( conn . remote_address , ( ) ) <EOL> self . assertEqual ( conn . status , "<STR_LIT>" ) <EOL> for kind in all_kinds : <EOL> cons = p . get_connections ( kind = kind ) <EOL> if kind in ( "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> assert cons != [ ] , cons <EOL> else : <EOL> self . assertEqual ( cons , [ ] , cons ) <EOL> elif p . pid == getattr ( udp6_proc , "<STR_LIT>" , None ) : <EOL> self . assertEqual ( conn . family , socket . AF_INET6 ) <EOL> self . assertEqual ( conn . type , socket . SOCK_DGRAM ) <EOL> self . assertIn ( conn . local_address [ <NUM_LIT:0> ] , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertEqual ( conn . remote_address , ( ) ) <EOL> self . assertEqual ( conn . status , "<STR_LIT>" ) <EOL> for kind in all_kinds : <EOL> cons = p . get_connections ( kind = kind ) <EOL> if kind in ( "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> assert cons != [ ] , cons <EOL> else : <EOL> self . assertEqual ( cons , [ ] , cons ) <EOL> @ skipUnless ( POSIX ) <EOL> def test_get_num_fds ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> start = p . get_num_fds ( ) <EOL> file = open ( TESTFN , '<STR_LIT:w>' ) <EOL> self . assertEqual ( p . get_num_fds ( ) , start + <NUM_LIT:1> ) <EOL> sock = socket . socket ( ) <EOL> self . assertEqual ( p . get_num_fds ( ) , start + <NUM_LIT:2> ) <EOL> file . close ( ) <EOL> sock . close ( ) <EOL> self . assertEqual ( p . get_num_fds ( ) , start ) <EOL> def test_get_num_ctx_switches ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> before = sum ( p . get_num_ctx_switches ( ) ) <EOL> for x in range ( <NUM_LIT> ) : <EOL> after = sum ( p . get_num_ctx_switches ( ) ) <EOL> if after > before : <EOL> return <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_parent_ppid ( self ) : <EOL> this_parent = os . getpid ( ) <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertEqual ( p . ppid , this_parent ) <EOL> self . assertEqual ( p . parent . pid , this_parent ) <EOL> for p in psutil . process_iter ( ) : <EOL> if p . pid == sproc . pid : <EOL> continue <EOL> self . assertTrue ( p . ppid != this_parent ) <EOL> def test_get_children ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertEqual ( p . get_children ( ) , [ ] ) <EOL> self . assertEqual ( p . get_children ( recursive = True ) , [ ] ) <EOL> sproc = get_test_subprocess ( ) <EOL> children1 = p . get_children ( ) <EOL> children2 = p . get_children ( recursive = True ) <EOL> for children in ( children1 , children2 ) : <EOL> self . assertEqual ( len ( children ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( children [ <NUM_LIT:0> ] . pid , sproc . pid ) <EOL> self . assertEqual ( children [ <NUM_LIT:0> ] . ppid , os . getpid ( ) ) <EOL> def test_get_children_recursive ( self ) : <EOL> s = "<STR_LIT>" <EOL> s += "<STR_LIT>" <EOL> s += "<STR_LIT>" <EOL> s += "<STR_LIT>" <EOL> s += "<STR_LIT>" <EOL> get_test_subprocess ( cmd = [ PYTHON , "<STR_LIT:-c>" , s ] ) <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> self . assertEqual ( len ( p . get_children ( recursive = False ) ) , <NUM_LIT:1> ) <EOL> stop_at = time . time ( ) + <NUM_LIT> <EOL> while time . time ( ) < stop_at : <EOL> children = p . get_children ( recursive = True ) <EOL> if len ( children ) > <NUM_LIT:1> : <EOL> break <EOL> self . assertEqual ( len ( children ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( children [ <NUM_LIT:0> ] . ppid , os . getpid ( ) ) <EOL> self . assertEqual ( children [ <NUM_LIT:1> ] . ppid , children [ <NUM_LIT:0> ] . pid ) <EOL> def test_get_children_duplicates ( self ) : <EOL> from psutil . _compat import defaultdict <EOL> table = defaultdict ( int ) <EOL> for p in psutil . process_iter ( ) : <EOL> try : <EOL> table [ p . ppid ] += <NUM_LIT:1> <EOL> except psutil . Error : <EOL> pass <EOL> pid = max ( sorted ( table , key = lambda x : table [ x ] ) ) <EOL> p = psutil . Process ( pid ) <EOL> try : <EOL> c = p . get_children ( recursive = True ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> else : <EOL> self . assertEqual ( len ( c ) , len ( set ( c ) ) ) <EOL> def test_suspend_resume ( self ) : <EOL> sproc = get_test_subprocess ( wait = True ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . suspend ( ) <EOL> for x in range ( <NUM_LIT:100> ) : <EOL> if p . status == psutil . STATUS_STOPPED : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( str ( p . status ) , "<STR_LIT>" ) <EOL> p . resume ( ) <EOL> assert p . status != psutil . STATUS_STOPPED , p . status <EOL> def test_invalid_pid ( self ) : <EOL> self . assertRaises ( TypeError , psutil . Process , "<STR_LIT:1>" ) <EOL> self . assertRaises ( TypeError , psutil . Process , None ) <EOL> self . assertRaises ( psutil . NoSuchProcess , psutil . Process , - <NUM_LIT:1> ) <EOL> def test_as_dict ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> d = p . as_dict ( ) <EOL> try : <EOL> import json <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> json . dumps ( d ) <EOL> d = p . as_dict ( attrs = [ '<STR_LIT>' , '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( sorted ( d . keys ( ) ) , [ '<STR_LIT>' , '<STR_LIT:name>' ] ) <EOL> p = psutil . Process ( min ( psutil . get_pid_list ( ) ) ) <EOL> d = p . as_dict ( attrs = [ '<STR_LIT>' ] , ad_value = '<STR_LIT:foo>' ) <EOL> if not isinstance ( d [ '<STR_LIT>' ] , list ) : <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT:foo>' ) <EOL> def test_zombie_process ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> for name in dir ( p ) : <EOL> if name . startswith ( '<STR_LIT:_>' ) or name in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> continue <EOL> try : <EOL> meth = getattr ( p , name ) <EOL> if callable ( meth ) : <EOL> meth ( ) <EOL> except psutil . NoSuchProcess : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" % name ) <EOL> try : <EOL> if os . name == '<STR_LIT>' : <EOL> p . set_nice ( <NUM_LIT:1> ) <EOL> else : <EOL> p . set_nice ( psutil . NORMAL_PRIORITY_CLASS ) <EOL> except psutil . NoSuchProcess : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> if hasattr ( p , '<STR_LIT>' ) : <EOL> self . assertRaises ( psutil . NoSuchProcess , p . set_ionice , <NUM_LIT:2> ) <EOL> self . assertRaises ( psutil . NoSuchProcess , p . send_signal , signal . SIGTERM ) <EOL> self . assertRaises ( psutil . NoSuchProcess , p . set_nice , <NUM_LIT:0> ) <EOL> self . assertFalse ( p . is_running ( ) ) <EOL> if hasattr ( p , "<STR_LIT>" ) : <EOL> self . assertRaises ( psutil . NoSuchProcess , p . set_cpu_affinity , [ <NUM_LIT:0> ] ) <EOL> def test__str__ ( self ) : <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> self . assertIn ( str ( sproc . pid ) , str ( p ) ) <EOL> if not OSX : <EOL> self . assertIn ( os . path . basename ( PYTHON ) , str ( p ) ) <EOL> sproc = get_test_subprocess ( ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> p . kill ( ) <EOL> p . wait ( ) <EOL> self . assertIn ( str ( sproc . pid ) , str ( p ) ) <EOL> self . assertIn ( "<STR_LIT>" , str ( p ) ) <EOL> @ skipIf ( LINUX ) <EOL> def test_pid_0 ( self ) : <EOL> p = psutil . Process ( <NUM_LIT:0> ) <EOL> self . assertTrue ( p . name ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertEqual ( p . uids . real , <NUM_LIT:0> ) <EOL> self . assertEqual ( p . gids . real , <NUM_LIT:0> ) <EOL> self . assertIn ( p . ppid , ( <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> self . assertEqual ( p . cmdline , [ ] ) <EOL> try : <EOL> p . get_num_threads ( ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> try : <EOL> p . get_memory_info ( ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> if POSIX : <EOL> self . assertEqual ( p . username , '<STR_LIT:root>' ) <EOL> elif WINDOWS : <EOL> self . assertEqual ( p . username , '<STR_LIT>' ) <EOL> else : <EOL> p . username <EOL> self . assertIn ( <NUM_LIT:0> , psutil . get_pid_list ( ) ) <EOL> self . assertTrue ( psutil . pid_exists ( <NUM_LIT:0> ) ) <EOL> def test_Popen ( self ) : <EOL> cmd = [ PYTHON , "<STR_LIT:-c>" , "<STR_LIT>" ] <EOL> proc = psutil . Popen ( cmd , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> try : <EOL> proc . name <EOL> proc . stdin <EOL> self . assertTrue ( hasattr ( proc , '<STR_LIT:name>' ) ) <EOL> self . assertTrue ( hasattr ( proc , '<STR_LIT>' ) ) <EOL> self . assertRaises ( AttributeError , getattr , proc , '<STR_LIT:foo>' ) <EOL> finally : <EOL> proc . kill ( ) <EOL> proc . wait ( ) <EOL> class TestFetchAllProcesses ( unittest . TestCase ) : <EOL> if not hasattr ( unittest . TestCase , "<STR_LIT>" ) : <EOL> def assertIn ( self , member , container , msg = None ) : <EOL> if member not in container : <EOL> self . fail ( msg or '<STR_LIT>' % ( repr ( member ) , repr ( container ) ) ) <EOL> def setUp ( self ) : <EOL> if POSIX : <EOL> import pwd <EOL> pall = pwd . getpwall ( ) <EOL> self . _uids = set ( [ x . pw_uid for x in pall ] ) <EOL> self . _usernames = set ( [ x . pw_name for x in pall ] ) <EOL> def test_fetch_all ( self ) : <EOL> valid_procs = <NUM_LIT:0> <EOL> excluded_names = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> attrs = [ ] <EOL> for name in dir ( psutil . Process ) : <EOL> if name . startswith ( "<STR_LIT:_>" ) : <EOL> continue <EOL> if name . startswith ( "<STR_LIT>" ) : <EOL> continue <EOL> if name in excluded_names : <EOL> continue <EOL> attrs . append ( name ) <EOL> for p in psutil . process_iter ( ) : <EOL> for name in attrs : <EOL> try : <EOL> try : <EOL> attr = getattr ( p , name , None ) <EOL> if attr is not None and callable ( attr ) : <EOL> ret = attr ( ) <EOL> else : <EOL> ret = attr <EOL> valid_procs += <NUM_LIT:1> <EOL> except ( psutil . NoSuchProcess , psutil . AccessDenied ) : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . assertEqual ( err . pid , p . pid ) <EOL> if err . name : <EOL> self . assertEqual ( err . name , p . name ) <EOL> self . assertTrue ( str ( err ) ) <EOL> self . assertTrue ( err . msg ) <EOL> else : <EOL> if ret not in ( <NUM_LIT:0> , <NUM_LIT:0.0> , [ ] , None , '<STR_LIT>' ) : <EOL> assert ret , ret <EOL> meth = getattr ( self , name ) <EOL> meth ( ret ) <EOL> except Exception : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> trace = traceback . format_exc ( ) <EOL> self . fail ( '<STR_LIT>' <EOL> % ( trace , p , name , ret ) ) <EOL> self . assertTrue ( valid_procs > <NUM_LIT:0> ) <EOL> def cmdline ( self , ret ) : <EOL> pass <EOL> def exe ( self , ret ) : <EOL> if not ret : <EOL> assert ret == '<STR_LIT>' <EOL> else : <EOL> assert os . path . isabs ( ret ) , ret <EOL> if POSIX : <EOL> assert os . path . isfile ( ret ) , ret <EOL> if hasattr ( os , '<STR_LIT>' ) and hasattr ( os , "<STR_LIT>" ) : <EOL> self . assertTrue ( os . access ( ret , os . X_OK ) ) <EOL> def ppid ( self , ret ) : <EOL> self . assertTrue ( ret >= <NUM_LIT:0> ) <EOL> def name ( self , ret ) : <EOL> self . assertTrue ( isinstance ( ret , str ) ) <EOL> self . assertTrue ( ret ) <EOL> def create_time ( self , ret ) : <EOL> self . assertTrue ( ret > <NUM_LIT:0> ) <EOL> if not WINDOWS : <EOL> self . assertTrue ( ret >= psutil . BOOT_TIME ) <EOL> time . strftime ( "<STR_LIT>" , time . localtime ( ret ) ) <EOL> def uids ( self , ret ) : <EOL> for uid in ret : <EOL> self . assertTrue ( uid >= <NUM_LIT:0> ) <EOL> self . assertIn ( uid , self . _uids ) <EOL> def gids ( self , ret ) : <EOL> for gid in ret : <EOL> self . assertTrue ( gid >= <NUM_LIT:0> ) <EOL> def username ( self , ret ) : <EOL> self . assertTrue ( ret ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . assertIn ( ret , self . _usernames ) <EOL> def status ( self , ret ) : <EOL> self . assertTrue ( ret >= <NUM_LIT:0> ) <EOL> self . assertTrue ( str ( ret ) != '<STR_LIT:?>' ) <EOL> def get_io_counters ( self , ret ) : <EOL> for field in ret : <EOL> if field != - <NUM_LIT:1> : <EOL> self . assertTrue ( field >= <NUM_LIT:0> ) <EOL> def get_ionice ( self , ret ) : <EOL> self . assertTrue ( ret . ioclass >= <NUM_LIT:0> ) <EOL> self . assertTrue ( ret . value >= <NUM_LIT:0> ) <EOL> def get_num_threads ( self , ret ) : <EOL> self . assertTrue ( ret >= <NUM_LIT:1> ) <EOL> def get_threads ( self , ret ) : <EOL> for t in ret : <EOL> self . assertTrue ( t . id >= <NUM_LIT:0> ) <EOL> self . assertTrue ( t . user_time >= <NUM_LIT:0> ) <EOL> self . assertTrue ( t . system_time >= <NUM_LIT:0> ) <EOL> def get_cpu_times ( self , ret ) : <EOL> self . assertTrue ( ret . user >= <NUM_LIT:0> ) <EOL> self . assertTrue ( ret . system >= <NUM_LIT:0> ) <EOL> def get_memory_info ( self , ret ) : <EOL> self . assertTrue ( ret . rss >= <NUM_LIT:0> ) <EOL> self . assertTrue ( ret . vms >= <NUM_LIT:0> ) <EOL> def get_ext_memory_info ( self , ret ) : <EOL> for name in ret . _fields : <EOL> self . assertTrue ( getattr ( ret , name ) >= <NUM_LIT:0> ) <EOL> if POSIX and ret . vms != <NUM_LIT:0> : <EOL> for name in ret . _fields : <EOL> if name != '<STR_LIT>' : <EOL> value = getattr ( ret , name ) <EOL> assert ret . vms > value , ret <EOL> elif WINDOWS : <EOL> assert ret . peak_wset >= ret . wset , ret <EOL> assert ret . peak_paged_pool >= ret . paged_pool , ret <EOL> assert ret . peak_nonpaged_pool >= ret . nonpaged_pool , ret <EOL> assert ret . peak_pagefile >= ret . pagefile , ret <EOL> def get_open_files ( self , ret ) : <EOL> for f in ret : <EOL> if WINDOWS : <EOL> assert f . fd == - <NUM_LIT:1> , f <EOL> else : <EOL> assert isinstance ( f . fd , int ) , f <EOL> assert os . path . isabs ( f . path ) , f <EOL> assert os . path . isfile ( f . path ) , f <EOL> def get_num_fds ( self , ret ) : <EOL> self . assertTrue ( ret >= <NUM_LIT:0> ) <EOL> def get_connections ( self , ret ) : <EOL> valid_conn_states = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> for conn in ret : <EOL> self . assertIn ( conn . type , ( socket . SOCK_STREAM , socket . SOCK_DGRAM ) ) <EOL> self . assertIn ( conn . family , ( socket . AF_INET , socket . AF_INET6 ) ) <EOL> check_ip_address ( conn . local_address , conn . family ) <EOL> check_ip_address ( conn . remote_address , conn . family ) <EOL> if conn . status not in valid_conn_states : <EOL> self . fail ( "<STR_LIT>" % conn . status ) <EOL> if conn . family == socket . AF_INET : <EOL> s = socket . socket ( conn . family , conn . type ) <EOL> s . bind ( ( conn . local_address [ <NUM_LIT:0> ] , <NUM_LIT:0> ) ) <EOL> s . close ( ) <EOL> if not WINDOWS and hasattr ( socket , '<STR_LIT>' ) : <EOL> dupsock = None <EOL> try : <EOL> try : <EOL> dupsock = socket . fromfd ( conn . fd , conn . family , conn . type ) <EOL> except ( socket . error , OSError ) : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . args [ <NUM_LIT:0> ] == errno . EBADF : <EOL> continue <EOL> raise <EOL> if hasattr ( dupsock , "<STR_LIT>" ) : <EOL> self . assertEqual ( dupsock . family , conn . family ) <EOL> self . assertEqual ( dupsock . type , conn . type ) <EOL> finally : <EOL> if dupsock is not None : <EOL> dupsock . close ( ) <EOL> def getcwd ( self , ret ) : <EOL> if ret is not None : <EOL> assert os . path . isabs ( ret ) , ret <EOL> try : <EOL> st = os . stat ( ret ) <EOL> except OSError : <EOL> err = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if err . errno != errno . ENOENT : <EOL> raise <EOL> else : <EOL> self . assertTrue ( stat . S_ISDIR ( st . st_mode ) ) <EOL> def get_memory_percent ( self , ret ) : <EOL> assert <NUM_LIT:0> <= ret <= <NUM_LIT:100> , ret <EOL> def is_running ( self , ret ) : <EOL> self . assertTrue ( ret ) <EOL> def get_cpu_affinity ( self , ret ) : <EOL> assert ret != [ ] , ret <EOL> def terminal ( self , ret ) : <EOL> if ret is not None : <EOL> assert os . path . isabs ( ret ) , ret <EOL> assert os . path . exists ( ret ) , ret <EOL> def get_memory_maps ( self , ret ) : <EOL> for nt in ret : <EOL> for fname in nt . _fields : <EOL> value = getattr ( nt , fname ) <EOL> if fname == '<STR_LIT:path>' : <EOL> if not value . startswith ( '<STR_LIT:[>' ) : <EOL> assert os . path . isabs ( nt . path ) , nt . path <EOL> elif fname in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertTrue ( value ) <EOL> else : <EOL> assert isinstance ( value , ( int , long ) ) <EOL> assert value >= <NUM_LIT:0> , value <EOL> def get_num_handles ( self , ret ) : <EOL> if WINDOWS : <EOL> assert ret >= <NUM_LIT:0> <EOL> else : <EOL> assert ret > <NUM_LIT:0> <EOL> def get_nice ( self , ret ) : <EOL> if POSIX : <EOL> assert - <NUM_LIT:20> <= ret <= <NUM_LIT:20> , ret <EOL> else : <EOL> priorities = [ getattr ( psutil , x ) for x in dir ( psutil ) <EOL> if x . endswith ( '<STR_LIT>' ) ] <EOL> self . assertIn ( ret , priorities ) <EOL> def get_num_ctx_switches ( self , ret ) : <EOL> self . assertTrue ( ret . voluntary >= <NUM_LIT:0> ) <EOL> self . assertTrue ( ret . involuntary >= <NUM_LIT:0> ) <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> class LimitedUserTestCase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> PROCESS_UID = os . getuid ( ) <EOL> PROCESS_GID = os . getgid ( ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> TestCase . __init__ ( self , * args , ** kwargs ) <EOL> for attr in [ x for x in dir ( self ) if x . startswith ( '<STR_LIT:test>' ) ] : <EOL> meth = getattr ( self , attr ) <EOL> def test_ ( self ) : <EOL> try : <EOL> meth ( ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> setattr ( self , attr , types . MethodType ( test_ , self ) ) <EOL> def setUp ( self ) : <EOL> os . setegid ( <NUM_LIT:1000> ) <EOL> os . seteuid ( <NUM_LIT:1000> ) <EOL> TestCase . setUp ( self ) <EOL> def tearDown ( self ) : <EOL> os . setegid ( self . PROCESS_UID ) <EOL> os . seteuid ( self . PROCESS_GID ) <EOL> TestCase . tearDown ( self ) <EOL> def test_nice ( self ) : <EOL> try : <EOL> psutil . Process ( os . getpid ( ) ) . set_nice ( - <NUM_LIT:1> ) <EOL> except psutil . AccessDenied : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def cleanup ( ) : <EOL> reap_children ( search_all = True ) <EOL> DEVNULL . close ( ) <EOL> safe_remove ( TESTFN ) <EOL> atexit . register ( cleanup ) <EOL> def test_main ( ) : <EOL> tests = [ ] <EOL> test_suite = unittest . TestSuite ( ) <EOL> tests . append ( TestCase ) <EOL> tests . append ( TestFetchAllProcesses ) <EOL> if POSIX : <EOL> from _posix import PosixSpecificTestCase <EOL> tests . append ( PosixSpecificTestCase ) <EOL> if LINUX : <EOL> from _linux import LinuxSpecificTestCase as stc <EOL> elif WINDOWS : <EOL> from _windows import WindowsSpecificTestCase as stc <EOL> from _windows import TestDualProcessImplementation <EOL> tests . append ( TestDualProcessImplementation ) <EOL> elif OSX : <EOL> from _osx import OSXSpecificTestCase as stc <EOL> elif BSD : <EOL> from _bsd import BSDSpecificTestCase as stc <EOL> tests . append ( stc ) <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> if os . getuid ( ) == <NUM_LIT:0> : <EOL> tests . append ( LimitedUserTestCase ) <EOL> else : <EOL> atexit . register ( warn , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> for test_class in tests : <EOL> test_suite . addTest ( unittest . makeSuite ( test_class ) ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( test_suite ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_main ( ) </s>
