author Christian Holler <>
Fri, 20 Sep 2013 13:37:53 +0200
changeset 148229 67fa64cde217038d28342a5af024129f9cc2bc28
parent 144129 037e6a08032fd65347687619ec6de9a02fe00f21
child 221195 697eb6db7d96dc21e817cd27a7e46ed4ab00f9bb
permissions -rw-r--r--
Bug 917242 - Symbolize ASan traces in automation. r=ted

#!/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.

"""BSD specific tests.  These are implicitly run by"""

import unittest
import subprocess
import time
import re
import sys
import os

import psutil

from psutil._compat import PY3
from test_psutil import *

PAGESIZE = os.sysconf("SC_PAGE_SIZE")
MUSE_AVAILABLE = which('muse')

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(": ") + 2:]
        return int(result)
    except ValueError:
        return result

def muse(field):
    """Thin wrapper around 'muse' cmdline utility."""
    out = sh('muse', stderr=DEVNULL)
    for line in out.split('\n'):
        if line.startswith(field):
        raise ValueError("line not found")
    return int(line.split()[1])

class BSDSpecificTestCase(unittest.TestCase):

    def setUp(self): = get_test_subprocess().pid

    def tearDown(self):

    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"
        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(
        start_psutil = time.strftime("%a %b %e %H:%M:%S %Y",
        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')
            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(, total)
            # 10 MB tollerance
            if abs( - free) > 10 * 1024 * 1024:
      "psutil=%s, df=%s" % (, free))
            if abs(usage.used - used) > 10 * 1024 * 1024:
      "psutil=%s, df=%s" % (usage.used, used))

    def test_memory_maps(self):
        out = sh('procstat -v %s' %
        maps = psutil.Process(
        lines = out.split('\n')[1:]
        while lines:
            line = lines.pop()
            fields = line.split()
            _, start, stop, perms, res = fields[:5]
            map = maps.pop()
            self.assertEqual("%s-%s" % (start, stop), map.addr)
            self.assertEqual(int(res), map.rss)
            if not map.path.startswith('['):
                self.assertEqual(fields[10], map.path)

    # --- 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_active(self):
        syst = sysctl("vm.stats.vm.v_active_count") * PAGESIZE
        self.assertAlmostEqual(psutil.virtual_memory().active, syst,

    def test_vmem_inactive(self):
        syst = sysctl("vm.stats.vm.v_inactive_count") * PAGESIZE
        self.assertAlmostEqual(psutil.virtual_memory().inactive, syst,

    def test_vmem_wired(self):
        syst = sysctl("vm.stats.vm.v_wire_count") * PAGESIZE
        self.assertAlmostEqual(psutil.virtual_memory().wired, syst,

    def test_vmem_cached(self):
        syst = sysctl("vm.stats.vm.v_cache_count") * PAGESIZE
        self.assertAlmostEqual(psutil.virtual_memory().cached, syst,

    def test_vmem_free(self):
        syst = sysctl("vm.stats.vm.v_free_count") * PAGESIZE
        self.assertAlmostEqual(psutil.virtual_memory().free, syst,

    def test_vmem_buffers(self):
        syst = sysctl("vfs.bufspace")
        self.assertAlmostEqual(psutil.virtual_memory().buffers, syst,

    # --- virtual_memory(); tests against muse

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_total(self):
        num = muse('Total')
        self.assertEqual(psutil.virtual_memory().total, num)

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_active(self):
        num = muse('Active')
        self.assertAlmostEqual(psutil.virtual_memory().active, num,

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_inactive(self):
        num = muse('Inactive')
        self.assertAlmostEqual(psutil.virtual_memory().inactive, num,

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_wired(self):
        num = muse('Wired')
        self.assertAlmostEqual(psutil.virtual_memory().wired, num,

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_cached(self):
        num = muse('Cache')
        self.assertAlmostEqual(psutil.virtual_memory().cached, num,

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_free(self):
        num = muse('Free')
        self.assertAlmostEqual(psutil.virtual_memory().free, num,

    @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available")
    def test_buffers(self):
        num = muse('Buffer')
        self.assertAlmostEqual(psutil.virtual_memory().buffers, num,

def test_main():
    test_suite = unittest.TestSuite()
    result = unittest.TextTestRunner(verbosity=2).run(test_suite)
    return result.wasSuccessful()

if __name__ == '__main__':
    if not test_main():