Edit File: test_testutils.py
#!/usr/bin/env python3 # -*- coding: utf-8 -*- # 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. """ Tests for testing utils (psutil.tests namespace). """ import collections import contextlib import errno import os import socket import stat import subprocess from psutil import FREEBSD from psutil import NETBSD from psutil import POSIX from psutil._common import open_binary from psutil._common import open_text from psutil._common import supports_ipv6 from psutil.tests import bind_socket from psutil.tests import bind_unix_socket from psutil.tests import call_until from psutil.tests import chdir from psutil.tests import CI_TESTING from psutil.tests import create_sockets from psutil.tests import get_free_port from psutil.tests import HAS_CONNECTIONS_UNIX from psutil.tests import is_namedtuple from psutil.tests import mock from psutil.tests import process_namespace from psutil.tests import PsutilTestCase from psutil.tests import PYTHON_EXE from psutil.tests import reap_children from psutil.tests import retry from psutil.tests import retry_on_failure from psutil.tests import safe_mkdir from psutil.tests import safe_rmpath from psutil.tests import serialrun from psutil.tests import system_namespace from psutil.tests import tcp_socketpair from psutil.tests import terminate from psutil.tests import TestMemoryLeak from psutil.tests import unittest from psutil.tests import unix_socketpair from psutil.tests import wait_for_file from psutil.tests import wait_for_pid import psutil import psutil.tests # =================================================================== # --- Unit tests for test utilities. # =================================================================== class TestRetryDecorator(PsutilTestCase): @mock.patch('time.sleep') def test_retry_success(self, sleep): # Fail 3 times out of 5; make sure the decorated fun returns. @retry(retries=5, interval=1, logfun=None) def foo(): while queue: queue.pop() 1 / 0 return 1 queue = list(range(3)) self.assertEqual(foo(), 1) self.assertEqual(sleep.call_count, 3) @mock.patch('time.sleep') def test_retry_failure(self, sleep): # Fail 6 times out of 5; th function is supposed to raise exc. @retry(retries=5, interval=1, logfun=None) def foo(): while queue: queue.pop() 1 / 0 return 1 queue = list(range(6)) self.assertRaises(ZeroDivisionError, foo) self.assertEqual(sleep.call_count, 5) @mock.patch('time.sleep') def test_exception_arg(self, sleep): @retry(exception=ValueError, interval=1) def foo(): raise TypeError self.assertRaises(TypeError, foo) self.assertEqual(sleep.call_count, 0) @mock.patch('time.sleep') def test_no_interval_arg(self, sleep): # if interval is not specified sleep is not supposed to be called @retry(retries=5, interval=None, logfun=None) def foo(): 1 / 0 self.assertRaises(ZeroDivisionError, foo) self.assertEqual(sleep.call_count, 0) @mock.patch('time.sleep') def test_retries_arg(self, sleep): @retry(retries=5, interval=1, logfun=None) def foo(): 1 / 0 self.assertRaises(ZeroDivisionError, foo) self.assertEqual(sleep.call_count, 5) @mock.patch('time.sleep') def test_retries_and_timeout_args(self, sleep): self.assertRaises(ValueError, retry, retries=5, timeout=1) class TestSyncTestUtils(PsutilTestCase): def test_wait_for_pid(self): wait_for_pid(os.getpid()) nopid = max(psutil.pids()) + 99999 with mock.patch('psutil.tests.retry.__iter__', return_value=iter([0])): self.assertRaises(psutil.NoSuchProcess, wait_for_pid, nopid) def test_wait_for_file(self): testfn = self.get_testfn() with open(testfn, 'w') as f: f.write('foo') wait_for_file(testfn) assert not os.path.exists(testfn) def test_wait_for_file_empty(self): testfn = self.get_testfn() with open(testfn, 'w'): pass wait_for_file(testfn, empty=True) assert not os.path.exists(testfn) def test_wait_for_file_no_file(self): testfn = self.get_testfn() with mock.patch('psutil.tests.retry.__iter__', return_value=iter([0])): self.assertRaises(IOError, wait_for_file, testfn) def test_wait_for_file_no_delete(self): testfn = self.get_testfn() with open(testfn, 'w') as f: f.write('foo') wait_for_file(testfn, delete=False) assert os.path.exists(testfn) def test_call_until(self): ret = call_until(lambda: 1, "ret == 1") self.assertEqual(ret, 1) class TestFSTestUtils(PsutilTestCase): def test_open_text(self): with open_text(__file__) as f: self.assertEqual(f.mode, 'rt') def test_open_binary(self): with open_binary(__file__) as f: self.assertEqual(f.mode, 'rb') def test_safe_mkdir(self): testfn = self.get_testfn() safe_mkdir(testfn) assert os.path.isdir(testfn) safe_mkdir(testfn) assert os.path.isdir(testfn) def test_safe_rmpath(self): # test file is removed testfn = self.get_testfn() open(testfn, 'w').close() safe_rmpath(testfn) assert not os.path.exists(testfn) # test no exception if path does not exist safe_rmpath(testfn) # test dir is removed os.mkdir(testfn) safe_rmpath(testfn) assert not os.path.exists(testfn) # test other exceptions are raised with mock.patch('psutil.tests.os.stat', side_effect=OSError(errno.EINVAL, "")) as m: with self.assertRaises(OSError): safe_rmpath(testfn) assert m.called def test_chdir(self): testfn = self.get_testfn() base = os.getcwd() os.mkdir(testfn) with chdir(testfn): self.assertEqual(os.getcwd(), os.path.join(base, testfn)) self.assertEqual(os.getcwd(), base) class TestProcessUtils(PsutilTestCase): def test_reap_children(self): subp = self.spawn_testproc() p = psutil.Process(subp.pid) assert p.is_running() reap_children() assert not p.is_running() assert not psutil.tests._pids_started assert not psutil.tests._subprocesses_started def test_spawn_children_pair(self): child, grandchild = self.spawn_children_pair() self.assertNotEqual(child.pid, grandchild.pid) assert child.is_running() assert grandchild.is_running() children = psutil.Process().children() self.assertEqual(children, [child]) children = psutil.Process().children(recursive=True) self.assertEqual(len(children), 2) self.assertIn(child, children) self.assertIn(grandchild, children) self.assertEqual(child.ppid(), os.getpid()) self.assertEqual(grandchild.ppid(), child.pid) terminate(child) assert not child.is_running() assert grandchild.is_running() terminate(grandchild) assert not grandchild.is_running() @unittest.skipIf(not POSIX, "POSIX only") def test_spawn_zombie(self): parent, zombie = self.spawn_zombie() self.assertEqual(zombie.status(), psutil.STATUS_ZOMBIE) def test_terminate(self): # by subprocess.Popen p = self.spawn_testproc() terminate(p) self.assertProcessGone(p) terminate(p) # by psutil.Process p = psutil.Process(self.spawn_testproc().pid) terminate(p) self.assertProcessGone(p) terminate(p) # by psutil.Popen cmd = [PYTHON_EXE, "-c", "import time; time.sleep(60);"] p = psutil.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) terminate(p) self.assertProcessGone(p) terminate(p) # by PID pid = self.spawn_testproc().pid terminate(pid) self.assertProcessGone(p) terminate(pid) # zombie if POSIX: parent, zombie = self.spawn_zombie() terminate(parent) terminate(zombie) self.assertProcessGone(parent) self.assertProcessGone(zombie) class TestNetUtils(PsutilTestCase): def bind_socket(self): port = get_free_port() with contextlib.closing(bind_socket(addr=('', port))) as s: self.assertEqual(s.getsockname()[1], port) @unittest.skipIf(not POSIX, "POSIX only") def test_bind_unix_socket(self): name = self.get_testfn() sock = bind_unix_socket(name) with contextlib.closing(sock): self.assertEqual(sock.family, socket.AF_UNIX) self.assertEqual(sock.type, socket.SOCK_STREAM) self.assertEqual(sock.getsockname(), name) assert os.path.exists(name) assert stat.S_ISSOCK(os.stat(name).st_mode) # UDP name = self.get_testfn() sock = bind_unix_socket(name, type=socket.SOCK_DGRAM) with contextlib.closing(sock): self.assertEqual(sock.type, socket.SOCK_DGRAM) def tcp_tcp_socketpair(self): addr = ("127.0.0.1", get_free_port()) server, client = tcp_socketpair(socket.AF_INET, addr=addr) with contextlib.closing(server): with contextlib.closing(client): # Ensure they are connected and the positions are # correct. self.assertEqual(server.getsockname(), addr) self.assertEqual(client.getpeername(), addr) self.assertNotEqual(client.getsockname(), addr) @unittest.skipIf(not POSIX, "POSIX only") @unittest.skipIf(NETBSD or FREEBSD, "/var/run/log UNIX socket opened by default") def test_unix_socketpair(self): p = psutil.Process() num_fds = p.num_fds() assert not p.connections(kind='unix') name = self.get_testfn() server, client = unix_socketpair(name) try: assert os.path.exists(name) assert stat.S_ISSOCK(os.stat(name).st_mode) self.assertEqual(p.num_fds() - num_fds, 2) self.assertEqual(len(p.connections(kind='unix')), 2) self.assertEqual(server.getsockname(), name) self.assertEqual(client.getpeername(), name) finally: client.close() server.close() def test_create_sockets(self): with create_sockets() as socks: fams = collections.defaultdict(int) types = collections.defaultdict(int) for s in socks: fams[s.family] += 1 # work around http://bugs.python.org/issue30204 types[s.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE)] += 1 self.assertGreaterEqual(fams[socket.AF_INET], 2) if supports_ipv6(): self.assertGreaterEqual(fams[socket.AF_INET6], 2) if POSIX and HAS_CONNECTIONS_UNIX: self.assertGreaterEqual(fams[socket.AF_UNIX], 2) self.assertGreaterEqual(types[socket.SOCK_STREAM], 2) self.assertGreaterEqual(types[socket.SOCK_DGRAM], 2) @serialrun class TestMemLeakClass(TestMemoryLeak): def test_times(self): def fun(): cnt['cnt'] += 1 cnt = {'cnt': 0} self.execute(fun, times=10, warmup_times=15) self.assertEqual(cnt['cnt'], 26) def test_param_err(self): self.assertRaises(ValueError, self.execute, lambda: 0, times=0) self.assertRaises(ValueError, self.execute, lambda: 0, times=-1) self.assertRaises(ValueError, self.execute, lambda: 0, warmup_times=-1) self.assertRaises(ValueError, self.execute, lambda: 0, tolerance=-1) self.assertRaises(ValueError, self.execute, lambda: 0, retries=-1) @retry_on_failure() @unittest.skipIf(CI_TESTING, "skipped on CI") def test_leak_mem(self): ls = [] def fun(ls=ls): ls.append("x" * 24 * 1024) try: # will consume around 3M in total self.assertRaisesRegex(AssertionError, "extra-mem", self.execute, fun, times=50) finally: del ls def test_unclosed_files(self): def fun(): f = open(__file__) self.addCleanup(f.close) box.append(f) box = [] kind = "fd" if POSIX else "handle" self.assertRaisesRegex(AssertionError, "unclosed " + kind, self.execute, fun) def test_tolerance(self): def fun(): ls.append("x" * 24 * 1024) ls = [] times = 100 self.execute(fun, times=times, warmup_times=0, tolerance=200 * 1024 * 1024) self.assertEqual(len(ls), times + 1) def test_execute_w_exc(self): def fun(): 1 / 0 self.execute_w_exc(ZeroDivisionError, fun) with self.assertRaises(ZeroDivisionError): self.execute_w_exc(OSError, fun) def fun(): pass with self.assertRaises(AssertionError): self.execute_w_exc(ZeroDivisionError, fun) class TestTestingUtils(PsutilTestCase): def test_process_namespace(self): p = psutil.Process() ns = process_namespace(p) ns.test() fun = [x for x in ns.iter(ns.getters) if x[1] == 'ppid'][0][0] self.assertEqual(fun(), p.ppid()) def test_system_namespace(self): ns = system_namespace() fun = [x for x in ns.iter(ns.getters) if x[1] == 'net_if_addrs'][0][0] self.assertEqual(fun(), psutil.net_if_addrs()) class TestOtherUtils(PsutilTestCase): def test_is_namedtuple(self): assert is_namedtuple(collections.namedtuple('foo', 'a b c')(1, 2, 3)) assert not is_namedtuple(tuple()) if __name__ == '__main__': from psutil.tests.runner import run_from_name run_from_name(__file__)
Back to File Manager