|
@@ -1196,30 +1196,31 @@ class TestUnixSockServer(unittest.TestCase):
|
|
|
# We test with UDP, as it can be "connected" without other
|
|
|
# endpoint. Note that in the current implementation _guess_remote()
|
|
|
# unconditionally returns SOCK_STREAM.
|
|
|
- with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
|
- sock.connect(('127.0.0.1', 12345))
|
|
|
- self.assertEqual((socket.AF_INET, socket.SOCK_STREAM,
|
|
|
- ('127.0.0.1', 12345)),
|
|
|
- self.unix._guess_remote(sock.fileno()))
|
|
|
-
|
|
|
+ sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
|
+ sock.connect(('127.0.0.1', 12345))
|
|
|
+ self.assertEqual((socket.AF_INET, socket.SOCK_STREAM,
|
|
|
+ ('127.0.0.1', 12345)),
|
|
|
+ self.unix._guess_remote(sock.fileno()))
|
|
|
+ sock.close()
|
|
|
if socket.has_ipv6:
|
|
|
# Don't check IPv6 address on hosts not supporting them
|
|
|
- with socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) as sock:
|
|
|
- sock.connect(('::1', 12345))
|
|
|
- self.assertEqual((socket.AF_INET6, socket.SOCK_STREAM,
|
|
|
- ('::1', 12345, 0, 0)),
|
|
|
- self.unix._guess_remote(sock.fileno()))
|
|
|
-
|
|
|
+ sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
|
|
|
+ sock.connect(('::1', 12345))
|
|
|
+ self.assertEqual((socket.AF_INET6, socket.SOCK_STREAM,
|
|
|
+ ('::1', 12345, 0, 0)),
|
|
|
+ self.unix._guess_remote(sock.fileno()))
|
|
|
+ sock.close()
|
|
|
# Try when pretending there's no IPv6 support
|
|
|
# (No need to pretend when there's really no IPv6)
|
|
|
xfrout.socket.has_ipv6 = False
|
|
|
- with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
|
- sock.connect(('127.0.0.1', 12345))
|
|
|
- self.assertEqual((socket.AF_INET, socket.SOCK_STREAM,
|
|
|
- ('127.0.0.1', 12345)),
|
|
|
- self.unix._guess_remote(sock.fileno()))
|
|
|
+ sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
|
+ sock.connect(('127.0.0.1', 12345))
|
|
|
+ self.assertEqual((socket.AF_INET, socket.SOCK_STREAM,
|
|
|
+ ('127.0.0.1', 12345)),
|
|
|
+ self.unix._guess_remote(sock.fileno()))
|
|
|
# Return it back
|
|
|
xfrout.socket.has_ipv6 = True
|
|
|
+ sock.close()
|
|
|
|
|
|
def test_receive_query_message(self):
|
|
|
send_msg = b"\xd6=\x00\x00\x00\x01\x00"
|