Index: third_party/psutil/test/test_memory_leaks.py |
diff --git a/third_party/psutil/test/test_memory_leaks.py b/third_party/psutil/test/test_memory_leaks.py |
deleted file mode 100644 |
index b6a3a4e7cfd074a5846ba755220e0ad0654e5974..0000000000000000000000000000000000000000 |
--- a/third_party/psutil/test/test_memory_leaks.py |
+++ /dev/null |
@@ -1,211 +0,0 @@ |
-#!/usr/bin/env python |
-# |
-# $Id: test_memory_leaks.py 1142 2011-10-05 18:45:49Z g.rodola $ |
-# |
-# Copyright (c) 2009, Jay Loden, 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 os |
-import gc |
-import unittest |
-import time |
- |
-import psutil |
-from test_psutil import reap_children, skipUnless, skipIf, \ |
- POSIX, LINUX, WINDOWS, OSX, BSD, PY3 |
- |
-LOOPS = 1000 |
-TOLERANCE = 4096 |
- |
-if PY3: |
- xrange = range |
- |
- |
-class Base(unittest.TestCase): |
- |
- def execute(self, function, *args, **kwargs): |
- # step 1 |
- for x in xrange(LOOPS): |
- self.call(function, *args, **kwargs) |
- del x |
- gc.collect() |
- rss1 = self.get_mem() |
- |
- # step 2 |
- for x in xrange(LOOPS): |
- self.call(function, *args, **kwargs) |
- del x |
- gc.collect() |
- rss2 = self.get_mem() |
- |
- # comparison |
- 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 1: |
- 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)) |
- |
- @staticmethod |
- def get_mem(): |
- return psutil.Process(os.getpid()).get_memory_info()[0] |
- |
- def call(self): |
- 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): |
- p = psutil.Process(os.getpid()) |
- obj = getattr(p, function) |
- if callable(obj): |
- obj(*args, **kwargs) |
- |
- def test_name(self): |
- self.execute('name') |
- |
- def test_cmdline(self): |
- self.execute('cmdline') |
- |
- def test_ppid(self): |
- self.execute('ppid') |
- |
- @skipIf(WINDOWS) |
- def test_uids(self): |
- self.execute('uids') |
- |
- @skipIf(WINDOWS) |
- def test_gids(self): |
- self.execute('gids') |
- |
- def test_status(self): |
- self.execute('status') |
- |
- @skipIf(POSIX) |
- def test_username(self): |
- self.execute('username') |
- |
- def test_create_time(self): |
- self.execute('create_time') |
- |
- def test_get_num_threads(self): |
- self.execute('get_num_threads') |
- |
- def test_get_threads(self): |
- self.execute('get_threads') |
- |
- def test_get_cpu_times(self): |
- self.execute('get_cpu_times') |
- |
- def test_get_memory_info(self): |
- self.execute('get_memory_info') |
- |
- def test_is_running(self): |
- self.execute('is_running') |
- |
- @skipIf(WINDOWS) |
- def test_terminal(self): |
- self.execute('terminal') |
- |
- @skipUnless(WINDOWS) |
- def test_resume(self): |
- self.execute('resume') |
- |
- @skipUnless(WINDOWS) |
- def test_getcwd(self): |
- self.execute('getcwd') |
- |
- @skipUnless(WINDOWS or OSX) |
- def test_get_open_files(self): |
- self.execute('get_open_files') |
- |
- @skipUnless(WINDOWS or OSX) |
- def test_get_connections(self): |
- self.execute('get_connections') |
- |
- |
-class TestModuleFunctionsLeaks(Base): |
- """Test leaks of psutil module functions.""" |
- |
- def setUp(self): |
- gc.collect() |
- |
- def call(self, function, *args, **kwargs): |
- obj = getattr(psutil, function) |
- if callable(obj): |
- retvalue = obj(*args, **kwargs) |
- |
- def test_get_pid_list(self): |
- self.execute('get_pid_list') |
- |
- @skipIf(POSIX) |
- def test_pid_exists(self): |
- self.execute('pid_exists', os.getpid()) |
- |
- def test_process_iter(self): |
- self.execute('process_iter') |
- |
- def test_phymem_usage(self): |
- self.execute('phymem_usage') |
- |
- def test_virtmem_usage(self): |
- self.execute('virtmem_usage') |
- |
- def test_cpu_times(self): |
- self.execute('cpu_times') |
- |
- def test_per_cpu_times(self): |
- self.execute('cpu_times', percpu=True) |
- |
- @skipUnless(WINDOWS) |
- def test_disk_usage(self): |
- self.execute('disk_usage', '.') |
- |
- def test_disk_partitions(self): |
- self.execute('disk_partitions') |
- |
- def test_network_io_counters(self): |
- self.execute('network_io_counters') |
- |
- def test_disk_io_counters(self): |
- self.execute('disk_io_counters') |
- |
-def test_main(): |
- test_suite = unittest.TestSuite() |
- test_suite.addTest(unittest.makeSuite(TestProcessObjectLeaks)) |
- test_suite.addTest(unittest.makeSuite(TestModuleFunctionsLeaks)) |
- unittest.TextTestRunner(verbosity=2).run(test_suite) |
- |
-if __name__ == '__main__': |
- test_main() |
- |