|
|
|
|
|
|
|
|
|
|
|
|
| """Notes about unicode handling in psutil
|
| ======================================.
|
|
|
| Starting from version 5.3.0 psutil adds unicode support, see:
|
| https://github.com/giampaolo/psutil/issues/1040
|
| The notes below apply to *any* API returning a string such as
|
| process exe(), cwd() or username():
|
|
|
| * all strings are encoded by using the OS filesystem encoding
|
| (sys.getfilesystemencoding()) which varies depending on the platform
|
| (e.g. "UTF-8" on macOS, "mbcs" on Win)
|
| * no API call is supposed to crash with UnicodeDecodeError
|
| * instead, in case of badly encoded data returned by the OS, the
|
| following error handlers are used to replace the corrupted characters in
|
| the string:
|
| * sys.getfilesystemencodeerrors() or "surrogatescape" on POSIX and
|
| "replace" on Windows.
|
|
|
| For a detailed explanation of how psutil handles unicode see #1040.
|
|
|
| Tests
|
| =====
|
|
|
| List of APIs returning or dealing with a string:
|
| ('not tested' means they are not tested to deal with non-ASCII strings):
|
|
|
| * Process.cmdline()
|
| * Process.cwd()
|
| * Process.environ()
|
| * Process.exe()
|
| * Process.memory_maps()
|
| * Process.name()
|
| * Process.net_connections('unix')
|
| * Process.open_files()
|
| * Process.username() (not tested)
|
|
|
| * disk_io_counters() (not tested)
|
| * disk_partitions() (not tested)
|
| * disk_usage(str)
|
| * net_connections('unix')
|
| * net_if_addrs() (not tested)
|
| * net_if_stats() (not tested)
|
| * net_io_counters() (not tested)
|
| * sensors_fans() (not tested)
|
| * sensors_temperatures() (not tested)
|
| * users() (not tested)
|
|
|
| * WindowsService.binpath() (not tested)
|
| * WindowsService.description() (not tested)
|
| * WindowsService.display_name() (not tested)
|
| * WindowsService.name() (not tested)
|
| * WindowsService.status() (not tested)
|
| * WindowsService.username() (not tested)
|
|
|
| In here we create a unicode path with a funky non-ASCII name and (where
|
| possible) make psutil return it back (e.g. on name(), exe(), open_files(),
|
| etc.) and make sure that:
|
|
|
| * psutil never crashes with UnicodeDecodeError
|
| * the returned path matches
|
| """
|
|
|
| import os
|
| import shutil
|
| import warnings
|
| from contextlib import closing
|
|
|
| import psutil
|
| from psutil import BSD
|
| from psutil import POSIX
|
| from psutil import WINDOWS
|
| from psutil.tests import ASCII_FS
|
| from psutil.tests import CI_TESTING
|
| from psutil.tests import HAS_ENVIRON
|
| from psutil.tests import HAS_MEMORY_MAPS
|
| from psutil.tests import HAS_NET_CONNECTIONS_UNIX
|
| from psutil.tests import INVALID_UNICODE_SUFFIX
|
| from psutil.tests import PYPY
|
| from psutil.tests import TESTFN_PREFIX
|
| from psutil.tests import UNICODE_SUFFIX
|
| from psutil.tests import PsutilTestCase
|
| from psutil.tests import bind_unix_socket
|
| from psutil.tests import chdir
|
| from psutil.tests import copyload_shared_lib
|
| from psutil.tests import create_py_exe
|
| from psutil.tests import get_testfn
|
| from psutil.tests import pytest
|
| from psutil.tests import safe_mkdir
|
| from psutil.tests import safe_rmpath
|
| from psutil.tests import skip_on_access_denied
|
| from psutil.tests import spawn_testproc
|
| from psutil.tests import terminate
|
|
|
|
|
| def try_unicode(suffix):
|
| """Return True if both the fs and the subprocess module can
|
| deal with a unicode file name.
|
| """
|
| sproc = None
|
| testfn = get_testfn(suffix=suffix)
|
| try:
|
| safe_rmpath(testfn)
|
| create_py_exe(testfn)
|
| sproc = spawn_testproc(cmd=[testfn])
|
| shutil.copyfile(testfn, testfn + '-2')
|
| safe_rmpath(testfn + '-2')
|
| except (UnicodeEncodeError, OSError):
|
| return False
|
| else:
|
| return True
|
| finally:
|
| if sproc is not None:
|
| terminate(sproc)
|
| safe_rmpath(testfn)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| class BaseUnicodeTest(PsutilTestCase):
|
| funky_suffix = None
|
|
|
| @classmethod
|
| def setUpClass(cls):
|
| super().setUpClass()
|
| cls.skip_tests = False
|
| cls.funky_name = None
|
| if cls.funky_suffix is not None:
|
| if not try_unicode(cls.funky_suffix):
|
| cls.skip_tests = True
|
| else:
|
| cls.funky_name = get_testfn(suffix=cls.funky_suffix)
|
| create_py_exe(cls.funky_name)
|
|
|
| def setUp(self):
|
| super().setUp()
|
| if self.skip_tests:
|
| raise pytest.skip("can't handle unicode str")
|
|
|
|
|
| @pytest.mark.xdist_group(name="serial")
|
| @pytest.mark.skipif(ASCII_FS, reason="ASCII fs")
|
| class TestFSAPIs(BaseUnicodeTest):
|
| """Test FS APIs with a funky, valid, UTF8 path name."""
|
|
|
| funky_suffix = UNICODE_SUFFIX
|
|
|
| def expect_exact_path_match(self):
|
| with warnings.catch_warnings():
|
| warnings.simplefilter("ignore")
|
| return self.funky_name in os.listdir(".")
|
|
|
|
|
|
|
| def test_proc_exe(self):
|
| cmd = [
|
| self.funky_name,
|
| "-c",
|
| "import time; [time.sleep(0.1) for x in range(100)]",
|
| ]
|
| subp = self.spawn_testproc(cmd)
|
| p = psutil.Process(subp.pid)
|
| exe = p.exe()
|
| assert isinstance(exe, str)
|
| if self.expect_exact_path_match():
|
| assert os.path.normcase(exe) == os.path.normcase(self.funky_name)
|
|
|
| def test_proc_name(self):
|
| cmd = [
|
| self.funky_name,
|
| "-c",
|
| "import time; [time.sleep(0.1) for x in range(100)]",
|
| ]
|
| subp = self.spawn_testproc(cmd)
|
| name = psutil.Process(subp.pid).name()
|
| assert isinstance(name, str)
|
| if self.expect_exact_path_match():
|
| assert name == os.path.basename(self.funky_name)
|
|
|
| def test_proc_cmdline(self):
|
| cmd = [
|
| self.funky_name,
|
| "-c",
|
| "import time; [time.sleep(0.1) for x in range(100)]",
|
| ]
|
| subp = self.spawn_testproc(cmd)
|
| p = psutil.Process(subp.pid)
|
| cmdline = p.cmdline()
|
| for part in cmdline:
|
| assert isinstance(part, str)
|
| if self.expect_exact_path_match():
|
| assert cmdline == cmd
|
|
|
| def test_proc_cwd(self):
|
| dname = self.funky_name + "2"
|
| self.addCleanup(safe_rmpath, dname)
|
| safe_mkdir(dname)
|
| with chdir(dname):
|
| p = psutil.Process()
|
| cwd = p.cwd()
|
| assert isinstance(p.cwd(), str)
|
| if self.expect_exact_path_match():
|
| assert cwd == dname
|
|
|
| @pytest.mark.skipif(PYPY and WINDOWS, reason="fails on PYPY + WINDOWS")
|
| def test_proc_open_files(self):
|
| p = psutil.Process()
|
| start = set(p.open_files())
|
| with open(self.funky_name, 'rb'):
|
| new = set(p.open_files())
|
| path = (new - start).pop().path
|
| assert isinstance(path, str)
|
| if BSD and not path:
|
|
|
| raise pytest.skip("open_files on BSD is broken")
|
| if self.expect_exact_path_match():
|
| assert os.path.normcase(path) == os.path.normcase(self.funky_name)
|
|
|
| @pytest.mark.skipif(not POSIX, reason="POSIX only")
|
| def test_proc_net_connections(self):
|
| name = self.get_testfn(suffix=self.funky_suffix)
|
| sock = bind_unix_socket(name)
|
| with closing(sock):
|
| conn = psutil.Process().net_connections('unix')[0]
|
| assert isinstance(conn.laddr, str)
|
| assert conn.laddr == name
|
|
|
| @pytest.mark.skipif(not POSIX, reason="POSIX only")
|
| @pytest.mark.skipif(
|
| not HAS_NET_CONNECTIONS_UNIX, reason="can't list UNIX sockets"
|
| )
|
| @skip_on_access_denied()
|
| def test_net_connections(self):
|
| def find_sock(cons):
|
| for conn in cons:
|
| if os.path.basename(conn.laddr).startswith(TESTFN_PREFIX):
|
| return conn
|
| raise ValueError("connection not found")
|
|
|
| name = self.get_testfn(suffix=self.funky_suffix)
|
| sock = bind_unix_socket(name)
|
| with closing(sock):
|
| cons = psutil.net_connections(kind='unix')
|
| conn = find_sock(cons)
|
| assert isinstance(conn.laddr, str)
|
| assert conn.laddr == name
|
|
|
| def test_disk_usage(self):
|
| dname = self.funky_name + "2"
|
| self.addCleanup(safe_rmpath, dname)
|
| safe_mkdir(dname)
|
| psutil.disk_usage(dname)
|
|
|
| @pytest.mark.skipif(not HAS_MEMORY_MAPS, reason="not supported")
|
| @pytest.mark.skipif(PYPY, reason="unstable on PYPY")
|
| def test_memory_maps(self):
|
| with copyload_shared_lib(suffix=self.funky_suffix) as funky_path:
|
|
|
| def normpath(p):
|
| return os.path.realpath(os.path.normcase(p))
|
|
|
| libpaths = [
|
| normpath(x.path) for x in psutil.Process().memory_maps()
|
| ]
|
|
|
| libpaths = [x for x in libpaths if TESTFN_PREFIX in x]
|
| assert normpath(funky_path) in libpaths
|
| for path in libpaths:
|
| assert isinstance(path, str)
|
|
|
|
|
| @pytest.mark.skipif(CI_TESTING, reason="unreliable on CI")
|
| class TestFSAPIsWithInvalidPath(TestFSAPIs):
|
| """Test FS APIs with a funky, invalid path name."""
|
|
|
| funky_suffix = INVALID_UNICODE_SUFFIX
|
|
|
| def expect_exact_path_match(self):
|
| return True
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestNonFSAPIS(BaseUnicodeTest):
|
| """Unicode tests for non fs-related APIs."""
|
|
|
| funky_suffix = UNICODE_SUFFIX
|
|
|
| @pytest.mark.skipif(not HAS_ENVIRON, reason="not supported")
|
| @pytest.mark.skipif(PYPY and WINDOWS, reason="segfaults on PYPY + WINDOWS")
|
| def test_proc_environ(self):
|
|
|
|
|
| env = os.environ.copy()
|
| env['FUNNY_ARG'] = self.funky_suffix
|
| sproc = self.spawn_testproc(env=env)
|
| p = psutil.Process(sproc.pid)
|
| env = p.environ()
|
| for k, v in env.items():
|
| assert isinstance(k, str)
|
| assert isinstance(v, str)
|
| assert env['FUNNY_ARG'] == self.funky_suffix
|
|
|
