python中bind的用法_Python socket.bind方法代码示例
本文整理汇总了Python中socket.bind方法的典型用法代码示例。如果您正苦于以下问题：Python socket.bind方法的具体用法？Python socket.bind怎么用？Python socket.bind使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块socket的用法示例。
在下文中一共展示了socket.bind方法的21个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _resolveIPv6
点赞 6
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _resolveIPv6(ip, port):
"""
Resolve an IPv6 literal into an IPv6 address.
This is necessary to resolve any embedded scope identifiers to the relevant
C{sin6_scope_id} for use with C{socket.connect()}, C{socket.listen()}, or
C{socket.bind()}; see U{RFC 3493 } for
more information.
@param ip: An IPv6 address literal.
@type ip: C{str}
@param port: A port number.
@type port: C{int}
@return: a 4-tuple of C{(host, port, flow, scope)}, suitable for use as an
IPv6 address.
@raise socket.gaierror: if either the IP or port is not numeric as it
should be.
"""
return socket.getaddrinfo(ip, port, 0, 0, 0, _NUMERIC_ONLY)[0][4]
开发者ID:proxysh，项目名称:Safejumper-for-Desktop，代码行数:24，
示例2: _testTCPServerOption
点赞 6
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testTCPServerOption(self, level, option, values):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._testSetAndGetOption(sock, level, option, values)
# now bind and listen on the socket i.e. cause the implementation socket to be created
sock.bind( (HOST, PORT) )
sock.listen(50)
retrieved_option_value = sock.getsockopt(level, option)
msg = "TCP server option value '(%s,%s)'='%s' did not propagate to implementation socket. Got %s" % (level, option, values[-1], retrieved_option_value)
if option == socket.SO_RCVBUF:
# NOTE: see similar bsd/solaris workaround above
self.assert_(retrieved_option_value >= values[-1], msg)
else:
self.failUnlessEqual(retrieved_option_value, values[-1], msg)
self._testSetAndGetOption(sock, level, option, values)
finally:
sock.close()
开发者ID:Acmesec，项目名称:CTFCrackTools-V2，代码行数:20，
示例3: testSO_ERROR
点赞 6
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testSO_ERROR(self):
good = bad = None
try:
good = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
good.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
good.bind((HOST, PORT))
good.listen(1)
bad = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
bad.bind((HOST, PORT))
bad.listen(1)
self.fail("Listen operation against same port did not generate an expected error")
except socket.error, se:
self.failUnlessEqual(bad.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR), se[0])
# try again, should now be reset
self.failUnlessEqual(bad.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR), 0)
开发者ID:Acmesec，项目名称:CTFCrackTools-V2，代码行数:18，
示例4: startVideoGeneratorServer
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def startVideoGeneratorServer():
server_address = (settings.server_location, int(settings.server_port_vid_gen))
print('Starting video generator server on %s port %s' % server_address)
socket.bind(server_address)
socket.listen(5)
thread = Thread(target=waitConnect)
thread.start()
servertick = Thread(target=serverTick)
servertick.start()
开发者ID:HA6Bots，项目名称:Automatic-Youtube-Reddit-Text-To-Speech-Video-Generator-and-Uploader，代码行数:11，
示例5: get_sockaddr
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def get_sockaddr(host, port, family):
"""Return a fully qualified socket address that can be passed to
:func:`socket.bind`."""
if family == af_unix:
return host.split("://", 1)[1]
try:
res = socket.getaddrinfo(
host, port, family, socket.SOCK_STREAM, socket.IPPROTO_TCP
)
except socket.gaierror:
return host, port
return res[0][4]
开发者ID:Frank-qlu，项目名称:recruit，代码行数:14，
示例6: __init__
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def __init__(self, host, port, bindAddress, connector, reactor=None):
# BaseClient.__init__ is invoked later
self.connector = connector
self.addr = (host, port)
whenDone = self.resolveAddress
err = None
skt = None
if abstract.isIPAddress(host):
self._requiresResolution = False
elif abstract.isIPv6Address(host):
self._requiresResolution = False
self.addr = _resolveIPv6(host, port)
self.addressFamily = socket.AF_INET6
self._addressType = address.IPv6Address
else:
self._requiresResolution = True
try:
skt = self.createInternetSocket()
except socket.error as se:
err = error.ConnectBindError(se.args[0], se.args[1])
whenDone = None
if whenDone and bindAddress is not None:
try:
if abstract.isIPv6Address(bindAddress[0]):
bindinfo = _resolveIPv6(*bindAddress)
else:
bindinfo = bindAddress
skt.bind(bindinfo)
except socket.error as se:
err = error.ConnectBindError(se.args[0], se.args[1])
whenDone = None
self._finishInit(whenDone, skt, err, reactor)
开发者ID:proxysh，项目名称:Safejumper-for-Desktop，代码行数:36，
示例7: startListening
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def startListening(self):
"""Create and bind my socket, and begin listening on it.
This is called on unserialization, and must be called after creating a
server to begin listening on the specified port.
"""
if self._preexistingSocket is None:
# Create a new socket and make it listen
try:
skt = self.createInternetSocket()
if self.addressFamily == socket.AF_INET6:
addr = _resolveIPv6(self.interface, self.port)
else:
addr = (self.interface, self.port)
skt.bind(addr)
except socket.error as le:
raise CannotListenError(self.interface, self.port, le)
skt.listen(self.backlog)
else:
# Re-use the externally specified socket
skt = self._preexistingSocket
self._preexistingSocket = None
# Avoid shutting it down at the end.
self._shouldShutdown = False
# Make sure that if we listened on port 0, we update that to
# reflect what the OS actually assigned us.
self._realPortNumber = skt.getsockname()[1]
log.msg("%s starting on %s" % (
self._getLogPrefix(self.factory), self._realPortNumber))
# The order of the next 5 lines is kind of bizarre. If no one
# can explain it, perhaps we should re-arrange them.
self.factory.doStart()
self.connected = True
self.socket = skt
self.fileno = self.socket.fileno
self.numberAccepts = 100
self.startReading()
开发者ID:proxysh，项目名称:Safejumper-for-Desktop，代码行数:43，
示例8: setUp
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def setUp(self):
self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.serv.bind((self.HOST, self.PORT))
self.serv.listen(1)
开发者ID:ofermend，项目名称:medicare-demo，代码行数:7，
示例9: testSockName
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testSockName(self):
# Testing getsockname()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("0.0.0.0", PORT+1))
name = sock.getsockname()
self.assertEqual(name, ("0.0.0.0", PORT+1))
开发者ID:ofermend，项目名称:medicare-demo，代码行数:8，
示例10: _testUDPOption
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testUDPOption(self, level, option, values):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self._testSetAndGetOption(sock, level, option, values)
# now bind the socket i.e. cause the implementation socket to be created
sock.bind( (HOST, PORT) )
self.failUnlessEqual(sock.getsockopt(level, option), values[-1], \
"Option value '(%s, %s)'='%s' did not propagate to implementation socket" % (level, option, values[-1]) )
self._testSetAndGetOption(sock, level, option, values)
finally:
sock.close()
开发者ID:ofermend，项目名称:medicare-demo，代码行数:13，
示例11: _testTCPClientOption
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testTCPClientOption(self, level, option, values):
sock = None
try:
# First listen on a server socket, so that the connection won't be refused.
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.bind( (HOST, PORT) )
server_sock.listen(50)
# Now do the tests
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._testSetAndGetOption(sock, level, option, values)
# now connect the socket i.e. cause the implementation socket to be created
# First bind, so that the SO_REUSEADDR setting propagates
sock.bind( (HOST, PORT+1) )
sock.connect( (HOST, PORT) )
msg = "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1])
if option in (socket.SO_RCVBUF, socket.SO_SNDBUF):
# NOTE: there's no guarantee that bufsize will be the
# exact setsockopt value, particularly after
# establishing a connection. seems it will be *at least*
# the values we test (which are rather small) on
# BSDs.
self.assert_(sock.getsockopt(level, option) >= values[-1], msg)
else:
self.failUnlessEqual(sock.getsockopt(level, option), values[-1], msg)
self._testSetAndGetOption(sock, level, option, values)
finally:
server_sock.close()
if sock:
sock.close()
开发者ID:ofermend，项目名称:medicare-demo，代码行数:31，
示例12: testBindSpecific
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindSpecific(self):
self.sock.bind( (self.HOST, self.PORT) ) # Use a specific port
actual_port = self.sock.getsockname()[1]
self.failUnless(actual_port == self.PORT,
"Binding to specific port number should have returned same number: %d != %d" % (actual_port, self.PORT))
开发者ID:ofermend，项目名称:medicare-demo，代码行数:7，
示例13: testBindEphemeral
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindEphemeral(self):
self.sock.bind( (self.HOST, 0) ) # let system choose a free port
self.failUnless(self.sock.getsockname()[1] != 0, "Binding to port zero should have allocated an ephemeral port number")
开发者ID:ofermend，项目名称:medicare-demo，代码行数:5，
示例14: testShutdown
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testShutdown(self):
self.sock.bind( (self.HOST, self.PORT) )
self.sock.shutdown(socket.SHUT_RDWR)
开发者ID:ofermend，项目名称:medicare-demo，代码行数:5，
示例15: testBroadcast
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBroadcast(self):
self.serv.bind( ("", self.PORT) )
msg = self.serv.recv(len(EIGHT_BIT_MSG))
self.assertEqual(msg, EIGHT_BIT_MSG)
开发者ID:ofermend，项目名称:medicare-demo，代码行数:6，
示例16: _testNonBlockingConnect
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testNonBlockingConnect(self):
# Testing non-blocking connect
self.cli.setblocking(0)
result = self.cli.connect_ex((self.HOST, self.PORT))
rfds, wfds, xfds = select.select([], [self.cli], [])
self.failUnless(self.cli in wfds)
try:
self.cli.send(MSG)
except socket.error:
self.fail("Sending on connected socket should not have raised socket.error")
#
# AMAK: 20070518
# Introduced a new test for connect with bind to specific local address
#
开发者ID:ofermend，项目名称:medicare-demo，代码行数:17，
示例17: testBindException
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindException(self):
# First bind to the target port
self.s.bind( (HOST, PORT) )
self.s.listen(50)
try:
# And then try to bind again
t = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
t.bind( (HOST, PORT) )
t.listen(50)
except socket.error, se:
self.failUnlessEqual(se[0], errno.EADDRINUSE)
开发者ID:ofermend，项目名称:medicare-demo，代码行数:13，
示例18: testBindNonTupleEndpointRaisesTypeError
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindNonTupleEndpointRaisesTypeError(self):
try:
self.socket.bind(HOST, PORT)
except TypeError:
pass
else:
self.fail("Illegal non-tuple bind address did not raise TypeError")
开发者ID:ofermend，项目名称:medicare-demo，代码行数:9，
示例19: testAddrTupleIDNAHostname
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testAddrTupleIDNAHostname(self):
idna_domain = u"al\u00e1n.com"
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if socket.supports('idna'):
try:
s.bind( (idna_domain, 80) )
except socket.error:
# We're not worried about socket errors, i.e. bind problems, etc.
pass
except Exception, x:
self.fail("Unexpected exception raised for socket.bind(%s)" % repr(idna_domain))
开发者ID:ofermend，项目名称:medicare-demo，代码行数:13，
示例20: testGetsocknameBoundNoImpl
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testGetsocknameBoundNoImpl(self):
self.s.bind( ("localhost", 0) )
try:
self.s.getsockname()
except socket.error, se:
self.fail("getsockname() on bound socket should have not raised socket.error")
开发者ID:ofermend，项目名称:medicare-demo，代码行数:8，
示例21: _create_impl_socket
点赞 5
# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _create_impl_socket(self):
self.s.bind(("localhost", 0))
self.s.listen(5)
开发者ID:ofermend，项目名称:medicare-demo，代码行数:5，
注：本文中的socket.bind方法示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。