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.authenticator.already_listening(17))
        self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
        mock_process.assert_called_once_with(4416)

 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.authenticator.already_listening(17))
     self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
     self.assertEqual(mock_process.call_count, 0)

def net_connections(kind, _pid=-1):
    """Return socket connections.  If pid == -1 return system-wide
    connections (as opposed to connections opened by one process only).
    Only INET sockets are returned (UNIX are not).
    """
    cmap = _common.conn_tmap.copy()
    if _pid == -1:
        cmap.pop('unix', 0)
    if kind not in cmap:
        raise ValueError("invalid %r kind argument; choose between %s"
                         % (kind, ', '.join([repr(x) for x in cmap])))
    families, types = _common.conn_tmap[kind]
    rawlist = cext.net_connections(_pid, families, types)
    ret = []
    for item in rawlist:
        fd, fam, type_, laddr, raddr, status, pid = item
        if fam not in families:
            continue
        if type_ not in types:
            continue
        status = TCP_STATUSES[status]
        if _pid == -1:
            nt = _common.sconn(fd, fam, type_, laddr, raddr, status, pid)
        else:
            nt = _common.pconn(fd, fam, type_, laddr, raddr, status)
        ret.append(nt)
    return ret

 def retrieve(self, kind, pid=None):
     if kind not in self.tmap:
         raise ValueError("invalid %r kind argument; choose between %s"
                          % (kind, ', '.join([repr(x) for x in self.tmap])))
     if pid is not None:
         inodes = self.get_proc_inodes(pid)
         if not inodes:
             # no connections for this process
             return []
     else:
         inodes = self.get_all_inodes()
     ret = []
     for f, family, type_ in self.tmap[kind]:
         if family in (socket.AF_INET, socket.AF_INET6):
             ls = self.process_inet(
                 "/proc/net/%s" % f, family, type_, inodes, filter_pid=pid)
         else:
             ls = self.process_unix(
                 "/proc/net/%s" % f, family, inodes, filter_pid=pid)
         for fd, family, type_, laddr, raddr, status, bound_pid in ls:
             if pid:
                 conn = _common.pconn(fd, family, type_, laddr, raddr,
                                      status)
             else:
                 conn = _common.sconn(fd, family, type_, laddr, raddr,
                                      status, bound_pid)
             ret.append(conn)
     return ret

 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.authenticator.already_listening(17)
     self.assertTrue(result)
     self.assertEqual(mock_get_utility.call_count, 1)
     mock_process.assert_called_once_with(4416)

def net_connections(kind):
    if kind not in _common.conn_tmap:
        raise ValueError("invalid %r kind argument; choose between %s"
                         % (kind, ', '.join([repr(x) for x in conn_tmap])))
    families, types = conn_tmap[kind]
    ret = []
    rawlist = cext.net_connections()
    for item in rawlist:
        fd, fam, type, laddr, raddr, status, pid = item
        # TODO: apply filter at C level
        if fam in families and type in types:
            status = TCP_STATUSES[status]
            nt = _common.sconn(fd, fam, type, laddr, raddr, status, pid)
            ret.append(nt)
    return ret

def net_connections(kind='inet'):
    # Note: on OSX this will fail with AccessDenied unless
    # the process is owned by root.
    ret = []
    for pid in pids():
        try:
            cons = Process(pid).connections(kind)
        except NoSuchProcess:
            continue
        else:
            if cons:
                for c in cons:
                    c = list(c) + [pid]
                    ret.append(_common.sconn(*c))
    return ret

def net_connections(kind, _pid=-1):
    """Return socket connections.  If pid == -1 return system-wide
    connections (as opposed to connections opened by one process only).
    """
    if kind not in conn_tmap:
        raise ValueError("invalid %r kind argument; choose between %s"
                         % (kind, ', '.join([repr(x) for x in conn_tmap])))
    families, types = conn_tmap[kind]
    rawlist = cext.net_connections(_pid, families, types)
    ret = []
    for item in rawlist:
        fd, fam, type, laddr, raddr, status, pid = item
        status = TCP_STATUSES[status]
        if _pid == -1:
            nt = _common.sconn(fd, fam, type, laddr, raddr, status, pid)
        else:
            nt = _common.pconn(fd, fam, type, laddr, raddr, status)
        ret.append(nt)
    return ret