<s> """<STR_LIT>""" <EOL> import contextlib <EOL> import errno <EOL> import io <EOL> import os <EOL> import pprint <EOL> import re <EOL> import shutil <EOL> import socket <EOL> import struct <EOL> import tempfile <EOL> import textwrap <EOL> import time <EOL> import warnings <EOL> try : <EOL> from unittest import mock <EOL> except ImportError : <EOL> import mock <EOL> import psutil <EOL> from psutil import LINUX <EOL> from psutil . _compat import PY3 <EOL> from psutil . _compat import u <EOL> from psutil . tests import call_until <EOL> from psutil . tests import get_kernel_version <EOL> from psutil . tests import importlib <EOL> from psutil . tests import MEMORY_TOLERANCE <EOL> from psutil . tests import PYPY <EOL> from psutil . tests import pyrun <EOL> from psutil . tests import reap_children <EOL> from psutil . tests import retry_before_failing <EOL> from psutil . tests import run_test_module_by_name <EOL> from psutil . tests import safe_remove <EOL> from psutil . tests import sh <EOL> from psutil . tests import skip_on_not_implemented <EOL> from psutil . tests import TESTFN <EOL> from psutil . tests import TRAVIS <EOL> from psutil . tests import unittest <EOL> from psutil . tests import which <EOL> HERE = os . path . abspath ( os . path . dirname ( __file__ ) ) <EOL> SIOCGIFADDR = <NUM_LIT> <EOL> SIOCGIFCONF = <NUM_LIT> <EOL> SIOCGIFHWADDR = <NUM_LIT> <EOL> if LINUX : <EOL> SECTOR_SIZE = psutil . _psplatform . SECTOR_SIZE <EOL> def get_ipv4_address ( ifname ) : <EOL> import fcntl <EOL> ifname = ifname [ : <NUM_LIT:15> ] <EOL> if PY3 : <EOL> ifname = bytes ( ifname , '<STR_LIT:ascii>' ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> with contextlib . closing ( s ) : <EOL> return socket . inet_ntoa ( <EOL> fcntl . ioctl ( s . fileno ( ) , <EOL> SIOCGIFADDR , <EOL> struct . pack ( '<STR_LIT>' , ifname ) ) [ <NUM_LIT:20> : <NUM_LIT> ] ) <EOL> def get_mac_address ( ifname ) : <EOL> import fcntl <EOL> ifname = ifname [ : <NUM_LIT:15> ] <EOL> if PY3 : <EOL> ifname = bytes ( ifname , '<STR_LIT:ascii>' ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> with contextlib . closing ( s ) : <EOL> info = fcntl . ioctl ( <EOL> s . fileno ( ) , SIOCGIFHWADDR , struct . pack ( '<STR_LIT>' , ifname ) ) <EOL> if PY3 : <EOL> def ord ( x ) : <EOL> return x <EOL> else : <EOL> import __builtin__ <EOL> ord = __builtin__ . ord <EOL> return '<STR_LIT>' . join ( [ '<STR_LIT>' % ord ( char ) for char in info [ <NUM_LIT> : <NUM_LIT> ] ] ) [ : - <NUM_LIT:1> ] <EOL> def free_swap ( ) : <EOL> """<STR_LIT>""" <EOL> lines = sh ( '<STR_LIT>' ) . split ( '<STR_LIT:\n>' ) <EOL> for line in lines : <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> _ , total , used , free = line . split ( ) <EOL> return ( int ( total ) * <NUM_LIT> , int ( used ) * <NUM_LIT> , int ( free ) * <NUM_LIT> ) <EOL> raise ValueError ( <EOL> "<STR_LIT>" % '<STR_LIT:\n>' . join ( lines ) ) <EOL> def free_physmem ( ) : <EOL> """<STR_LIT>""" <EOL> lines = sh ( '<STR_LIT>' ) . split ( '<STR_LIT:\n>' ) <EOL> for line in lines : <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> total , used , free , shared = [ int ( x ) * <NUM_LIT> for x in line . split ( ) [ <NUM_LIT:1> : <NUM_LIT:5> ] ] <EOL> return ( total , used , free , shared ) <EOL> raise ValueError ( <EOL> "<STR_LIT>" % '<STR_LIT:\n>' . join ( lines ) ) <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestSystemVirtualMemory ( unittest . TestCase ) : <EOL> def test_total ( self ) : <EOL> total , used , free , shared = free_physmem ( ) <EOL> self . assertEqual ( total , psutil . virtual_memory ( ) . total ) <EOL> @ retry_before_failing ( ) <EOL> def test_used ( self ) : <EOL> total , used , free , shared = free_physmem ( ) <EOL> self . assertAlmostEqual ( used , psutil . virtual_memory ( ) . used , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> def test_free ( self ) : <EOL> total , used , free , shared = free_physmem ( ) <EOL> self . assertAlmostEqual ( free , psutil . virtual_memory ( ) . free , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> def test_buffers ( self ) : <EOL> buffers = int ( sh ( '<STR_LIT>' ) . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:2> ] . split ( ) [ <NUM_LIT:4> ] ) * <NUM_LIT> <EOL> self . assertAlmostEqual ( buffers , psutil . virtual_memory ( ) . buffers , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> def test_cached ( self ) : <EOL> cached = int ( sh ( '<STR_LIT>' ) . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:2> ] . split ( ) [ <NUM_LIT:5> ] ) * <NUM_LIT> <EOL> self . assertAlmostEqual ( cached , psutil . virtual_memory ( ) . cached , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> @ unittest . skipIf ( TRAVIS , "<STR_LIT>" ) <EOL> def test_shared ( self ) : <EOL> total , used , free , shared = free_physmem ( ) <EOL> self . assertAlmostEqual ( shared , psutil . virtual_memory ( ) . shared , <EOL> delta = MEMORY_TOLERANCE ) <EOL> def test_warnings_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> with warnings . catch_warnings ( record = True ) as ws : <EOL> warnings . simplefilter ( "<STR_LIT>" ) <EOL> ret = psutil . _pslinux . virtual_memory ( ) <EOL> assert m . called <EOL> self . assertEqual ( len ( ws ) , <NUM_LIT:1> ) <EOL> w = ws [ <NUM_LIT:0> ] <EOL> self . assertTrue ( w . filename . endswith ( '<STR_LIT>' ) ) <EOL> self . assertIn ( <EOL> "<STR_LIT>" , str ( w . message ) ) <EOL> self . assertEqual ( ret . cached , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . active , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . inactive , <NUM_LIT:0> ) <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestSystemSwapMemory ( unittest . TestCase ) : <EOL> def test_total ( self ) : <EOL> total , used , free = free_swap ( ) <EOL> return self . assertAlmostEqual ( total , psutil . swap_memory ( ) . total , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> def test_used ( self ) : <EOL> total , used , free = free_swap ( ) <EOL> return self . assertAlmostEqual ( used , psutil . swap_memory ( ) . used , <EOL> delta = MEMORY_TOLERANCE ) <EOL> @ retry_before_failing ( ) <EOL> def test_free ( self ) : <EOL> total , used , free = free_swap ( ) <EOL> return self . assertAlmostEqual ( free , psutil . swap_memory ( ) . free , <EOL> delta = MEMORY_TOLERANCE ) <EOL> def test_warnings_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> with warnings . catch_warnings ( record = True ) as ws : <EOL> warnings . simplefilter ( "<STR_LIT>" ) <EOL> ret = psutil . _pslinux . swap_memory ( ) <EOL> assert m . called <EOL> self . assertEqual ( len ( ws ) , <NUM_LIT:1> ) <EOL> w = ws [ <NUM_LIT:0> ] <EOL> self . assertTrue ( w . filename . endswith ( '<STR_LIT>' ) ) <EOL> self . assertIn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , str ( w . message ) ) <EOL> self . assertEqual ( ret . sin , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . sout , <NUM_LIT:0> ) <EOL> def test_no_vmstat_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True , <EOL> side_effect = IOError ) as m : <EOL> with warnings . catch_warnings ( record = True ) as ws : <EOL> warnings . simplefilter ( "<STR_LIT>" ) <EOL> ret = psutil . swap_memory ( ) <EOL> assert m . called <EOL> self . assertEqual ( len ( ws ) , <NUM_LIT:1> ) <EOL> w = ws [ <NUM_LIT:0> ] <EOL> self . assertTrue ( w . filename . endswith ( '<STR_LIT>' ) ) <EOL> self . assertIn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> str ( w . message ) ) <EOL> self . assertEqual ( ret . sin , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . sout , <NUM_LIT:0> ) <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestSystemCPU ( unittest . TestCase ) : <EOL> @ unittest . skipIf ( TRAVIS , "<STR_LIT>" ) <EOL> def test_cpu_times ( self ) : <EOL> fields = psutil . cpu_times ( ) . _fields <EOL> kernel_ver = re . findall ( '<STR_LIT>' , os . uname ( ) [ <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] <EOL> kernel_ver_info = tuple ( map ( int , kernel_ver . split ( '<STR_LIT:.>' ) ) ) <EOL> if kernel_ver_info >= ( <NUM_LIT:2> , <NUM_LIT:6> , <NUM_LIT:11> ) : <EOL> self . assertIn ( '<STR_LIT>' , fields ) <EOL> else : <EOL> self . assertNotIn ( '<STR_LIT>' , fields ) <EOL> if kernel_ver_info >= ( <NUM_LIT:2> , <NUM_LIT:6> , <NUM_LIT> ) : <EOL> self . assertIn ( '<STR_LIT>' , fields ) <EOL> else : <EOL> self . assertNotIn ( '<STR_LIT>' , fields ) <EOL> if kernel_ver_info >= ( <NUM_LIT:3> , <NUM_LIT:2> , <NUM_LIT:0> ) : <EOL> self . assertIn ( '<STR_LIT>' , fields ) <EOL> else : <EOL> self . assertNotIn ( '<STR_LIT>' , fields ) <EOL> @ unittest . skipUnless ( which ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> def test_cpu_count_logical_w_nproc ( self ) : <EOL> num = int ( sh ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( psutil . cpu_count ( logical = True ) , num ) <EOL> @ unittest . skipUnless ( which ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> def test_cpu_count_logical_w_lscpu ( self ) : <EOL> out = sh ( "<STR_LIT>" ) <EOL> num = len ( [ x for x in out . split ( '<STR_LIT:\n>' ) if not x . startswith ( '<STR_LIT:#>' ) ] ) <EOL> self . assertEqual ( psutil . cpu_count ( logical = True ) , num ) <EOL> def test_cpu_count_logical_mocked ( self ) : <EOL> import psutil . _pslinux <EOL> original = psutil . _pslinux . cpu_count_logical ( ) <EOL> with mock . patch ( <EOL> '<STR_LIT>' , side_effect = ValueError ) as m : <EOL> self . assertEqual ( psutil . _pslinux . cpu_count_logical ( ) , original ) <EOL> assert m . called <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertIsNone ( psutil . _pslinux . cpu_count_logical ( ) ) <EOL> self . assertEqual ( m . call_count , <NUM_LIT:2> ) <EOL> self . assertEqual ( m . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:rb>' ) as f : <EOL> cpuinfo_data = f . read ( ) <EOL> fake_file = io . BytesIO ( cpuinfo_data ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_file , create = True ) as m : <EOL> self . assertEqual ( psutil . _pslinux . cpu_count_logical ( ) , original ) <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> return io . BytesIO ( b"<STR_LIT>" ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock , create = True ) : <EOL> self . assertEqual ( psutil . _pslinux . cpu_count_logical ( ) , original ) <EOL> def test_cpu_count_physical_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertIsNone ( psutil . _pslinux . cpu_count_physical ( ) ) <EOL> assert m . called <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestSystemNetwork ( unittest . TestCase ) : <EOL> def test_net_if_addrs_ips ( self ) : <EOL> for name , addrs in psutil . net_if_addrs ( ) . items ( ) : <EOL> for addr in addrs : <EOL> if addr . family == psutil . AF_LINK : <EOL> self . assertEqual ( addr . address , get_mac_address ( name ) ) <EOL> elif addr . family == socket . AF_INET : <EOL> self . assertEqual ( addr . address , get_ipv4_address ( name ) ) <EOL> def test_net_if_stats ( self ) : <EOL> for name , stats in psutil . net_if_stats ( ) . items ( ) : <EOL> try : <EOL> out = sh ( "<STR_LIT>" % name ) <EOL> except RuntimeError : <EOL> pass <EOL> else : <EOL> self . assertEqual ( stats . isup , '<STR_LIT>' in out ) <EOL> self . assertEqual ( stats . mtu , <EOL> int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) ) <EOL> def test_net_io_counters ( self ) : <EOL> def ifconfig ( nic ) : <EOL> ret = { } <EOL> out = sh ( "<STR_LIT>" % name ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:1> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:1> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> ret [ '<STR_LIT>' ] = int ( re . findall ( '<STR_LIT>' , out ) [ <NUM_LIT:0> ] ) <EOL> return ret <EOL> for name , stats in psutil . net_io_counters ( pernic = True ) . items ( ) : <EOL> try : <EOL> ifconfig_ret = ifconfig ( name ) <EOL> except RuntimeError : <EOL> continue <EOL> self . assertAlmostEqual ( <EOL> stats . bytes_recv , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> stats . bytes_sent , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> stats . packets_recv , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> stats . packets_sent , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> stats . errin , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT:10> ) <EOL> self . assertAlmostEqual ( <EOL> stats . errout , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT:10> ) <EOL> self . assertAlmostEqual ( <EOL> stats . dropin , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT:10> ) <EOL> self . assertAlmostEqual ( <EOL> stats . dropout , ifconfig_ret [ '<STR_LIT>' ] , delta = <NUM_LIT:10> ) <EOL> @ unittest . skipUnless ( which ( '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( TRAVIS , "<STR_LIT>" ) <EOL> def test_net_if_names ( self ) : <EOL> out = sh ( "<STR_LIT>" ) . strip ( ) <EOL> nics = [ x for x in psutil . net_if_addrs ( ) . keys ( ) if '<STR_LIT::>' not in x ] <EOL> found = <NUM_LIT:0> <EOL> for line in out . split ( '<STR_LIT:\n>' ) : <EOL> line = line . strip ( ) <EOL> if re . search ( "<STR_LIT>" , line ) : <EOL> found += <NUM_LIT:1> <EOL> name = line . split ( '<STR_LIT::>' ) [ <NUM_LIT:1> ] . strip ( ) <EOL> self . assertIn ( name , nics ) <EOL> self . assertEqual ( len ( nics ) , found , msg = "<STR_LIT>" % ( <EOL> pprint . pformat ( nics ) , out ) ) <EOL> @ mock . patch ( '<STR_LIT>' , side_effect = ValueError ) <EOL> @ mock . patch ( '<STR_LIT>' , return_value = False ) <EOL> def test_net_connections_ipv6_unsupported ( self , supports_ipv6 , inet_ntop ) : <EOL> try : <EOL> s = socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> self . addCleanup ( s . close ) <EOL> s . bind ( ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> except socket . error : <EOL> pass <EOL> psutil . net_connections ( kind = '<STR_LIT>' ) <EOL> def test_net_connections_mocked ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name == '<STR_LIT>' : <EOL> return io . StringIO ( textwrap . dedent ( u"""<STR_LIT>""" ) ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> return orig_open ( name , * args ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) as m : <EOL> psutil . net_connections ( kind = '<STR_LIT>' ) <EOL> assert m . called <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestSystemDisks ( unittest . TestCase ) : <EOL> @ unittest . skipUnless ( <EOL> hasattr ( os , '<STR_LIT>' ) , "<STR_LIT>" ) <EOL> @ skip_on_not_implemented ( ) <EOL> def test_disk_partitions_and_usage ( self ) : <EOL> def df ( path ) : <EOL> out = sh ( '<STR_LIT>' % path ) . strip ( ) <EOL> lines = out . split ( '<STR_LIT:\n>' ) <EOL> lines . pop ( <NUM_LIT:0> ) <EOL> line = lines . pop ( <NUM_LIT:0> ) <EOL> dev , total , used , free = line . split ( ) [ : <NUM_LIT:4> ] <EOL> if dev == '<STR_LIT:none>' : <EOL> dev = '<STR_LIT>' <EOL> total , used , free = int ( total ) , int ( used ) , int ( free ) <EOL> return dev , total , used , free <EOL> for part in psutil . disk_partitions ( all = False ) : <EOL> usage = psutil . disk_usage ( part . mountpoint ) <EOL> dev , total , used , free = df ( part . mountpoint ) <EOL> self . assertEqual ( usage . total , total ) <EOL> if abs ( usage . free - free ) > <NUM_LIT:10> * <NUM_LIT> * <NUM_LIT> : <EOL> self . fail ( "<STR_LIT>" % ( usage . free , free ) ) <EOL> if abs ( usage . used - used ) > <NUM_LIT:10> * <NUM_LIT> * <NUM_LIT> : <EOL> self . fail ( "<STR_LIT>" % ( usage . used , used ) ) <EOL> def test_disk_partitions_mocked ( self ) : <EOL> with open ( "<STR_LIT>" , "<STR_LIT:r>" ) as f : <EOL> data = f . read ( ) <EOL> if '<STR_LIT>' in data : <EOL> for part in psutil . disk_partitions ( ) : <EOL> if part . fstype == '<STR_LIT>' : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> else : <EOL> fake_file = io . StringIO ( u ( "<STR_LIT>" ) ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_file , create = True ) as m1 : <EOL> with mock . patch ( <EOL> '<STR_LIT>' , <EOL> return_value = [ ( '<STR_LIT>' , '<STR_LIT:/>' , '<STR_LIT>' , '<STR_LIT>' ) ] ) as m2 : <EOL> ret = psutil . disk_partitions ( ) <EOL> assert m1 . called <EOL> assert m2 . called <EOL> assert ret <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] . fstype , '<STR_LIT>' ) <EOL> def test_disk_io_counters_kernel_2_4_mocked ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name == '<STR_LIT>' : <EOL> return io . StringIO ( textwrap . dedent ( u"""<STR_LIT>""" ) ) <EOL> elif name == '<STR_LIT>' : <EOL> return io . StringIO ( <EOL> u ( "<STR_LIT>" ) ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> return orig_open ( name , * args ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) as m : <EOL> ret = psutil . disk_io_counters ( ) <EOL> assert m . called <EOL> self . assertEqual ( ret . read_count , <NUM_LIT:1> ) <EOL> self . assertEqual ( ret . read_merged_count , <NUM_LIT:2> ) <EOL> self . assertEqual ( ret . read_bytes , <NUM_LIT:3> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . read_time , <NUM_LIT:4> ) <EOL> self . assertEqual ( ret . write_count , <NUM_LIT:5> ) <EOL> self . assertEqual ( ret . write_merged_count , <NUM_LIT:6> ) <EOL> self . assertEqual ( ret . write_bytes , <NUM_LIT:7> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . write_time , <NUM_LIT:8> ) <EOL> self . assertEqual ( ret . busy_time , <NUM_LIT:10> ) <EOL> def test_disk_io_counters_kernel_2_6_full_mocked ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name == '<STR_LIT>' : <EOL> return io . StringIO ( textwrap . dedent ( u"""<STR_LIT>""" ) ) <EOL> elif name == '<STR_LIT>' : <EOL> return io . StringIO ( <EOL> u ( "<STR_LIT>" ) ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> return orig_open ( name , * args ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) as m : <EOL> ret = psutil . disk_io_counters ( ) <EOL> assert m . called <EOL> self . assertEqual ( ret . read_count , <NUM_LIT:1> ) <EOL> self . assertEqual ( ret . read_merged_count , <NUM_LIT:2> ) <EOL> self . assertEqual ( ret . read_bytes , <NUM_LIT:3> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . read_time , <NUM_LIT:4> ) <EOL> self . assertEqual ( ret . write_count , <NUM_LIT:5> ) <EOL> self . assertEqual ( ret . write_merged_count , <NUM_LIT:6> ) <EOL> self . assertEqual ( ret . write_bytes , <NUM_LIT:7> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . write_time , <NUM_LIT:8> ) <EOL> self . assertEqual ( ret . busy_time , <NUM_LIT:10> ) <EOL> def test_disk_io_counters_kernel_2_6_limited_mocked ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name == '<STR_LIT>' : <EOL> return io . StringIO ( textwrap . dedent ( u"""<STR_LIT>""" ) ) <EOL> elif name == '<STR_LIT>' : <EOL> return io . StringIO ( <EOL> u ( "<STR_LIT>" ) ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> return orig_open ( name , * args ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) as m : <EOL> ret = psutil . disk_io_counters ( ) <EOL> assert m . called <EOL> self . assertEqual ( ret . read_count , <NUM_LIT:1> ) <EOL> self . assertEqual ( ret . read_bytes , <NUM_LIT:2> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . write_count , <NUM_LIT:3> ) <EOL> self . assertEqual ( ret . write_bytes , <NUM_LIT:4> * SECTOR_SIZE ) <EOL> self . assertEqual ( ret . read_merged_count , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . read_time , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . write_merged_count , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . write_time , <NUM_LIT:0> ) <EOL> self . assertEqual ( ret . busy_time , <NUM_LIT:0> ) <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestMisc ( unittest . TestCase ) : <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_no_procfs_on_import ( self , tb ) : <EOL> my_procfs = tempfile . mkdtemp ( ) <EOL> with open ( os . path . join ( my_procfs , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> try : <EOL> orig_open = open <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> raise IOError ( errno . ENOENT , '<STR_LIT>' ) <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) : <EOL> importlib . reload ( psutil ) <EOL> assert tb . called <EOL> self . assertRaises ( IOError , psutil . cpu_times ) <EOL> self . assertRaises ( IOError , psutil . cpu_times , percpu = True ) <EOL> self . assertRaises ( IOError , psutil . cpu_percent ) <EOL> self . assertRaises ( IOError , psutil . cpu_percent , percpu = True ) <EOL> self . assertRaises ( IOError , psutil . cpu_times_percent ) <EOL> self . assertRaises ( <EOL> IOError , psutil . cpu_times_percent , percpu = True ) <EOL> psutil . PROCFS_PATH = my_procfs <EOL> self . assertEqual ( psutil . cpu_percent ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( sum ( psutil . cpu_times_percent ( ) ) , <NUM_LIT:0> ) <EOL> per_cpu_percent = psutil . cpu_percent ( percpu = True ) <EOL> self . assertEqual ( sum ( per_cpu_percent ) , <NUM_LIT:0> ) <EOL> per_cpu_times_percent = psutil . cpu_times_percent ( percpu = True ) <EOL> self . assertEqual ( sum ( map ( sum , per_cpu_times_percent ) ) , <NUM_LIT:0> ) <EOL> with open ( os . path . join ( my_procfs , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> f . write ( '<STR_LIT>' ) <EOL> self . assertNotEqual ( psutil . cpu_percent ( ) , <NUM_LIT:0> ) <EOL> self . assertNotEqual ( <EOL> sum ( psutil . cpu_percent ( percpu = True ) ) , <NUM_LIT:0> ) <EOL> self . assertNotEqual ( sum ( psutil . cpu_times_percent ( ) ) , <NUM_LIT:0> ) <EOL> self . assertNotEqual ( <EOL> sum ( map ( sum , psutil . cpu_times_percent ( percpu = True ) ) ) , <NUM_LIT:0> ) <EOL> finally : <EOL> shutil . rmtree ( my_procfs ) <EOL> importlib . reload ( psutil ) <EOL> self . assertEqual ( psutil . PROCFS_PATH , '<STR_LIT>' ) <EOL> @ unittest . skipUnless ( <EOL> get_kernel_version ( ) >= ( <NUM_LIT:2> , <NUM_LIT:6> , <NUM_LIT> ) , <EOL> "<STR_LIT>" ) <EOL> def test_prlimit_availability ( self ) : <EOL> p = psutil . Process ( os . getpid ( ) ) <EOL> p . rlimit ( psutil . RLIMIT_NOFILE ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> @ unittest . skipUnless ( <EOL> get_kernel_version ( ) >= ( <NUM_LIT:3> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) <EOL> def test_resource_consts_kernel_v ( self ) : <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> self . assertTrue ( hasattr ( psutil , "<STR_LIT>" ) ) <EOL> def test_boot_time_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> RuntimeError , <EOL> psutil . _pslinux . boot_time ) <EOL> assert m . called <EOL> def test_users_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = [ ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> <NUM_LIT> , True ) ] ) as m : <EOL> self . assertEqual ( psutil . users ( ) [ <NUM_LIT:0> ] . host , '<STR_LIT:localhost>' ) <EOL> assert m . called <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = [ ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> <NUM_LIT> , True ) ] ) as m : <EOL> self . assertEqual ( psutil . users ( ) [ <NUM_LIT:0> ] . host , '<STR_LIT:localhost>' ) <EOL> assert m . called <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = [ ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:foo>' , <EOL> <NUM_LIT> , True ) ] ) as m : <EOL> self . assertEqual ( psutil . users ( ) [ <NUM_LIT:0> ] . host , '<STR_LIT:foo>' ) <EOL> assert m . called <EOL> def test_procfs_path ( self ) : <EOL> tdir = tempfile . mkdtemp ( ) <EOL> try : <EOL> psutil . PROCFS_PATH = tdir <EOL> self . assertRaises ( IOError , psutil . virtual_memory ) <EOL> self . assertRaises ( IOError , psutil . cpu_times ) <EOL> self . assertRaises ( IOError , psutil . cpu_times , percpu = True ) <EOL> self . assertRaises ( IOError , psutil . boot_time ) <EOL> self . assertRaises ( IOError , psutil . net_connections ) <EOL> self . assertRaises ( IOError , psutil . net_io_counters ) <EOL> self . assertRaises ( IOError , psutil . net_if_stats ) <EOL> self . assertRaises ( IOError , psutil . disk_io_counters ) <EOL> self . assertRaises ( IOError , psutil . disk_partitions ) <EOL> self . assertRaises ( psutil . NoSuchProcess , psutil . Process ) <EOL> finally : <EOL> psutil . PROCFS_PATH = "<STR_LIT>" <EOL> os . rmdir ( tdir ) <EOL> def test_sector_size_mock ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if PY3 and isinstance ( name , bytes ) : <EOL> name = name . decode ( ) <EOL> if name . startswith ( "<STR_LIT>" ) : <EOL> flag . append ( None ) <EOL> raise IOError ( errno . ENOENT , '<STR_LIT>' ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> flag = [ ] <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> try : <EOL> with mock . patch ( patch_point , side_effect = open_mock ) : <EOL> importlib . reload ( psutil . _pslinux ) <EOL> importlib . reload ( psutil ) <EOL> self . assertEqual ( flag , [ None ] ) <EOL> self . assertEqual ( psutil . _pslinux . SECTOR_SIZE , <NUM_LIT> ) <EOL> finally : <EOL> importlib . reload ( psutil . _pslinux ) <EOL> importlib . reload ( psutil ) <EOL> @ unittest . skipUnless ( LINUX , "<STR_LIT>" ) <EOL> class TestProcess ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> safe_remove ( TESTFN ) <EOL> tearDown = setUp <EOL> def test_memory_maps ( self ) : <EOL> src = textwrap . dedent ( """<STR_LIT>""" % TESTFN ) <EOL> sproc = pyrun ( src ) <EOL> self . addCleanup ( reap_children ) <EOL> call_until ( lambda : os . listdir ( '<STR_LIT:.>' ) , "<STR_LIT>" % TESTFN ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> maps = p . memory_maps ( grouped = False ) <EOL> pmap = sh ( '<STR_LIT>' % p . pid ) . split ( '<STR_LIT:\n>' ) <EOL> del pmap [ <NUM_LIT:0> ] <EOL> del pmap [ <NUM_LIT:0> ] <EOL> while maps and pmap : <EOL> this = maps . pop ( <NUM_LIT:0> ) <EOL> other = pmap . pop ( <NUM_LIT:0> ) <EOL> addr , _ , rss , dirty , mode , path = other . split ( None , <NUM_LIT:5> ) <EOL> if not path . startswith ( '<STR_LIT:[>' ) and not path . endswith ( '<STR_LIT:]>' ) : <EOL> self . assertEqual ( path , os . path . basename ( this . path ) ) <EOL> self . assertEqual ( int ( rss ) * <NUM_LIT> , this . rss ) <EOL> self . assertEqual ( mode [ : <NUM_LIT:3> ] , this . perms [ : <NUM_LIT:3> ] ) <EOL> def test_memory_full_info ( self ) : <EOL> src = textwrap . dedent ( """<STR_LIT>""" % TESTFN ) <EOL> sproc = pyrun ( src ) <EOL> self . addCleanup ( reap_children ) <EOL> call_until ( lambda : os . listdir ( '<STR_LIT:.>' ) , "<STR_LIT>" % TESTFN ) <EOL> p = psutil . Process ( sproc . pid ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> mem = p . memory_full_info ( ) <EOL> maps = p . memory_maps ( grouped = False ) <EOL> self . assertAlmostEqual ( <EOL> mem . uss , sum ( [ x . private_dirty + x . private_clean for x in maps ] ) , <EOL> delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> mem . pss , sum ( [ x . pss for x in maps ] ) , delta = <NUM_LIT> ) <EOL> self . assertAlmostEqual ( <EOL> mem . swap , sum ( [ x . swap for x in maps ] ) , delta = <NUM_LIT> ) <EOL> @ unittest . skipIf ( PYPY , "<STR_LIT>" ) <EOL> def test_open_files_mode ( self ) : <EOL> def get_test_file ( ) : <EOL> p = psutil . Process ( ) <EOL> giveup_at = time . time ( ) + <NUM_LIT:2> <EOL> while True : <EOL> for file in p . open_files ( ) : <EOL> if file . path == os . path . abspath ( TESTFN ) : <EOL> return file <EOL> elif time . time ( ) > giveup_at : <EOL> break <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> with open ( TESTFN , "<STR_LIT:w>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT:w>" ) <EOL> with open ( TESTFN , "<STR_LIT:r>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT:r>" ) <EOL> with open ( TESTFN , "<STR_LIT:a>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT:a>" ) <EOL> with open ( TESTFN , "<STR_LIT>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT>" ) <EOL> with open ( TESTFN , "<STR_LIT>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT>" ) <EOL> with open ( TESTFN , "<STR_LIT>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT>" ) <EOL> if PY3 : <EOL> safe_remove ( TESTFN ) <EOL> with open ( TESTFN , "<STR_LIT:x>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT:w>" ) <EOL> safe_remove ( TESTFN ) <EOL> with open ( TESTFN , "<STR_LIT>" ) : <EOL> self . assertEqual ( get_test_file ( ) . mode , "<STR_LIT>" ) <EOL> def test_open_files_file_gone ( self ) : <EOL> p = psutil . Process ( ) <EOL> files = p . open_files ( ) <EOL> with tempfile . NamedTemporaryFile ( ) : <EOL> call_until ( p . open_files , "<STR_LIT>" % len ( files ) ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> side_effect = OSError ( errno . ENOENT , "<STR_LIT>" ) ) as m : <EOL> files = p . open_files ( ) <EOL> assert not files <EOL> assert m . called <EOL> with mock . patch ( '<STR_LIT>' , <EOL> side_effect = OSError ( errno . EINVAL , "<STR_LIT>" ) ) as m : <EOL> self . assertEqual ( p . open_files ( ) , [ ] ) <EOL> assert m . called <EOL> def test_terminal_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = { } ) as m : <EOL> self . assertIsNone ( psutil . _pslinux . Process ( os . getpid ( ) ) . terminal ( ) ) <EOL> assert m . called <EOL> def test_num_ctx_switches_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . num_ctx_switches ) <EOL> assert m . called <EOL> def test_num_threads_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . num_threads ) <EOL> assert m . called <EOL> def test_ppid_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . ppid ) <EOL> assert m . called <EOL> def test_uids_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . uids ) <EOL> assert m . called <EOL> def test_gids_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . gids ) <EOL> assert m . called <EOL> def test_cmdline_mocked ( self ) : <EOL> p = psutil . Process ( ) <EOL> fake_file = io . StringIO ( u ( '<STR_LIT>' ) ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_file , create = True ) as m : <EOL> p . cmdline ( ) == [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] <EOL> assert m . called <EOL> fake_file = io . StringIO ( u ( '<STR_LIT>' ) ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = fake_file , create = True ) as m : <EOL> p . cmdline ( ) == [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ] <EOL> assert m . called <EOL> def test_io_counters_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) as m : <EOL> self . assertRaises ( <EOL> NotImplementedError , <EOL> psutil . _pslinux . Process ( os . getpid ( ) ) . io_counters ) <EOL> assert m . called <EOL> def test_readlink_path_deleted_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = '<STR_LIT>' ) : <EOL> self . assertEqual ( psutil . Process ( ) . exe ( ) , "<STR_LIT>" ) <EOL> self . assertEqual ( psutil . Process ( ) . cwd ( ) , "<STR_LIT>" ) <EOL> def test_threads_mocked ( self ) : <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name . startswith ( '<STR_LIT>' % os . getpid ( ) ) : <EOL> raise IOError ( errno . ENOENT , "<STR_LIT>" ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> orig_open = open <EOL> patch_point = '<STR_LIT>' if PY3 else '<STR_LIT>' <EOL> with mock . patch ( patch_point , side_effect = open_mock ) as m : <EOL> ret = psutil . Process ( ) . threads ( ) <EOL> assert m . called <EOL> self . assertEqual ( ret , [ ] ) <EOL> def open_mock ( name , * args , ** kwargs ) : <EOL> if name . startswith ( '<STR_LIT>' % os . getpid ( ) ) : <EOL> raise IOError ( errno . EPERM , "<STR_LIT>" ) <EOL> else : <EOL> return orig_open ( name , * args , ** kwargs ) <EOL> with mock . patch ( patch_point , side_effect = open_mock ) : <EOL> self . assertRaises ( psutil . AccessDenied , psutil . Process ( ) . threads ) <EOL> @ unittest . skipIf ( TRAVIS , "<STR_LIT>" ) <EOL> def test_exe_mocked ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> side_effect = OSError ( errno . ENOENT , "<STR_LIT>" ) ) as m : <EOL> ret = psutil . Process ( ) . exe ( ) <EOL> assert m . called <EOL> self . assertEqual ( ret , "<STR_LIT>" ) <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = False ) : <EOL> self . assertRaises ( psutil . ZombieProcess , psutil . Process ( ) . exe ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> run_test_module_by_name ( __file__ ) </s>
