refactor BSD tests (use a single file for all BSD variants)

3 files changed, 87 insertions, 153 deletions

diff --git a/test/_freebsd.py b/test/_bsd.py
index 180e300b..73d2e9a1 100644
--- a/test/_freebsd.py
+++ b/test/_bsd.py
@@ -4,10 +4,13 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
-# TODO: add test for comparing connections with 'sockstat' cmd
+# TODO: (FreeBSD) add test for comparing connections with 'sockstat' cmd.
 
-"""FreeBSD specific tests.  These are implicitly run by test_psutil.py."""
+"""Tests specific to all BSD platforms.  These are implicitly run by
+test_psutil.
+py."""
 
+import datetime
 import os
 import subprocess
 import sys
@@ -15,9 +18,11 @@ import time
 
 import psutil
 from psutil._compat import PY3
+from test_psutil import BSD
 from test_psutil import FREEBSD
 from test_psutil import get_test_subprocess
 from test_psutil import MEMORY_TOLERANCE
+from test_psutil import OPENBSD
 from test_psutil import reap_children
 from test_psutil import retry_before_failing
 from test_psutil import sh
@@ -37,7 +42,10 @@ def sysctl(cmdline):
     returning only the value of interest.
     """
     result = sh("sysctl " + cmdline)
-    result = result[result.find(": ") + 2:]
+    if FREEBSD:
+        result = result[result.find(": ") + 2:]
+    elif OPENBSD:
+        result = result[result.find("=") + 1:]
     try:
         return int(result)
     except ValueError:
@@ -55,8 +63,14 @@ def muse(field):
     return int(line.split()[1])
 
 
-@unittest.skipUnless(FREEBSD, "not a FreeBSD system")
-class FreeBSDSpecificTestCase(unittest.TestCase):
+# =====================================================================
+# --- All BSD*
+# =====================================================================
+
+
+@unittest.skipUnless(BSD, "not a BSD system")
+class BSDSpecificTestCase(unittest.TestCase):
+    """Generic tests common to all BSD variants."""
 
     @classmethod
     def setUpClass(cls):
@@ -66,13 +80,6 @@ class FreeBSDSpecificTestCase(unittest.TestCase):
     def tearDownClass(cls):
         reap_children()
 
-    def test_boot_time(self):
-        s = sysctl('sysctl kern.boottime')
-        s = s[s.find(" sec = ") + 7:]
-        s = s[:s.find(',')]
-        btime = int(s)
-        self.assertEqual(btime, psutil.boot_time())
-
     def test_process_create_time(self):
         cmdline = "ps -o lstart -p %s" % self.pid
         p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE)
@@ -112,6 +119,38 @@ class FreeBSDSpecificTestCase(unittest.TestCase):
             if abs(usage.used - used) > 10 * 1024 * 1024:
                 self.fail("psutil=%s, df=%s" % (usage.used, used))
 
+    def test_cpu_count_logical(self):
+        syst = sysctl("hw.ncpu")
+        self.assertEqual(psutil.cpu_count(logical=True), syst)
+
+    def test_virtual_memory_total(self):
+        num = sysctl('hw.physmem')
+        self.assertEqual(num, psutil.virtual_memory().total)
+
+
+# =====================================================================
+# --- FreeBSD
+# =====================================================================
+
+
+@unittest.skipUnless(FREEBSD, "not a FreeBSD system")
+class FreeBSDSpecificTestCase(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.pid = get_test_subprocess().pid
+
+    @classmethod
+    def tearDownClass(cls):
+        reap_children()
+
+    def test_boot_time(self):
+        s = sysctl('sysctl kern.boottime')
+        s = s[s.find(" sec = ") + 7:]
+        s = s[:s.find(',')]
+        btime = int(s)
+        self.assertEqual(btime, psutil.boot_time())
+
     @retry_before_failing()
     def test_memory_maps(self):
         out = sh('procstat -v %s' % self.pid)
@@ -152,14 +191,6 @@ class FreeBSDSpecificTestCase(unittest.TestCase):
 
     # --- virtual_memory(); tests against sysctl
 
-    def test_vmem_total(self):
-        syst = sysctl("sysctl vm.stats.vm.v_page_count") * PAGESIZE
-        self.assertEqual(psutil.virtual_memory().total, syst)
-
-    def test_vmem_total_2(self):
-        num = sysctl('hw.physmem')
-        self.assertEqual(num, psutil.virtual_memory().total)
-
     @retry_before_failing()
     def test_vmem_active(self):
         syst = sysctl("vm.stats.vm.v_active_count") * PAGESIZE
@@ -196,63 +227,78 @@ class FreeBSDSpecificTestCase(unittest.TestCase):
         self.assertAlmostEqual(psutil.virtual_memory().buffers, syst,
                                delta=MEMORY_TOLERANCE)
 
-    def test_cpu_count_logical(self):
-        syst = sysctl("hw.ncpu")
-        self.assertEqual(psutil.cpu_count(logical=True), syst)
-
     # --- virtual_memory(); tests against muse
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
-    def test_total(self):
+    def test_muse_vmem_total(self):
         num = muse('Total')
         self.assertEqual(psutil.virtual_memory().total, num)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_active(self):
+    def test_muse_vmem_active(self):
         num = muse('Active')
         self.assertAlmostEqual(psutil.virtual_memory().active, num,
                                delta=MEMORY_TOLERANCE)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_inactive(self):
+    def test_muse_vmem_inactive(self):
         num = muse('Inactive')
         self.assertAlmostEqual(psutil.virtual_memory().inactive, num,
                                delta=MEMORY_TOLERANCE)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_wired(self):
+    def test_muse_vmem_wired(self):
         num = muse('Wired')
         self.assertAlmostEqual(psutil.virtual_memory().wired, num,
                                delta=MEMORY_TOLERANCE)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_cached(self):
+    def test_muse_vmem_cached(self):
         num = muse('Cache')
         self.assertAlmostEqual(psutil.virtual_memory().cached, num,
                                delta=MEMORY_TOLERANCE)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_free(self):
+    def test_muse_vmem_free(self):
         num = muse('Free')
         self.assertAlmostEqual(psutil.virtual_memory().free, num,
                                delta=MEMORY_TOLERANCE)
 
     @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
     @retry_before_failing()
-    def test_buffers(self):
+    def test_muse_vmem_buffers(self):
         num = muse('Buffer')
         self.assertAlmostEqual(psutil.virtual_memory().buffers, num,
                                delta=MEMORY_TOLERANCE)
 
 
+# =====================================================================
+# --- OpenBSD
+# =====================================================================
+
+
+@unittest.skipUnless(OPENBSD, "not an OpenBSD system")
+class OpenBSDSpecificTestCase(unittest.TestCase):
+
+    def test_boot_time(self):
+        s = sysctl('kern.boottime')
+        sys_bt = datetime.datetime.strptime(s, "%a %b %d %H:%M:%S %Y")
+        psutil_bt = datetime.datetime.fromtimestamp(psutil.boot_time())
+        self.assertEqual(sys_bt, psutil_bt)
+
+
 def main():
     test_suite = unittest.TestSuite()
-    test_suite.addTest(unittest.makeSuite(FreeBSDSpecificTestCase))
+    test_suite.addTest(unittest.makeSuite(BSDSpecificTestCase))
+    if FREEBSD:
+        test_suite.addTest(unittest.makeSuite(FreeBSDSpecificTestCase))
+    elif OPENBSD:
+        test_suite.addTest(unittest.makeSuite(OpenBSDSpecificTestCase))
     result = unittest.TextTestRunner(verbosity=2).run(test_suite)
     return result.wasSuccessful()
 
diff --git a/test/_openbsd.py b/test/_openbsd.py
deleted file mode 100644
index 7949ba34..00000000
--- a/test/_openbsd.py
+++ /dev/null
@@ -1,117 +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.
-
-"""OpenBSD specific tests.  These are implicitly run by test_psutil.py."""
-
-import datetime
-import subprocess
-import sys
-import time
-
-import psutil
-from psutil._compat import PY3
-from test_psutil import get_test_subprocess
-from test_psutil import OPENBSD
-from test_psutil import reap_children
-from test_psutil import sh
-from test_psutil import unittest
-
-
-def sysctl(cmdline):
-    """Expects a sysctl command with an argument and parse the result
-    returning only the value of interest.
-    """
-    result = sh("sysctl " + cmdline)
-    result = result[result.find("=") + 1:]
-    try:
-        return int(result)
-    except ValueError:
-        return result
-
-
-@unittest.skipUnless(OPENBSD, "not an OpenBSD system")
-class OpenBSDSpecificTestCase(unittest.TestCase):
-
-    @classmethod
-    def setUpClass(cls):
-        cls.pid = get_test_subprocess().pid
-
-    @classmethod
-    def tearDownClass(cls):
-        reap_children()
-
-    def test_boot_time(self):
-        s = sysctl('kern.boottime')
-        sys_bt = datetime.datetime.strptime(s, "%a %b %d %H:%M:%S %Y")
-        psutil_bt = datetime.datetime.fromtimestamp(psutil.boot_time())
-        self.assertEqual(sys_bt, psutil_bt)
-
-    def test_process_create_time(self):
-        cmdline = "ps -o lstart -p %s" % self.pid
-        p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE)
-        output = p.communicate()[0]
-        if PY3:
-            output = str(output, sys.stdout.encoding)
-        start_ps = output.replace('STARTED', '').strip()
-        start_psutil = psutil.Process(self.pid).create_time()
-        start_psutil = time.strftime("%a %b %e %H:%M:%S %Y",
-                                     time.localtime(start_psutil))
-        self.assertEqual(start_ps, start_psutil)
-
-    def test_disks(self):
-        # test psutil.disk_usage() and psutil.disk_partitions()
-        # against "df -a"
-        def df(path):
-            out = sh('df -k "%s"' % path).strip()
-            lines = out.split('\n')
-            lines.pop(0)
-            line = lines.pop(0)
-            dev, total, used, free = line.split()[:4]
-            if dev == 'none':
-                dev = ''
-            total = int(total) * 1024
-            used = int(used) * 1024
-            free = int(free) * 1024
-            return dev, total, used, free
-
-        for part in psutil.disk_partitions(all=False):
-            usage = psutil.disk_usage(part.mountpoint)
-            dev, total, used, free = df(part.mountpoint)
-            self.assertEqual(part.device, dev)
-            self.assertEqual(usage.total, total)
-            # 10 MB tollerance
-            if abs(usage.free - free) > 10 * 1024 * 1024:
-                self.fail("psutil=%s, df=%s" % (usage.free, free))
-            if abs(usage.used - used) > 10 * 1024 * 1024:
-                self.fail("psutil=%s, df=%s" % (usage.used, used))
-
-    def test_cpu_count_logical(self):
-        syst = sysctl("hw.ncpu")
-        self.assertEqual(psutil.cpu_count(logical=True), syst)
-
-    def test_swap_memory(self):
-        out = sh("pstat -s")
-        _, total, used, free, _, _ = out.split('\n')[1].split()
-        smem = psutil.swap_memory()
-        self.assertEqual(smem.total, int(total) * 512)
-        self.assertEqual(smem.used, int(used) * 512)
-        self.assertEqual(smem.free, int(free) * 512)
-
-    def test_virtual_memory_total(self):
-        num = sysctl('hw.physmem')
-        self.assertEqual(num, psutil.virtual_memory().total)
-
-
-def main():
-    test_suite = unittest.TestSuite()
-    test_suite.addTest(unittest.makeSuite(OpenBSDSpecificTestCase))
-    result = unittest.TextTestRunner(verbosity=2).run(test_suite)
-    return result.wasSuccessful()
-
-
-if __name__ == '__main__':
-    if not main():
-        sys.exit(1)
diff --git a/test/test_psutil.py b/test/test_psutil.py
index d8c910ab..5494e27b 100644
--- a/test/test_psutil.py
+++ b/test/test_psutil.py
@@ -3237,12 +3237,17 @@ def main():
         tests.append(TestDualProcessImplementation)
     elif OSX:
         from _osx import OSXSpecificTestCase as stc
-    elif FREEBSD:
-        from _freebsd import FreeBSDSpecificTestCase as stc
-    elif OPENBSD:
-        from _openbsd import OpenBSDSpecificTestCase as stc
     elif SUNOS:
         from _sunos import SunOSSpecificTestCase as stc
+    elif BSD:
+        from _bsd import BSDSpecificTestCase as stc
+        if FREEBSD:
+            from _bsd import FreeBSDSpecificTestCase
+            tests.append(FreeBSDSpecificTestCase)
+        elif OPENBSD:
+            from _bsd import OpenBSDSpecificTestCase
+            tests.append(OpenBSDSpecificTestCase)
+
     if stc is not None:
         tests.append(stc)