我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用_socket.SOCK_STREAM。
def testSockName(self): # Testing getsockname() port = self._get_unused_port() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) sock.bind(("0.0.0.0", port)) name = sock.getsockname() # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate # it reasonable to get the host's addr in addition to 0.0.0.0. # At least for eCos. This is required for the S/390 to pass. try: my_ip_addr = socket.gethostbyname(socket.gethostname()) except socket.error: # Probably name lookup wasn't set up right; skip this test self.skipTest('name lookup failure') self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0]) self.assertEqual(name[1], port)
def testSockName(self): # Testing getsockname() port = support.find_unused_port() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) sock.bind(("0.0.0.0", port)) name = sock.getsockname() # XXX(nnorwitz): http://tinyurl.com/os5jz seems to indicate # it reasonable to get the host's addr in addition to 0.0.0.0. # At least for eCos. This is required for the S/390 to pass. try: my_ip_addr = socket.gethostbyname(socket.gethostname()) except OSError: # Probably name lookup wasn't set up right; skip this test self.skipTest('name lookup failure') self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0]) self.assertEqual(name[1], port)
def test_dealloc_warn(self): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) r = repr(sock) with self.assertWarns(ResourceWarning) as cm: sock = None support.gc_collect() self.assertIn(r, str(cm.warning.args[0])) # An open socket file object gets dereferenced after the socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) f = sock.makefile('rb') r = repr(sock) sock = None support.gc_collect() with self.assertWarns(ResourceWarning): f = None support.gc_collect()
def testTypes(self): families = [socket.AF_INET, socket.AF_INET6] types = [socket.SOCK_STREAM, socket.SOCK_DGRAM] for f in families: for t in types: try: source = socket.socket(f, t) except OSError: continue # This combination is not supported try: data = source.share(os.getpid()) shared = socket.fromshare(data) try: self.compareSockets(source, shared) finally: shared.close() finally: source.close()
def try_address(host, port=0, family=socket.AF_INET): """Try to bind a socket on the given host:port and return True if that has been possible.""" try: sock = socket.socket(family, socket.SOCK_STREAM) sock.bind((host, port)) except (socket.error, socket.gaierror): return False else: sock.close() return True
def setUp(self): self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.port = test_support.bind_port(self.serv) self.serv.listen(1)
def clientSetUp(self): self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def test_csocket_repr(self): s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) try: expected = ('<socket object, fd=%s, family=%s, type=%s, protocol=%s>' % (s.fileno(), s.family, s.type, s.proto)) self.assertEqual(repr(s), expected) finally: s.close() expected = ('<socket object, fd=-1, family=%s, type=%s, protocol=%s>' % (s.family, s.type, s.proto)) self.assertEqual(repr(s), expected)
def test_weakref(self): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) p = weakref.proxy(s) self.assertEqual(p.fileno(), s.fileno()) s.close() s = None try: p.fileno() except ReferenceError: pass else: self.fail('Socket proxy still exists')
def testGetSockOpt(self): # Testing getsockopt() # We know a socket should start without reuse==0 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) self.assertFalse(reuse != 0, "initial mode is reuse")
def testSetSockOpt(self): # Testing setsockopt() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) self.assertFalse(reuse == 0, "failed to set reuse mode")
def testSendAfterClose(self): # testing send() after close() with timeout sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1) sock.close() self.assertRaises(socket.error, sock.send, "spam")
def testNewAttributes(self): # testing .family, .type and .protocol sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.assertEqual(sock.family, socket.AF_INET) self.assertEqual(sock.type, socket.SOCK_STREAM) self.assertEqual(sock.proto, 0) sock.close()
def test_listen_backlog(self): for backlog in 0, -1: srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.bind((HOST, 0)) srv.listen(backlog) srv.close()
def test_listen_backlog_overflow(self): # Issue 15989 import _testcapi srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.bind((HOST, 0)) self.assertRaises(OverflowError, srv.listen, _testcapi.INT_MAX + 1) srv.close()
def test_flowinfo(self): self.assertRaises(OverflowError, socket.getnameinfo, ('::1',0, 0xffffffff), 0) s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) try: self.assertRaises(OverflowError, s.bind, ('::1', 0, -10)) finally: s.close()
def testFromFd(self): # Testing fromfd() fd = self.cli_conn.fileno() sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(sock.close) msg = sock.recv(1024) self.assertEqual(msg, MSG)
def test_connect(self): port = test_support.find_unused_port() cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.addCleanup(cli.close) with self.assertRaises(socket.error) as cm: cli.connect((HOST, port)) self.assertEqual(cm.exception.errno, errno.ECONNREFUSED)
def testLinuxAbstractNamespace(self): address = "\x00python-test-hello\x00\xff" s1 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s1.bind(address) s1.listen(1) s2 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s2.connect(s1.getsockname()) s1.accept() self.assertEqual(s1.getsockname(), address) self.assertEqual(s2.getpeername(), address)
def testMaxName(self): address = "\x00" + "h" * (self.UNIX_PATH_MAX - 1) s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.bind(address) self.assertEqual(s.getsockname(), address)
def testNameOverflow(self): address = "\x00" + "h" * self.UNIX_PATH_MAX s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.assertRaises(socket.error, s.bind, address)
def setUp(self): self.srv = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM) self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE, TIPC_LOWER, TIPC_UPPER) self.srv.bind(srvaddr) self.srv.listen(5) self.serverExplicitReady() self.conn, self.connaddr = self.srv.accept()
def clientSetUp(self): # There is a hittable race between serverExplicitReady() and the # accept() call; sleep a little while to avoid it, otherwise # we could get an exception time.sleep(0.1) self.cli = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM) addr = (socket.TIPC_ADDR_NAME, TIPC_STYPE, TIPC_LOWER + (TIPC_UPPER - TIPC_LOWER) / 2, 0) self.cli.connect(addr) self.cliaddr = self.cli.getsockname()
