| |
| |
|
|
| |
| |
| |
|
|
| """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: |
| * Python 3: sys.getfilesystemencodeerrors() (PY 3.6+) or |
| "surrogatescape" on POSIX and "replace" on Windows |
| * Python 2: "replace" |
| * on Python 2 all APIs return bytes (str type), never unicode |
| * on Python 2, you can go back to unicode by doing: |
| |
| >>> unicode(p.exe(), sys.getdefaultencoding(), errors="replace") |
| |
| 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.connections('unix') |
| * Process.cwd() |
| * Process.environ() |
| * Process.exe() |
| * Process.memory_maps() |
| * Process.name() |
| * 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 traceback |
| import unittest |
| import warnings |
| from contextlib import closing |
|
|
| import psutil |
| from psutil import BSD |
| from psutil import POSIX |
| from psutil import WINDOWS |
| from psutil._compat import PY3 |
| from psutil._compat import super |
| from psutil._compat import u |
| from psutil.tests import APPVEYOR |
| from psutil.tests import ASCII_FS |
| from psutil.tests import CI_TESTING |
| from psutil.tests import HAS_CONNECTIONS_UNIX |
| from psutil.tests import HAS_ENVIRON |
| from psutil.tests import HAS_MEMORY_MAPS |
| 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 safe_mkdir |
| from psutil.tests import safe_rmpath |
| from psutil.tests import serialrun |
| from psutil.tests import skip_on_access_denied |
| from psutil.tests import spawn_testproc |
| from psutil.tests import terminate |
|
|
|
|
| if APPVEYOR: |
|
|
| def safe_rmpath(path): |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| from psutil.tests import safe_rmpath as rm |
|
|
| try: |
| return rm(path) |
| except WindowsError: |
| traceback.print_exc() |
|
|
|
|
| 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, IOError): |
| 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 self.skipTest("can't handle unicode str") |
|
|
|
|
| @serialrun |
| @unittest.skipIf(ASCII_FS, "ASCII fs") |
| @unittest.skipIf(PYPY and not PY3, "too much trouble on PYPY2") |
| class TestFSAPIs(BaseUnicodeTest): |
| """Test FS APIs with a funky, valid, UTF8 path name.""" |
|
|
| funky_suffix = UNICODE_SUFFIX |
|
|
| def expect_exact_path_match(self): |
| |
| |
| here = '.' if isinstance(self.funky_name, str) else u('.') |
| with warnings.catch_warnings(): |
| warnings.simplefilter("ignore") |
| return self.funky_name in os.listdir(here) |
|
|
| |
|
|
| def test_proc_exe(self): |
| cmd = [self.funky_name, "-c", "import time; time.sleep(10)"] |
| subp = self.spawn_testproc(cmd) |
| p = psutil.Process(subp.pid) |
| exe = p.exe() |
| self.assertIsInstance(exe, str) |
| if self.expect_exact_path_match(): |
| self.assertEqual( |
| os.path.normcase(exe), os.path.normcase(self.funky_name) |
| ) |
|
|
| def test_proc_name(self): |
| cmd = [self.funky_name, "-c", "import time; time.sleep(10)"] |
| subp = self.spawn_testproc(cmd) |
| name = psutil.Process(subp.pid).name() |
| self.assertIsInstance(name, str) |
| if self.expect_exact_path_match(): |
| self.assertEqual(name, os.path.basename(self.funky_name)) |
|
|
| def test_proc_cmdline(self): |
| cmd = [self.funky_name, "-c", "import time; time.sleep(10)"] |
| subp = self.spawn_testproc(cmd) |
| p = psutil.Process(subp.pid) |
| cmdline = p.cmdline() |
| for part in cmdline: |
| self.assertIsInstance(part, str) |
| if self.expect_exact_path_match(): |
| self.assertEqual(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() |
| self.assertIsInstance(p.cwd(), str) |
| if self.expect_exact_path_match(): |
| self.assertEqual(cwd, dname) |
|
|
| @unittest.skipIf(PYPY and WINDOWS, "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 |
| self.assertIsInstance(path, str) |
| if BSD and not path: |
| |
| return self.skipTest("open_files on BSD is broken") |
| if self.expect_exact_path_match(): |
| self.assertEqual( |
| os.path.normcase(path), os.path.normcase(self.funky_name) |
| ) |
|
|
| @unittest.skipIf(not POSIX, "POSIX only") |
| def test_proc_connections(self): |
| name = self.get_testfn(suffix=self.funky_suffix) |
| try: |
| sock = bind_unix_socket(name) |
| except UnicodeEncodeError: |
| if PY3: |
| raise |
| else: |
| raise unittest.SkipTest("not supported") |
| with closing(sock): |
| conn = psutil.Process().connections('unix')[0] |
| self.assertIsInstance(conn.laddr, str) |
| self.assertEqual(conn.laddr, name) |
|
|
| @unittest.skipIf(not POSIX, "POSIX only") |
| @unittest.skipIf(not HAS_CONNECTIONS_UNIX, "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) |
| try: |
| sock = bind_unix_socket(name) |
| except UnicodeEncodeError: |
| if PY3: |
| raise |
| else: |
| raise unittest.SkipTest("not supported") |
| with closing(sock): |
| cons = psutil.net_connections(kind='unix') |
| conn = find_sock(cons) |
| self.assertIsInstance(conn.laddr, str) |
| self.assertEqual(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) |
|
|
| @unittest.skipIf(not HAS_MEMORY_MAPS, "not supported") |
| @unittest.skipIf(not PY3, "ctypes does not support unicode on PY2") |
| @unittest.skipIf(PYPY, "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] |
| self.assertIn(normpath(funky_path), libpaths) |
| for path in libpaths: |
| self.assertIsInstance(path, str) |
|
|
|
|
| @unittest.skipIf(CI_TESTING, "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 if PY3 else 'è' |
|
|
| @unittest.skipIf(not HAS_ENVIRON, "not supported") |
| @unittest.skipIf(PYPY and WINDOWS, "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(): |
| self.assertIsInstance(k, str) |
| self.assertIsInstance(v, str) |
| self.assertEqual(env['FUNNY_ARG'], self.funky_suffix) |
|
|
|
|
| if __name__ == '__main__': |
| from psutil.tests.runner import run_from_name |
|
|
| run_from_name(__file__) |
|
|
