def test_vmem_used(self):
     lines = sh('free').split('\n')[1:]
     total = int(lines[0].split()[1])
     free = int(lines[0].split()[3])
     used = (total - free) * 1024
     self.assertAlmostEqual(used, psutil.virtual_memory().used,
                            delta=MEMORY_TOLERANCE)

 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_swapmem_total(self):
     out = sh('sysctl vm.swapusage')
     out = out.replace('vm.swapusage: ', '')
     total, used, free = re.findall('\d+.\d+\w', out)
     psutil_smem = psutil.swap_memory()
     self.assertEqual(psutil_smem.total, human2bytes(total))
     self.assertEqual(psutil_smem.used, human2bytes(used))
     self.assertEqual(psutil_smem.free, human2bytes(free))

 def test_users(self):
     out = sh("who")
     lines = out.split('\n')
     users = [x.split()[0] for x in lines]
     self.assertEqual(len(users), len(psutil.users()))
     terminals = [x.split()[1] for x in lines]
     for u in psutil.users():
         self.assertTrue(u.name in users, u.name)
         self.assertTrue(u.terminal in terminals, u.terminal)

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

def vm_stat(field):
    """Wrapper around 'vm_stat' cmdline utility."""
    out = sh('vm_stat')
    for line in out.split('\n'):
        if field in line:
            break
    else:
        raise ValueError("line not found")
    return int(re.search('\d+', line).group(0)) * PAGESIZE

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

 def df(path):
     out = sh('df -P -B 1 "%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, used, free = int(total), int(used), int(free)
     return dev, total, used, free

 def test_net_if_names(self):
     out = sh("ip addr").strip()
     nics = psutil.net_if_addrs()
     found = 0
     for line in out.split('\n'):
         line = line.strip()
         if re.search("^\d+:", line):
             found += 1
             name = line.split(':')[1].strip()
             self.assertIn(name, nics.keys())
     self.assertEqual(len(nics), found, msg="%s\n---\n%s" % (
         pprint.pformat(nics), out))

 def test_uids_gids(self):
     out = sh('procstat -s %s' % self.pid)
     euid, ruid, suid, egid, rgid, sgid = out.split('\n')[1].split()[2:8]
     p = psutil.Process(self.pid)
     uids = p.uids()
     gids = p.gids()
     self.assertEqual(uids.real, int(ruid))
     self.assertEqual(uids.effective, int(euid))
     self.assertEqual(uids.saved, int(suid))
     self.assertEqual(gids.real, int(rgid))
     self.assertEqual(gids.effective, int(egid))
     self.assertEqual(gids.saved, int(sgid))

 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

 def test_memory_maps(self):
     out = sh('procstat -v %s' % self.pid)
     maps = psutil.Process(self.pid).get_memory_maps(grouped=False)
     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)

 def test_memory_maps(self):
     sproc = get_test_subprocess()
     time.sleep(1)
     p = psutil.Process(sproc.pid)
     maps = p.get_memory_maps(grouped=False)
     pmap = sh('pmap -x %s' % p.pid).split('\n')
     del pmap[0]; del pmap[0]  # get rid of header
     while maps and pmap:
         this = maps.pop(0)
         other = pmap.pop(0)
         addr, _, rss, dirty, mode, path = other.split(None, 5)
         if not path.startswith('[') and not path.endswith(']'):
             self.assertEqual(path, os.path.basename(this.path))
         self.assertEqual(int(rss) * 1024, this.rss)
         # test only rwx chars, ignore 's' and 'p'
         self.assertEqual(mode[:3], this.perms[:3])

    def test_swap_memory(self):
        out = sh('env PATH=/usr/sbin:/sbin:%s swap -l' % os.environ['PATH'])
        lines = out.strip().split('\n')[1:]
        if not lines:
            raise ValueError('no swap device(s) configured')
        total = free = 0
        for line in lines:
            line = line.split()
            t, f = line[-2:]
            total += int(int(t) * 512)
            free += int(int(f) * 512)
        used = total - free

        psutil_swap = psutil.swap_memory()
        self.assertEqual(psutil_swap.total, total)
        self.assertEqual(psutil_swap.used, used)
        self.assertEqual(psutil_swap.free, free)

    def test_swap_memory(self):
        out = sh('swap -l -k')
        lines = out.strip().split('\n')[1:]
        if not lines:
            raise ValueError('no swap device(s) configured')
        total = free = 0
        for line in lines:
            line = line.split()
            t, f = line[-2:]
            t = t.replace('K', '')
            f = f.replace('K', '')
            total += int(int(t) * 1024)
            free += int(int(f) * 1024)
        used = total - free

        psutil_swap = psutil.swap_memory()
        self.assertEqual(psutil_swap.total, total)
        self.assertEqual(psutil_swap.used, used)
        self.assertEqual(psutil_swap.free, free)

 def test_vmem_free(self):
     lines = sh('free').split('\n')[1:]
     free = int(lines[0].split()[3]) * 1024
     self.assertAlmostEqual(free, psutil.virtual_memory().free,
                            delta=MEMORY_TOLERANCE)

 def test_cpu_count_logical_w_lscpu(self):
     out = sh("lscpu -p")
     num = len([x for x in out.split('\n') if not x.startswith('#')])
     self.assertEqual(psutil.cpu_count(logical=True), num)

 def test_vmem_cached(self):
     cached = int(sh('vmstat').split('\n')[2].split()[5]) * 1024
     self.assertAlmostEqual(cached, psutil.virtual_memory().cached,
                            delta=MEMORY_TOLERANCE)

 def test_vmem_buffers(self):
     buffers = int(sh('vmstat').split('\n')[2].split()[4]) * 1024
     self.assertAlmostEqual(buffers, psutil.virtual_memory().buffers,
                            delta=MEMORY_TOLERANCE)

 def test_swapmem_free(self):
     lines = sh('free').split('\n')[1:]
     free = int(lines[2 if OLD_PROCPS_NG_VERSION else 1].split()[1]) * 1024
     self.assertAlmostEqual(free, psutil.swap_memory().free,
                            delta=MEMORY_TOLERANCE)