Menu
▾
▴
[Jython-checkins] SF.net SVN: jython: [4952] branches/asm/Lib
|
From: <am...@us...> - 2008-07-15 22:51:54
|
Revision: 4952
http://jython.svn.sourceforge.net/jython/?rev=4952&view=rev
Author: amak
Date: 2008-07-15 15:51:51 -0700 (Tue, 15 Jul 2008)
Log Message:
-----------
Backing out what I thought were simple changes, which appear to have broken the build.
Modified Paths:
--------------
branches/asm/Lib/socket.py
branches/asm/Lib/test/test_socket.py
Removed Paths:
-------------
branches/asm/Lib/test/test_socket_ipv6.py
Modified: branches/asm/Lib/socket.py
===================================================================
--- branches/asm/Lib/socket.py 2008-07-15 22:37:38 UTC (rev 4951)
+++ branches/asm/Lib/socket.py 2008-07-15 22:51:51 UTC (rev 4952)
@@ -147,8 +147,8 @@
SHUT_WR = 1
SHUT_RDWR = 2
-__all__ = ['AF_UNSPEC', 'AF_INET', 'AF_INET6', 'AI_PASSIVE', 'SOCK_DGRAM',
- 'SOCK_RAW', 'SOCK_RDM', 'SOCK_SEQPACKET', 'SOCK_STREAM', 'SOL_SOCKET',
+__all__ = [ 'AF_INET', 'SOCK_DGRAM', 'SOCK_RAW',
+ 'SOCK_RDM', 'SOCK_SEQPACKET', 'SOCK_STREAM', 'SOL_SOCKET',
'SO_BROADCAST', 'SO_KEEPALIVE', 'SO_LINGER', 'SO_OOBINLINE',
'SO_RCVBUF', 'SO_REUSEADDR', 'SO_SNDBUF', 'SO_TIMEOUT', 'TCP_NODELAY',
'SocketType', 'error', 'herror', 'gaierror', 'timeout',
@@ -158,12 +158,8 @@
'SHUT_RD', 'SHUT_WR', 'SHUT_RDWR',
]
-AF_UNSPEC = 0
AF_INET = 2
-AF_INET6 = 23
-AI_PASSIVE=1
-
SOCK_DGRAM = 1
SOCK_STREAM = 2
SOCK_RAW = 3 # not supported
@@ -376,7 +372,8 @@
bytes_sent = self.jchannel.send(byte_buf, socket_address)
return bytes_sent
- def sendto(self, byte_array, host, port, flags):
+ def sendto(self, byte_array, address, flags):
+ host, port = _unpack_address_tuple(address)
socket_address = java.net.InetSocketAddress(host, port)
if self.mode == MODE_TIMEOUT:
return self._do_send_net(byte_array, socket_address, flags)
@@ -433,10 +430,6 @@
else:
return self._do_receive_nio(0, num_bytes, flags)
-# Name and address functions
-
-has_ipv6 = 1
-
def _gethostbyaddr(name):
# This is as close as I can get; at least the types are correct...
addresses = java.net.InetAddress.getAllByName(gethostbyname(name))
@@ -502,29 +495,11 @@
else:
return _udpsocket()
-def getaddrinfo(host, port, family=None, socktype=None, proto=None, flags=None):
- try:
- if not family in [AF_INET, AF_INET6, AF_UNSPEC]:
- raise NotSupportedError()
- filter_fns = []
- filter_fns.append({
- AF_INET: lambda x: isinstance(x, java.net.Inet4Address),
- AF_INET6: lambda x: isinstance(x, java.net.Inet6Address),
- AF_UNSPEC: lambda x: isinstance(x, java.net.InetAddress),
- }[family])
- # Cant see a way to support AI_PASSIVE right now.
- # if flags and flags & AI_PASSIVE:
- # pass
- results = []
- for a in java.net.InetAddress.getAllByName(host):
- if len([f for f in filter_fns if f(a)]):
- family = {java.net.Inet4Address: AF_INET, java.net.Inet6Address: AF_INET6}[a.class]
- # TODO: Include flowinfo and scopeid in a 4-tuple for IPv6 addresses
- results.append( (family, socktype, proto, a.getCanonicalHostName(), (a.getHostAddress(), port)) )
- return results
- except java.lang.Exception, jlx:
- raise _map_exception(jlx)
+def getaddrinfo(host, port, family=0, socktype=SOCK_STREAM, proto=0, flags=0):
+ return ( (AF_INET, socktype, 0, "", (gethostbyname(host), port)), )
+has_ipv6 = 1
+
def getnameinfo(sock_addr, flags):
raise NotImplementedError("getnameinfo not yet supported on jython.")
@@ -634,22 +609,13 @@
def _get_jsocket(self):
return self.sock_impl.jsocket
-def _unpack_address_tuple(address_tuple, for_tx=False):
- # TODO: Upgrade to support the 4-tuples used for IPv6 addresses
- # which include flowinfo and scope_id.
- # To be upgraded in synch with getaddrinfo
+def _unpack_address_tuple(address_tuple):
error_message = "Address must be a tuple of (hostname, port)"
if type(address_tuple) is not type( () ) \
or type(address_tuple[0]) is not type("") \
or type(address_tuple[1]) is not type(0):
raise TypeError(error_message)
- hostname = address_tuple[0].strip()
- if hostname == "<broadcast>":
- if for_tx:
- hostname = "255.255.255.255"
- else:
- hostname = "0.0.0.0"
- return hostname, address_tuple[1]
+ return address_tuple[0], address_tuple[1]
class _tcpsocket(_nonblocking_api_mixin):
@@ -666,7 +632,7 @@
assert not self.sock_impl
assert not self.local_addr
# Do the address format check
- _unpack_address_tuple(addr)
+ host, port = _unpack_address_tuple(addr)
self.local_addr = addr
def listen(self, backlog=50):
@@ -675,7 +641,7 @@
assert not self.sock_impl
self.server = 1
if self.local_addr:
- host, port = _unpack_address_tuple(self.local_addr)
+ host, port = self.local_addr
else:
host, port = "", 0
self.sock_impl = _server_socket_impl(host, port, backlog, self.pending_options[SO_REUSEADDR])
@@ -712,7 +678,7 @@
host, port = self._get_host_port(addr)
self.sock_impl = _client_socket_impl()
if self.local_addr: # Has the socket been bound to a local address?
- bind_host, bind_port = _unpack_address_tuple(self.local_addr)
+ bind_host, bind_port = self.local_addr
self.sock_impl.bind(bind_host, bind_port, self.pending_options[SO_REUSEADDR])
self._config() # Configure timeouts, etc, now that the socket exists
self.sock_impl.connect(host, port)
@@ -870,9 +836,8 @@
if not self.sock_impl:
self.sock_impl = _datagram_socket_impl()
self._config()
- host, port = _unpack_address_tuple(addr, True)
byte_array = java.lang.String(data).getBytes('iso-8859-1')
- result = self.sock_impl.sendto(byte_array, host, port, flags)
+ result = self.sock_impl.sendto(byte_array, addr, flags)
return result
except java.lang.Exception, jlx:
raise _map_exception(jlx)
Modified: branches/asm/Lib/test/test_socket.py
===================================================================
--- branches/asm/Lib/test/test_socket.py 2008-07-15 22:37:38 UTC (rev 4951)
+++ branches/asm/Lib/test/test_socket.py 2008-07-15 22:51:51 UTC (rev 4952)
@@ -801,21 +801,6 @@
self.cli.settimeout(10)
self.cli.sendto(EIGHT_BIT_MSG, 0, (HOST, PORT))
-class UDPBroadcastTest(ThreadedUDPSocketTest):
-
- def setUp(self):
- self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-
- def testBroadcast(self):
- self.serv.bind( ("<broadcast>", PORT) )
- msg = self.serv.recv(len(EIGHT_BIT_MSG))
- self.assertEqual(msg, EIGHT_BIT_MSG)
-
- def _testBroadcast(self):
- self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
- self.cli.sendto(EIGHT_BIT_MSG, ("<broadcast>", PORT) )
-
class BasicSocketPairTest(SocketPairTest):
def __init__(self, methodName='runTest'):
@@ -1464,10 +1449,6 @@
if sys.platform[:4] == 'java':
tests.append(TestJythonTCPExceptions)
tests.append(TestJythonUDPExceptions)
- # TODO: Broadcast requires permission, and is blocked by some firewalls
- # Need some way to discover the network setup on the test machine
- if False:
- tests.append(UDPBroadcastTest)
suites = [unittest.makeSuite(klass, 'test') for klass in tests]
test_support.run_suite(unittest.TestSuite(suites))
Deleted: branches/asm/Lib/test/test_socket_ipv6.py
===================================================================
--- branches/asm/Lib/test/test_socket_ipv6.py 2008-07-15 22:37:38 UTC (rev 4951)
+++ branches/asm/Lib/test/test_socket_ipv6.py 2008-07-15 22:51:51 UTC (rev 4952)
@@ -1,61 +0,0 @@
-"""
-AMAK: 20080714:
-This module contains IPv6 related tests.
-I have placed them in a separate module from the rest of the socket tests, because
-1. The tests pass on my IPv6 enabled windows box, on JVMs >= 1.5
-2. They don't pass on Ubuntu, on any JVM version. The equivalent java code to lookup
- IPv6 addresses doesn't work either, so I need to research how to configure Ubuntu
- so that java.net.Inet6Address.getAllByName() actually returns IPv6 addresses.
-3. I don't want to include these tests with the standard socket tests until
- the network status and config of the various test environments, e.g. Mac, BSD, Solaris, etc,
- are known.
-"""
-
-import unittest
-import test_support
-
-import socket
-
-class NameLookupTests(unittest.TestCase):
-
- def testLocalhostV4Lookup(self):
- results = socket.getaddrinfo("localhost", 80, socket.AF_INET, socket.SOCK_STREAM, 0)
- self.failUnlessEqual(len(results), 1)
- self.failUnlessEqual('127.0.0.1', results[0][4][0])
-
- def testRemoteV4Lookup(self):
- results = socket.getaddrinfo("www.python.org", 80, socket.AF_INET, socket.SOCK_STREAM, 0)
- self.failUnlessEqual(len(results), 1)
- self.failUnlessEqual('82.94.237.218', results[0][4][0])
-
- def testLocalhostV6Lookup(self):
- results = socket.getaddrinfo("localhost", 80, socket.AF_INET6, socket.SOCK_STREAM, 0)
- self.failUnlessEqual(len(results), 1)
- self.failUnlessEqual('0:0:0:0:0:0:0:1', results[0][4][0])
-
- def testRemoteV6Lookup(self):
- # This test relies on those nice Dutch folks at sixxs.org keeping the same IPv6 address for their gateway
- results = socket.getaddrinfo("www.python.org.sixxs.org", 80, socket.AF_INET6, socket.SOCK_STREAM, 0)
- self.failUnlessEqual(len(results), 1)
- self.failUnlessEqual('2001:838:2:1:2a0:24ff:feab:3b53', results[0][4][0])
-
- def testLocalhostV4AndV6Lookup(self):
- results = socket.getaddrinfo("localhost", 80, socket.AF_UNSPEC, socket.SOCK_STREAM, 0)
- self.failUnlessEqual(len(results), 2)
-
- def testRemoteV4AndV6Lookup(self):
- # Need a remote host with both IPv4 and IPv6 addresses; pass for now
- pass
-
- def testAI_PASSIVE(self):
- pass
-
-def test_main():
- tests = [
- NameLookupTests,
- ]
- suites = [unittest.makeSuite(klass, 'test') for klass in tests]
- test_support.run_suite(unittest.TestSuite(suites))
-
-if __name__ == "__main__":
- test_main()
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|