Python socket.SO_TYPE属性代码示例

本文整理汇总了Python中socket.SO_TYPE属性的典型用法代码示例。如果您正苦于以下问题:Python socket.SO_TYPE属性的具体用法?Python socket.SO_TYPE怎么用?Python socket.SO_TYPE使用的例子?那么恭喜您, 这里精选的属性代码示例或许可以为您提供帮助。您也可以进一步了解该属性所在模块socket的用法示例。

在下文中一共展示了socket.SO_TYPE属性的9个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: is_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def is_socket(fd):
    """ Determine if the file descriptor is a socket.
        Return ``False`` if querying the socket type of `fd` raises an
        error; otherwise return ``True``.
        """
    result = False
    file_socket = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW)
    try:
        socket_type = file_socket.getsockopt(
            socket.SOL_SOCKET, socket.SO_TYPE)
    except socket.error, exc:
        exc_errno = exc.args[0]
        if exc_errno == errno.ENOTSOCK:
            # Socket operation on non-socket
            pass
        else:
            # Some other socket error
            result = True 
开发者ID:blackye,项目名称:luscan-devel,代码行数:24,代码来源:daemon.py


示例2: _is_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def _is_socket(cls, stream):
        """Check if the given stream is a socket."""
        try:
            fd = stream.fileno()
        except ValueError:
            # If it has no file descriptor, it's not a socket
            return False
        sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW)
        try:
            # This will raise a socket.error if it's not a socket
            sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE)
        except socket.error as ex:
            if ex.args[0] != errno.ENOTSOCK:
                # It must be a socket
                return True
        else:
            # If an exception wasn't raised, it's a socket
            return True 
开发者ID:jnrbsn,项目名称:daemonocle,代码行数:21,代码来源:core.py


示例3: test_create_sockets

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
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) 
开发者ID:birforce,项目名称:vnpy_crypto,代码行数:17,代码来源:test_misc.py


示例4: check_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def check_socket(self, sock, conn=None):
        """Given a socket, makes sure it matches the one obtained
        via psutil. It assumes this process created one connection
        only (the one supposed to be checked).
        """
        if conn is None:
            conn = self.get_conn_from_sock(sock)
        check_connection_ntuple(conn)
        # fd, family, type
        if conn.fd != -1:
            self.assertEqual(conn.fd, sock.fileno())
        self.assertEqual(conn.family, sock.family)
        # see: http://bugs.python.org/issue30204
        self.assertEqual(
            conn.type, sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE))
        # local address
        laddr = sock.getsockname()
        if not laddr and PY3 and isinstance(laddr, bytes):
            # See: http://bugs.python.org/issue30205
            laddr = laddr.decode()
        if sock.family == AF_INET6:
            laddr = laddr[:2]
        if sock.family == AF_UNIX and OPENBSD:
            # No addresses are set for UNIX sockets on OpenBSD.
            pass
        else:
            self.assertEqual(conn.laddr, laddr)
        # XXX Solaris can't retrieve system-wide UNIX sockets
        if sock.family == AF_UNIX and HAS_CONNECTIONS_UNIX:
            cons = thisproc.connections(kind='all')
            self.compare_procsys_connections(os.getpid(), cons)
        return conn 
开发者ID:birforce,项目名称:vnpy_crypto,代码行数:37,代码来源:test_connections.py


示例5: check_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def check_socket(self, sock):
        """Given a socket, makes sure it matches the one obtained
        via psutil. It assumes this process created one connection
        only (the one supposed to be checked).
        """
        conn = self.get_conn_from_sock(sock)
        self.check_connection_ntuple(conn)
        # fd, family, type
        if conn.fd != -1:
            self.assertEqual(conn.fd, sock.fileno())
        self.assertEqual(conn.family, sock.family)
        # see: http://bugs.python.org/issue30204
        self.assertEqual(
            conn.type, sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE))
        # local address
        laddr = sock.getsockname()
        if not laddr and PY3 and isinstance(laddr, bytes):
            # See: http://bugs.python.org/issue30205
            laddr = laddr.decode()
        if sock.family == AF_INET6:
            laddr = laddr[:2]
        if sock.family == AF_UNIX and OPENBSD:
            # No addresses are set for UNIX sockets on OpenBSD.
            pass
        else:
            self.assertEqual(conn.laddr, laddr)
        # XXX Solaris can't retrieve system-wide UNIX sockets
        if sock.family == AF_UNIX and HAS_CONNECTIONS_UNIX:
            cons = thisproc.connections(kind='all')
            self.compare_procsys_connections(os.getpid(), cons, kind='all')
        return conn 
开发者ID:giampaolo,项目名称:psutil,代码行数:36,代码来源:test_connections.py


示例6: is_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def is_socket(fd):
    """Determine if the file descriptor is a socket."""
    file_socket = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW)
    try:
        file_socket.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE)
    except socket.error as ex:
        return ex.args[0] != errno.ENOTSOCK
    else:
        return True 
开发者ID:edgedb,项目名称:edgedb,代码行数:12,代码来源:lib.py


示例7: is_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def is_socket(fd):
    """ Determine if the file descriptor is a socket.
        Return ``False`` if querying the socket type of `fd` raises an
        error; otherwise return ``True``.
        """
    result = False
    file_socket = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW)
    try:
        file_socket.getsockopt(
            socket.SOL_SOCKET, socket.SO_TYPE)
    except socket.error as exc:
        exc_errno = exc.args[0]
        if exc_errno == errno.ENOTSOCK:
            # Socket operation on non-socket
            pass
        else:
            # Some other socket error
            result = True
    else:
        # No error getting socket type
        result = True
    return result 
开发者ID:candlepin,项目名称:virt-who,代码行数:29,代码来源:daemon.py


示例8: testSO_TYPE

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def testSO_TYPE(self):
        for socket_type in [socket.SOCK_STREAM, socket.SOCK_DGRAM]:
            s = socket.socket(socket.AF_INET, socket_type)
            self.failUnlessEqual(s.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE), socket_type) 
开发者ID:Acmesec,项目名称:CTFCrackTools-V2,代码行数:6,代码来源:test_socket.py


示例9: _create

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_TYPE [as 别名]
def _create(cls, sock, server_side=False, do_handshake_on_connect=True,
                suppress_ragged_eofs=True, server_hostname=None,
                context=None, session=None):
        if sock.getsockopt(SOL_SOCKET, SO_TYPE) != SOCK_STREAM:
            raise NotImplementedError("only stream sockets are supported")
        if server_side:
            if server_hostname:
                raise ValueError("server_hostname can only be specified "
                                 "in client mode")
            if session is not None:
                raise ValueError("session can only be specified in "
                                 "client mode")
        if context.check_hostname and not server_hostname:
            raise ValueError("check_hostname requires server_hostname")
        kwargs = dict(
            family=sock.family, type=sock.type, proto=sock.proto,
            fileno=sock.fileno()
        )
        self = cls.__new__(cls, **kwargs)
        super(SSLSocket, self).__init__(**kwargs)
        self.settimeout(sock.gettimeout())
        sock.detach()
        self._context = context
        self._session = session
        self._closed = False
        self._sslobj = None
        self.server_side = server_side
        self.server_hostname = context._encode_hostname(server_hostname)
        self.do_handshake_on_connect = do_handshake_on_connect
        self.suppress_ragged_eofs = suppress_ragged_eofs
        # See if we are connected
        try:
            self.getpeername()
        except OSError as e:
            if e.errno != errno.ENOTCONN:
                raise
            connected = False
        else:
            connected = True
        self._connected = connected
        if connected:
            # create the SSL object
            try:
                self._sslobj = self._context._wrap_socket(
                    self, server_side, self.server_hostname,
                    owner=self, session=self._session,
                )
                if do_handshake_on_connect:
                    timeout = self.gettimeout()
                    if timeout == 0.0:
                        # non-blocking
                        raise ValueError("do_handshake_on_connect should not be specified for non-blocking sockets")
                    self.do_handshake()
            except (OSError, ValueError):
                self.close()
                raise
        return self 
开发者ID:CedricGuillemet,项目名称:Imogen,代码行数:63,代码来源:ssl.py



注:本文中的socket.SO_TYPE属性示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。