Merge branch 'master' of github.com:giampaolo/psutil

MANIFEST.in

@@ -103,6 +103,7 @@ include psutil/tests/test_posix.py
 include psutil/tests/test_process.py
 include psutil/tests/test_sunos.py
 include psutil/tests/test_system.py
+include psutil/tests/test_testutils.py
 include psutil/tests/test_unicode.py
 include psutil/tests/test_windows.py
 include scripts/battery.py

Makefile

@@ -124,6 +124,10 @@ test-misc:  ## Run miscellaneous tests.
 	${MAKE} install
 	$(TEST_PREFIX) $(PYTHON) psutil/tests/test_misc.py
 
+test-testutils:  ## Run test utils tests.
+	${MAKE} install
+	$(TEST_PREFIX) $(PYTHON) psutil/tests/test_testutils.py
+
 test-unicode:  ## Test APIs dealing with strings.
 	${MAKE} install
 	$(TEST_PREFIX) $(PYTHON) psutil/tests/test_unicode.py

psutil/__init__.py

@@ -226,7 +226,7 @@
 
 __all__.extend(_psplatform.__extra__all__)
 __author__ = "Giampaolo Rodola'"
-__version__ = "5.7.0"
+__version__ = "5.7.1"
 version_info = tuple([int(num) for num in __version__.split('.')])
 
 _timer = getattr(time, 'monotonic', time.time)

psutil/_common.py

@@ -785,7 +785,7 @@ def hilite(s, color="green", bold=False):
     if not term_supports_colors():
         return s
     attr = []
-    colors = dict(green='32', red='91', brown='33')
+    colors = dict(green='32', red='91', brown='33', yellow='93')
     colors[None] = '29'
     try:
         color = colors[color]
@@ -812,7 +812,7 @@ def print_color(s, color="green", bold=False, file=sys.stdout):
         SetConsoleTextAttribute = \
             ctypes.windll.Kernel32.SetConsoleTextAttribute
 
-        colors = dict(green=2, red=4, brown=6)
+        colors = dict(green=2, red=4, brown=6, yellow=6)
         colors[None] = DEFAULT_COLOR
         try:
             color = colors[color]

psutil/_compat.py

@@ -5,13 +5,14 @@
 """Module which provides compatibility with older Python versions."""
 
 import collections
+import contextlib
 import errno
 import functools
 import os
 import sys
 
 __all__ = ["PY3", "long", "xrange", "unicode", "basestring", "u", "b",
-           "lru_cache", "which", "get_terminal_size",
+           "lru_cache", "which", "get_terminal_size", "redirect_stderr",
            "FileNotFoundError", "PermissionError", "ProcessLookupError",
            "InterruptedError", "ChildProcessError", "FileExistsError"]
 
@@ -343,3 +344,17 @@ def get_terminal_size(fallback=(80, 24)):
                 return (res[1], res[0])
             except Exception:
                 return fallback
+
+
+# python 3.4
+try:
+    from contextlib import redirect_stderr
+except ImportError:
+    @contextlib.contextmanager
+    def redirect_stderr(target):
+        original = sys.stderr
+        try:
+            sys.stderr = target
+            yield
+        finally:
+            sys.stderr = original

psutil/_pslinux.py

@@ -361,7 +361,6 @@ def calculate_avail_vmem(mems):
             if line.startswith(b'low'):
                 watermark_low += int(line.split()[1])
     watermark_low *= PAGESIZE
-    watermark_low = watermark_low
 
     avail = free - watermark_low
     pagecache = lru_active_file + lru_inactive_file

psutil/tests/__init__.py

@@ -9,12 +9,12 @@
 """
 
 from __future__ import print_function
-
 import atexit
 import contextlib
 import ctypes
 import errno
 import functools
+import gc
 import os
 import random
 import re
@@ -40,14 +40,17 @@
 from psutil import POSIX
 from psutil import SUNOS
 from psutil import WINDOWS
+from psutil._common import bytes2human
 from psutil._common import supports_ipv6
+from psutil._common import print_color
 from psutil._compat import ChildProcessError
 from psutil._compat import FileExistsError
 from psutil._compat import FileNotFoundError
 from psutil._compat import PY3
 from psutil._compat import u
 from psutil._compat import unicode
 from psutil._compat import which
+from psutil._compat import xrange
 
 if sys.version_info < (2, 7):
     import unittest2 as unittest  # requires "pip install unittest2"
@@ -84,7 +87,7 @@
     'ThreadTask'
     # test utils
     'unittest', 'skip_on_access_denied', 'skip_on_not_implemented',
-    'retry_on_failure',
+    'retry_on_failure', 'TestMemoryLeak',
     # install utils
     'install_pip', 'install_test_deps',
     # fs utils
@@ -814,6 +817,120 @@ def __str__(self):
 unittest.TestCase = TestCase
 
 
+@unittest.skipIf(PYPY, "unreliable on PYPY")
+class TestMemoryLeak(unittest.TestCase):
+    """Test framework class for detecting function memory leaks (typically
+    functions implemented in C).
+    It does so by calling a function many times, and checks whether the
+    process memory usage increased before and after having called the
+    function repeadetly.
+    Note that sometimes this may produce false positives.
+    PyPy appears to be completely unstable for this framework, probably
+    because of how its JIT handles memory, so tests on PYPY are
+    automatically skipped.
+    """
+    # Configurable class attrs.
+    times = 1200
+    warmup_times = 10
+    tolerance = 4096  # memory
+    retry_for = 3.0  # seconds
+    verbose = True
+
+    def setUp(self):
+        self._thisproc = psutil.Process()
+        gc.collect()
+
+    def _get_mem(self):
+        # USS is the closest thing we have to "real" memory usage and it
+        # should be less likely to produce false positives.
+        mem = self._thisproc.memory_full_info()
+        return getattr(mem, "uss", mem.rss)
+
+    def _call(self, fun):
+        return fun()
+
+    def _itercall(self, fun, iterator):
+        """Get 2 distinct memory samples, before and after having
+        called fun repeadetly, and return the memory difference.
+        """
+        ncalls = 0
+        gc.collect()
+        mem1 = self._get_mem()
+        for x in iterator:
+            ret = self._call(fun)
+            ncalls += 1
+            del x, ret
+        gc.collect()
+        mem2 = self._get_mem()
+        self.assertEqual(gc.garbage, [])
+        diff = mem2 - mem1
+        if diff < 0:
+            self._log("negative memory diff -%s" % (bytes2human(abs(diff))))
+        return (diff, ncalls)
+
+    def _call_ntimes(self, fun, times):
+        return self._itercall(fun, xrange(times))[0]
+
+    def _call_for(self, fun, secs):
+        def iterator(secs):
+            stop_at = time.time() + secs
+            while time.time() < stop_at:
+                yield
+        return self._itercall(fun, iterator(secs))
+
+    def _log(self, msg):
+        if self.verbose:
+            print_color(msg, color="yellow", file=sys.stderr)
+
+    def execute(self, fun, times=times, warmup_times=warmup_times,
+                tolerance=tolerance, retry_for=retry_for):
+        """Test a callable."""
+        if times <= 0:
+            raise ValueError("times must be > 0")
+        if warmup_times < 0:
+            raise ValueError("warmup_times must be >= 0")
+        if tolerance is not None and tolerance < 0:
+            raise ValueError("tolerance must be >= 0")
+        if retry_for is not None and retry_for < 0:
+            raise ValueError("retry_for must be >= 0")
+
+        # warm up
+        self._call_ntimes(fun, warmup_times)
+        mem1 = self._call_ntimes(fun, times)
+
+        if mem1 > 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
+            # anymore. Let's keep calling fun for N more seconds and
+            # fail if we notice any difference (ignore tolerance).
+            msg = "+%s after %s calls; try calling fun for another %s secs" % (
+                bytes2human(mem1), times, retry_for)
+            if not retry_for:
+                raise self.fail(msg)
+            else:
+                self._log(msg)
+
+            mem2, ncalls = self._call_for(fun, retry_for)
+            if mem2 > mem1:
+                # failure
+                msg = "+%s memory increase after %s calls; " % (
+                    bytes2human(mem1), times)
+                msg += "+%s after another %s calls over %s secs" % (
+                    bytes2human(mem2), ncalls, retry_for)
+                raise self.fail(msg)
+
+    def execute_w_exc(self, exc, fun, **kwargs):
+        """Convenience method to test a callable while making sure it
+        raises an exception on every call.
+        """
+        def call():
+            self.assertRaises(exc, fun)
+
+        self.execute(call, **kwargs)
+
+
 def retry_on_failure(retries=NO_RETRIES):
     """Decorator which runs a test function and retries N times before
     actually failing.