summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorGiampaolo Rodola <g.rodola@gmail.com>2020-04-22 22:04:40 +0200
committerGiampaolo Rodola <g.rodola@gmail.com>2020-04-22 22:04:40 +0200
commitfa5d5f2639b109dc4eeaf97156ff4a38a24c9f61 (patch)
tree01b2e218499c811d8452c7048b9aad095c3dcbce
parent63b074a32a061737ea79b2c15bd27af18cf47246 (diff)
downloadpsutil-fa5d5f2639b109dc4eeaf97156ff4a38a24c9f61.tar.gz
move test utils tests in their own new module
-rw-r--r--MANIFEST.in1
-rw-r--r--Makefile4
-rwxr-xr-xpsutil/tests/test_misc.py299
-rwxr-xr-xpsutil/tests/test_testutils.py325
4 files changed, 330 insertions, 299 deletions
diff --git a/MANIFEST.in b/MANIFEST.in
index 380a4fa2..d0d75240 100644
--- a/MANIFEST.in
+++ b/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
diff --git a/Makefile b/Makefile
index 5c4d5b70..fbd7ffcc 100644
--- a/Makefile
+++ b/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
diff --git a/psutil/tests/test_misc.py b/psutil/tests/test_misc.py
index c20cd941..ca0a0433 100755
--- a/psutil/tests/test_misc.py
+++ b/psutil/tests/test_misc.py
@@ -11,7 +11,6 @@ Miscellaneous tests.
import ast
import collections
-import contextlib
import errno
import json
import os
@@ -19,58 +18,33 @@ import pickle
import socket
import stat
-from psutil import FREEBSD
from psutil import LINUX
-from psutil import NETBSD
from psutil import POSIX
from psutil import WINDOWS
from psutil._common import memoize
from psutil._common import memoize_when_activated
from psutil._common import supports_ipv6
from psutil._common import wrap_numbers
-from psutil._common import open_text
-from psutil._common import open_binary
from psutil._compat import PY3
from psutil.tests import APPVEYOR
-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_proc_children_pair
-from psutil.tests import create_sockets
-from psutil.tests import create_zombie_proc
from psutil.tests import DEVNULL
-from psutil.tests import get_free_port
-from psutil.tests import get_test_subprocess
from psutil.tests import HAS_BATTERY
-from psutil.tests import HAS_CONNECTIONS_UNIX
from psutil.tests import HAS_MEMORY_MAPS
from psutil.tests import HAS_NET_IO_COUNTERS
from psutil.tests import HAS_SENSORS_BATTERY
from psutil.tests import HAS_SENSORS_FANS
from psutil.tests import HAS_SENSORS_TEMPERATURES
from psutil.tests import import_module_by_path
-from psutil.tests import is_namedtuple
from psutil.tests import mock
from psutil.tests import PYTHON_EXE
-from psutil.tests import reap_children
from psutil.tests import reload_module
-from psutil.tests import retry
from psutil.tests import ROOT_DIR
-from psutil.tests import safe_mkdir
-from psutil.tests import safe_rmpath
from psutil.tests import SCRIPTS_DIR
from psutil.tests import sh
-from psutil.tests import tcp_socketpair
-from psutil.tests import TESTFN
from psutil.tests import TOX
from psutil.tests import TRAVIS
from psutil.tests import unittest
-from psutil.tests import unix_socket_path
-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
@@ -787,279 +761,6 @@ class TestScripts(unittest.TestCase):
self.assert_stdout('sensors.py')
-# ===================================================================
-# --- Unit tests for test utilities.
-# ===================================================================
-
-
-class TestRetryDecorator(unittest.TestCase):
-
- @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(unittest.TestCase):
-
- def tearDown(self):
- safe_rmpath(TESTFN)
-
- 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):
- 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):
- 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):
- 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):
- 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(unittest.TestCase):
-
- def setUp(self):
- safe_rmpath(TESTFN)
-
- tearDown = setUp
-
- 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):
- 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
- 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):
- 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(unittest.TestCase):
-
- def test_reap_children(self):
- subp = get_test_subprocess()
- 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_create_proc_children_pair(self):
- p1, p2 = create_proc_children_pair()
- self.assertNotEqual(p1.pid, p2.pid)
- assert p1.is_running()
- assert p2.is_running()
- children = psutil.Process().children(recursive=True)
- self.assertEqual(len(children), 2)
- self.assertIn(p1, children)
- self.assertIn(p2, children)
- self.assertEqual(p1.ppid(), os.getpid())
- self.assertEqual(p2.ppid(), p1.pid)
-
- # make sure both of them are cleaned up
- reap_children()
- assert not p1.is_running()
- assert not p2.is_running()
- assert not psutil.tests._pids_started
- assert not psutil.tests._subprocesses_started
-
- @unittest.skipIf(not POSIX, "POSIX only")
- def test_create_zombie_proc(self):
- zpid = create_zombie_proc()
- self.addCleanup(reap_children, recursive=True)
- p = psutil.Process(zpid)
- self.assertEqual(p.status(), psutil.STATUS_ZOMBIE)
-
-
-class TestNetUtils(unittest.TestCase):
-
- 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):
- with unix_socket_path() as name:
- 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
- with unix_socket_path() as name:
- 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')
- with unix_socket_path() as name:
- 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)
-
-
-class TestOtherUtils(unittest.TestCase):
-
- 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
run(__file__)
diff --git a/psutil/tests/test_testutils.py b/psutil/tests/test_testutils.py
new file mode 100755
index 00000000..4c8b4a71
--- /dev/null
+++ b/psutil/tests/test_testutils.py
@@ -0,0 +1,325 @@
+#!/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
+
+from psutil import FREEBSD
+from psutil import NETBSD
+from psutil import POSIX
+from psutil._common import supports_ipv6
+from psutil._common import open_text
+from psutil._common import open_binary
+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 create_proc_children_pair
+from psutil.tests import create_sockets
+from psutil.tests import create_zombie_proc
+from psutil.tests import get_free_port
+from psutil.tests import get_test_subprocess
+from psutil.tests import HAS_CONNECTIONS_UNIX
+from psutil.tests import is_namedtuple
+from psutil.tests import mock
+from psutil.tests import reap_children
+from psutil.tests import retry
+from psutil.tests import safe_mkdir
+from psutil.tests import safe_rmpath
+from psutil.tests import tcp_socketpair
+from psutil.tests import TESTFN
+from psutil.tests import unittest
+from psutil.tests import unix_socket_path
+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(unittest.TestCase):
+
+ @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(unittest.TestCase):
+
+ def tearDown(self):
+ safe_rmpath(TESTFN)
+
+ 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):
+ 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):
+ 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):
+ 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):
+ 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(unittest.TestCase):
+
+ def setUp(self):
+ safe_rmpath(TESTFN)
+
+ tearDown = setUp
+
+ 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):
+ 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
+ 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):
+ 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(unittest.TestCase):
+
+ def test_reap_children(self):
+ subp = get_test_subprocess()
+ 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_create_proc_children_pair(self):
+ p1, p2 = create_proc_children_pair()
+ self.assertNotEqual(p1.pid, p2.pid)
+ assert p1.is_running()
+ assert p2.is_running()
+ children = psutil.Process().children(recursive=True)
+ self.assertEqual(len(children), 2)
+ self.assertIn(p1, children)
+ self.assertIn(p2, children)
+ self.assertEqual(p1.ppid(), os.getpid())
+ self.assertEqual(p2.ppid(), p1.pid)
+
+ # make sure both of them are cleaned up
+ reap_children()
+ assert not p1.is_running()
+ assert not p2.is_running()
+ assert not psutil.tests._pids_started
+ assert not psutil.tests._subprocesses_started
+
+ @unittest.skipIf(not POSIX, "POSIX only")
+ def test_create_zombie_proc(self):
+ zpid = create_zombie_proc()
+ self.addCleanup(reap_children, recursive=True)
+ p = psutil.Process(zpid)
+ self.assertEqual(p.status(), psutil.STATUS_ZOMBIE)
+
+
+class TestNetUtils(unittest.TestCase):
+
+ 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):
+ with unix_socket_path() as name:
+ 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
+ with unix_socket_path() as name:
+ 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')
+ with unix_socket_path() as name:
+ 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)
+
+
+class TestOtherUtils(unittest.TestCase):
+
+ 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
+ run(__file__)