diff options
Diffstat (limited to 'python/psutil/test/test_memory_leaks.py')
| m--------- | python/psutil | 0 | ||||
| -rw-r--r-- | python/psutil/test/test_memory_leaks.py | 445 |
2 files changed, 0 insertions, 445 deletions
diff --git a/python/psutil b/python/psutil new file mode 160000 +Subproject a0967043b5819b2edc61d9a12306289d5e7f98c diff --git a/python/psutil/test/test_memory_leaks.py b/python/psutil/test/test_memory_leaks.py deleted file mode 100644 index 6f02dc0ac..000000000 --- a/python/psutil/test/test_memory_leaks.py +++ /dev/null @@ -1,445 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -""" -A test script which attempts to detect memory leaks by calling C -functions many times and compare process memory usage before and -after the calls. It might produce false positives. -""" - -import functools -import gc -import os -import socket -import sys -import threading -import time - -import psutil -import psutil._common - -from psutil._compat import xrange, callable -from test_psutil import (WINDOWS, POSIX, OSX, LINUX, SUNOS, BSD, TESTFN, - RLIMIT_SUPPORT, TRAVIS) -from test_psutil import (reap_children, supports_ipv6, safe_remove, - get_test_subprocess) - -if sys.version_info < (2, 7): - import unittest2 as unittest # https://pypi.python.org/pypi/unittest2 -else: - import unittest - - -LOOPS = 1000 -TOLERANCE = 4096 -SKIP_PYTHON_IMPL = True - - -def skip_if_linux(): - return unittest.skipIf(LINUX and SKIP_PYTHON_IMPL, - "not worth being tested on LINUX (pure python)") - - -class Base(unittest.TestCase): - proc = psutil.Process() - - def execute(self, function, *args, **kwargs): - def call_many_times(): - for x in xrange(LOOPS - 1): - self.call(function, *args, **kwargs) - del x - gc.collect() - return self.get_mem() - - self.call(function, *args, **kwargs) - self.assertEqual(gc.garbage, []) - self.assertEqual(threading.active_count(), 1) - - # RSS comparison - # step 1 - rss1 = call_many_times() - # step 2 - rss2 = call_many_times() - - difference = rss2 - rss1 - if difference > TOLERANCE: - # This doesn't necessarily mean we have a leak yet. - # At this point we assume that after having called the - # function so many times the memory usage is stabilized - # and if there are no leaks it should not increase any - # more. - # Let's keep calling fun for 3 more seconds and fail if - # we notice any difference. - stop_at = time.time() + 3 - while True: - self.call(function, *args, **kwargs) - if time.time() >= stop_at: - break - del stop_at - gc.collect() - rss3 = self.get_mem() - difference = rss3 - rss2 - if rss3 > rss2: - self.fail("rss2=%s, rss3=%s, difference=%s" - % (rss2, rss3, difference)) - - def execute_w_exc(self, exc, function, *args, **kwargs): - kwargs['_exc'] = exc - self.execute(function, *args, **kwargs) - - def get_mem(self): - return psutil.Process().memory_info()[0] - - def call(self, function, *args, **kwargs): - raise NotImplementedError("must be implemented in subclass") - - -class TestProcessObjectLeaks(Base): - """Test leaks of Process class methods and properties""" - - def setUp(self): - gc.collect() - - def tearDown(self): - reap_children() - - def call(self, function, *args, **kwargs): - if callable(function): - if '_exc' in kwargs: - exc = kwargs.pop('_exc') - self.assertRaises(exc, function, *args, **kwargs) - else: - try: - function(*args, **kwargs) - except psutil.Error: - pass - else: - meth = getattr(self.proc, function) - if '_exc' in kwargs: - exc = kwargs.pop('_exc') - self.assertRaises(exc, meth, *args, **kwargs) - else: - try: - meth(*args, **kwargs) - except psutil.Error: - pass - - @skip_if_linux() - def test_name(self): - self.execute('name') - - @skip_if_linux() - def test_cmdline(self): - self.execute('cmdline') - - @skip_if_linux() - def test_exe(self): - self.execute('exe') - - @skip_if_linux() - def test_ppid(self): - self.execute('ppid') - - @unittest.skipUnless(POSIX, "POSIX only") - @skip_if_linux() - def test_uids(self): - self.execute('uids') - - @unittest.skipUnless(POSIX, "POSIX only") - @skip_if_linux() - def test_gids(self): - self.execute('gids') - - @skip_if_linux() - def test_status(self): - self.execute('status') - - def test_nice_get(self): - self.execute('nice') - - def test_nice_set(self): - niceness = psutil.Process().nice() - self.execute('nice', niceness) - - @unittest.skipUnless(hasattr(psutil.Process, 'ionice'), - "Linux and Windows Vista only") - def test_ionice_get(self): - self.execute('ionice') - - @unittest.skipUnless(hasattr(psutil.Process, 'ionice'), - "Linux and Windows Vista only") - def test_ionice_set(self): - if WINDOWS: - value = psutil.Process().ionice() - self.execute('ionice', value) - else: - from psutil._pslinux import cext - self.execute('ionice', psutil.IOPRIO_CLASS_NONE) - fun = functools.partial(cext.proc_ioprio_set, os.getpid(), -1, 0) - self.execute_w_exc(OSError, fun) - - @unittest.skipIf(OSX or SUNOS, "feature not supported on this platform") - @skip_if_linux() - def test_io_counters(self): - self.execute('io_counters') - - @unittest.skipUnless(WINDOWS, "not worth being tested on posix") - def test_username(self): - self.execute('username') - - @skip_if_linux() - def test_create_time(self): - self.execute('create_time') - - @skip_if_linux() - def test_num_threads(self): - self.execute('num_threads') - - @unittest.skipUnless(WINDOWS, "Windows only") - def test_num_handles(self): - self.execute('num_handles') - - @unittest.skipUnless(POSIX, "POSIX only") - @skip_if_linux() - def test_num_fds(self): - self.execute('num_fds') - - @skip_if_linux() - def test_threads(self): - self.execute('threads') - - @skip_if_linux() - def test_cpu_times(self): - self.execute('cpu_times') - - @skip_if_linux() - def test_memory_info(self): - self.execute('memory_info') - - @skip_if_linux() - def test_memory_info_ex(self): - self.execute('memory_info_ex') - - @unittest.skipUnless(POSIX, "POSIX only") - @skip_if_linux() - def test_terminal(self): - self.execute('terminal') - - @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, - "not worth being tested on POSIX (pure python)") - def test_resume(self): - self.execute('resume') - - @skip_if_linux() - def test_cwd(self): - self.execute('cwd') - - @unittest.skipUnless(WINDOWS or LINUX or BSD, - "Windows or Linux or BSD only") - def test_cpu_affinity_get(self): - self.execute('cpu_affinity') - - @unittest.skipUnless(WINDOWS or LINUX or BSD, - "Windows or Linux or BSD only") - def test_cpu_affinity_set(self): - affinity = psutil.Process().cpu_affinity() - self.execute('cpu_affinity', affinity) - if not TRAVIS: - self.execute_w_exc(ValueError, 'cpu_affinity', [-1]) - - @skip_if_linux() - def test_open_files(self): - safe_remove(TESTFN) # needed after UNIX socket test has run - with open(TESTFN, 'w'): - self.execute('open_files') - - # OSX implementation is unbelievably slow - @unittest.skipIf(OSX, "OSX implementation is too slow") - @skip_if_linux() - def test_memory_maps(self): - self.execute('memory_maps') - - @unittest.skipUnless(LINUX, "Linux only") - @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, - "only available on Linux >= 2.6.36") - def test_rlimit_get(self): - self.execute('rlimit', psutil.RLIMIT_NOFILE) - - @unittest.skipUnless(LINUX, "Linux only") - @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, - "only available on Linux >= 2.6.36") - def test_rlimit_set(self): - limit = psutil.Process().rlimit(psutil.RLIMIT_NOFILE) - self.execute('rlimit', psutil.RLIMIT_NOFILE, limit) - self.execute_w_exc(OSError, 'rlimit', -1) - - @skip_if_linux() - # Windows implementation is based on a single system-wide function - @unittest.skipIf(WINDOWS, "tested later") - def test_connections(self): - def create_socket(family, type): - sock = socket.socket(family, type) - sock.bind(('', 0)) - if type == socket.SOCK_STREAM: - sock.listen(1) - return sock - - socks = [] - socks.append(create_socket(socket.AF_INET, socket.SOCK_STREAM)) - socks.append(create_socket(socket.AF_INET, socket.SOCK_DGRAM)) - if supports_ipv6(): - socks.append(create_socket(socket.AF_INET6, socket.SOCK_STREAM)) - socks.append(create_socket(socket.AF_INET6, socket.SOCK_DGRAM)) - if hasattr(socket, 'AF_UNIX'): - safe_remove(TESTFN) - s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - s.bind(TESTFN) - s.listen(1) - socks.append(s) - kind = 'all' - # TODO: UNIX sockets are temporarily implemented by parsing - # 'pfiles' cmd output; we don't want that part of the code to - # be executed. - if SUNOS: - kind = 'inet' - try: - self.execute('connections', kind=kind) - finally: - for s in socks: - s.close() - - -p = get_test_subprocess() -DEAD_PROC = psutil.Process(p.pid) -DEAD_PROC.kill() -DEAD_PROC.wait() -del p - - -class TestProcessObjectLeaksZombie(TestProcessObjectLeaks): - """Same as above but looks for leaks occurring when dealing with - zombie processes raising NoSuchProcess exception. - """ - proc = DEAD_PROC - - def call(self, *args, **kwargs): - try: - TestProcessObjectLeaks.call(self, *args, **kwargs) - except psutil.NoSuchProcess: - pass - - if not POSIX: - def test_kill(self): - self.execute('kill') - - def test_terminate(self): - self.execute('terminate') - - def test_suspend(self): - self.execute('suspend') - - def test_resume(self): - self.execute('resume') - - def test_wait(self): - self.execute('wait') - - -class TestModuleFunctionsLeaks(Base): - """Test leaks of psutil module functions.""" - - def setUp(self): - gc.collect() - - def call(self, function, *args, **kwargs): - fun = getattr(psutil, function) - fun(*args, **kwargs) - - @skip_if_linux() - def test_cpu_count_logical(self): - psutil.cpu_count = psutil._psplatform.cpu_count_logical - self.execute('cpu_count') - - @skip_if_linux() - def test_cpu_count_physical(self): - psutil.cpu_count = psutil._psplatform.cpu_count_physical - self.execute('cpu_count') - - @skip_if_linux() - def test_boot_time(self): - self.execute('boot_time') - - @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, - "not worth being tested on POSIX (pure python)") - def test_pid_exists(self): - self.execute('pid_exists', os.getpid()) - - def test_virtual_memory(self): - self.execute('virtual_memory') - - # TODO: remove this skip when this gets fixed - @unittest.skipIf(SUNOS, - "not worth being tested on SUNOS (uses a subprocess)") - def test_swap_memory(self): - self.execute('swap_memory') - - @skip_if_linux() - def test_cpu_times(self): - self.execute('cpu_times') - - @skip_if_linux() - def test_per_cpu_times(self): - self.execute('cpu_times', percpu=True) - - @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, - "not worth being tested on POSIX (pure python)") - def test_disk_usage(self): - self.execute('disk_usage', '.') - - def test_disk_partitions(self): - self.execute('disk_partitions') - - @skip_if_linux() - def test_net_io_counters(self): - self.execute('net_io_counters') - - @unittest.skipIf(LINUX and not os.path.exists('/proc/diskstats'), - '/proc/diskstats not available on this Linux version') - @skip_if_linux() - def test_disk_io_counters(self): - self.execute('disk_io_counters') - - # XXX - on Windows this produces a false positive - @unittest.skipIf(WINDOWS, "XXX produces a false positive on Windows") - def test_users(self): - self.execute('users') - - @unittest.skipIf(LINUX, - "not worth being tested on Linux (pure python)") - def test_net_connections(self): - self.execute('net_connections') - - def test_net_if_addrs(self): - self.execute('net_if_addrs') - - @unittest.skipIf(TRAVIS, "EPERM on travis") - def test_net_if_stats(self): - self.execute('net_if_stats') - - -def main(): - test_suite = unittest.TestSuite() - tests = [TestProcessObjectLeaksZombie, - TestProcessObjectLeaks, - TestModuleFunctionsLeaks] - for test in tests: - test_suite.addTest(unittest.makeSuite(test)) - result = unittest.TextTestRunner(verbosity=2).run(test_suite) - return result.wasSuccessful() - -if __name__ == '__main__': - if not main(): - sys.exit(1) |