我们从Python开源项目中，提取了以下21个代码示例，用于说明如何使用psutil._common()。
def get_ips(): """ >>> from psutil._common import snic >>> import mock >>> MOCK = { ... "awdl0": [snic(family=30, address="fe80::3854:80ff:fe54:7bf8%awdl0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)], ... "en0": [snic(family=2, address="192.168.10.200", netmask="255.255.255.0", broadcast="192.168.10.255", ptp=None), ... snic(family=30, address="fe80::6e40:8ff:feac:4f94%en0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)], ... "bridge0": [snic(family=18, address="6e:40:08:ca:60:00", netmask=None, broadcast=None, ptp=None)], ... "lo0": [snic(family=2, address="127.0.0.1", netmask="255.0.0.0", broadcast=None, ptp=None), ... snic(family=30, address="fe80::1%lo0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)]} >>> with mock.patch("psutil.net_if_addrs", side_effect=lambda: MOCK): ... get_ips() ['127.0.0.1/255.0.0.0', '192.168.10.200/255.255.255.0'] """ return sorted(flatten(chain(get_network().values())))
def test_parse_environ_block(self): from psutil._common import parse_environ_block def k(s): return s.upper() if WINDOWS else s self.assertEqual(parse_environ_block("a=1\0"), {k("a"): "1"}) self.assertEqual(parse_environ_block("a=1\0b=2\0\0"), {k("a"): "1", k("b"): "2"}) self.assertEqual(parse_environ_block("a=1\0b=\0\0"), {k("a"): "1", k("b"): ""}) # ignore everything after \0\0 self.assertEqual(parse_environ_block("a=1\0b=2\0\0c=3\0"), {k("a"): "1", k("b"): "2"}) # ignore everything that is not an assignment self.assertEqual(parse_environ_block("xxx\0a=1\0"), {k("a"): "1"}) self.assertEqual(parse_environ_block("a=1\0=b=2\0"), {k("a"): "1"}) # do not fail if the block is incomplete self.assertEqual(parse_environ_block("a=1\0b=2"), {k("a"): "1"})
def test_supports_ipv6(self): if supports_ipv6(): with mock.patch('psutil._common.socket') as s: s.has_ipv6 = False assert not supports_ipv6() with mock.patch('psutil._common.socket.socket', side_effect=socket.error) as s: assert not supports_ipv6() assert s.called with mock.patch('psutil._common.socket.socket', side_effect=socket.gaierror) as s: assert not supports_ipv6() assert s.called with mock.patch('psutil._common.socket.socket.bind', side_effect=socket.gaierror) as s: assert not supports_ipv6() assert s.called else: if hasattr(socket, 'AF_INET6'): with self.assertRaises(Exception): sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) sock.bind(("::1", 0))
def test_isfile_strict(self): from psutil._common import isfile_strict this_file = os.path.abspath(__file__) assert isfile_strict(this_file) assert not isfile_strict(os.path.dirname(this_file)) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EPERM, "foo")): self.assertRaises(OSError, isfile_strict, this_file) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EACCES, "foo")): self.assertRaises(OSError, isfile_strict, this_file) with mock.patch('psutil._common.os.stat', side_effect=OSError(errno.EINVAL, "foo")): assert not isfile_strict(this_file) with mock.patch('psutil._common.stat.S_ISREG', return_value=False): assert not isfile_strict(this_file)
def test_race_condition(self, mock_get_utility, mock_process, mock_net): # This tests a race condition, or permission problem, or OS # incompatibility in which, for some reason, no process name can be # found to match the identified listening PID. from psutil._common import sconn conns = [ sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30), raddr=(), status="LISTEN", pid=None), sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783), raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234), sconn(fd=-1, family=10, type=1, laddr=("::1", 54321), raddr=("::1", 111), status="CLOSE_WAIT", pid=None), sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17), raddr=(), status="LISTEN", pid=4416)] mock_net.return_value = conns mock_process.side_effect = psutil.NoSuchProcess("No such PID") # We simulate being unable to find the process name of PID 4416, # which results in returning False. self.assertFalse(self._call(17)) self.assertEqual(mock_get_utility.generic_notification.call_count, 0) mock_process.assert_called_once_with(4416)
def test_listening_ipv4(self, mock_get_utility, mock_process, mock_net): from psutil._common import sconn conns = [ sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30), raddr=(), status="LISTEN", pid=None), sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783), raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234), sconn(fd=-1, family=10, type=1, laddr=("::1", 54321), raddr=("::1", 111), status="CLOSE_WAIT", pid=None), sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17), raddr=(), status="LISTEN", pid=4416)] mock_net.return_value = conns mock_process.name.return_value = "inetd" result = self._call(17) self.assertTrue(result) self.assertEqual(mock_get_utility.call_count, 1) mock_process.assert_called_once_with(4416)
def test_listening_ipv6(self, mock_get_utility, mock_process, mock_net): from psutil._common import sconn conns = [ sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30), raddr=(), status="LISTEN", pid=None), sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783), raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234), sconn(fd=-1, family=10, type=1, laddr=("::1", 54321), raddr=("::1", 111), status="CLOSE_WAIT", pid=None), sconn(fd=3, family=10, type=1, laddr=("::", 12345), raddr=(), status="LISTEN", pid=4420), sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17), raddr=(), status="LISTEN", pid=4416)] mock_net.return_value = conns mock_process.name.return_value = "inetd" result = self._call(12345) self.assertTrue(result) self.assertEqual(mock_get_utility.call_count, 1) mock_process.assert_called_once_with(4420)
def test_net_connections(self): def check(cons, families, types_): for conn in cons: self.assertIn(conn.family, families, msg=conn) if conn.family != getattr(socket, 'AF_UNIX', object()): self.assertIn(conn.type, types_, msg=conn) self.assertIsInstance(conn.status, (str, unicode)) from psutil._common import conn_tmap for kind, groups in conn_tmap.items(): if SUNOS and kind == 'unix': continue families, types_ = groups cons = psutil.net_connections(kind) self.assertEqual(len(cons), len(set(cons))) check(cons, families, types_)
def get_network(families=[socket.AF_INET]): """ >>> from psutil._common import snic >>> import mock >>> MOCK = { ... "awdl0": [snic(family=30, address="fe80::3854:80ff:fe54:7bf8%awdl0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)], ... "en0": [snic(family=2, address="192.168.10.200", netmask="255.255.255.0", broadcast="192.168.10.255", ptp=None), ... snic(family=30, address="fe80::6e40:8ff:feac:4f94%en0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)], ... "bridge0": [snic(family=18, address="6e:40:08:ca:60:00", netmask=None, broadcast=None, ptp=None)], ... "lo0": [snic(family=2, address="127.0.0.1", netmask="255.0.0.0", broadcast=None, ptp=None), ... snic(family=30, address="fe80::1%lo0", netmask="ffff:ffff:ffff:ffff::", broadcast=None, ptp=None)]} >>> with mock.patch("psutil.net_if_addrs", side_effect=lambda: MOCK): ... data_inet = get_network([socket.AF_INET]) ... sorted(data_inet.keys()) ['en0', 'lo0'] >>> with mock.patch("psutil.net_if_addrs", side_effect=lambda: MOCK): ... sorted(data_inet.values()) [[u'127.0.0.1/255.0.0.0'], [u'192.168.10.200/255.255.255.0']] >>> with mock.patch("psutil.net_if_addrs", side_effect=lambda: MOCK): ... data_inet6 = get_network([socket.AF_INET6]) ... sorted(flatten(data_inet6.values())) ['fe80::1%lo0/ffff:ffff:ffff:ffff::', 'fe80::3854:80ff:fe54:7bf8%awdl0/ffff:ffff:ffff:ffff::', 'fe80::6e40:8ff:feac:4f94%en0/ffff:ffff:ffff:ffff::'] """ nic = psutil.net_if_addrs() ips = defaultdict(list) # return nic for card, addresses in nic.items(): for address in addresses: if address.family in families: ips[card].append("{0.address}/{0.netmask}".format(address)) return dict(ips) # return flatten([[d.address for d in data if is_valid_ip(d)] for card, data in nic.items()])
def compare_proc_sys_cons(self, pid, proc_cons): from psutil._common import pconn sys_cons = [c[:-1] for c in psutil.net_connections(kind='all') if c.pid == pid] if FREEBSD: # on FreeBSD all fds are set to -1 proc_cons = [pconn(*[-1] + list(x[1:])) for x in proc_cons] self.assertEqual(sorted(proc_cons), sorted(sys_cons))
def test_net_connections(self): def check(cons, families, types_): for conn in cons: self.assertIn(conn.family, families, msg=conn) if conn.family != getattr(socket, 'AF_UNIX', object()): self.assertIn(conn.type, types_, msg=conn) from psutil._common import conn_tmap for kind, groups in conn_tmap.items(): if SUNOS and kind == 'unix': continue families, types_ = groups cons = psutil.net_connections(kind) self.assertEqual(len(cons), len(set(cons))) check(cons, families, types_)
def test_not_listening(self, mock_get_utility, mock_process, mock_net): from psutil._common import sconn conns = [ sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30), raddr=(), status="LISTEN", pid=None), sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783), raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234), sconn(fd=-1, family=10, type=1, laddr=("::1", 54321), raddr=("::1", 111), status="CLOSE_WAIT", pid=None)] mock_net.return_value = conns mock_process.name.return_value = "inetd" self.assertFalse(self._call(17)) self.assertEqual(mock_get_utility.generic_notification.call_count, 0) self.assertEqual(mock_process.call_count, 0)
