This commit was manufactured by cvs2svn to create tag 'r212'.
[python/dscho.git] / Lib / test / test_socket.py
blob1880b115fde0d3ab4870a32d6c894daa6abea672
1 # Not tested:
2 # socket.fromfd()
3 # sktobj.getsockopt()
4 # sktobj.recvfrom()
5 # sktobj.sendto()
6 # sktobj.setblocking()
7 # sktobj.setsockopt()
8 # sktobj.shutdown()
11 from test_support import verbose, TestFailed
12 import socket
13 import os
14 import time
16 def missing_ok(str):
17 try:
18 getattr(socket, str)
19 except AttributeError:
20 pass
22 try: raise socket.error
23 except socket.error: print "socket.error"
25 socket.AF_INET
27 socket.SOCK_STREAM
28 socket.SOCK_DGRAM
29 socket.SOCK_RAW
30 socket.SOCK_RDM
31 socket.SOCK_SEQPACKET
33 for optional in ("AF_UNIX",
35 "SO_DEBUG", "SO_ACCEPTCONN", "SO_REUSEADDR", "SO_KEEPALIVE",
36 "SO_DONTROUTE", "SO_BROADCAST", "SO_USELOOPBACK", "SO_LINGER",
37 "SO_OOBINLINE", "SO_REUSEPORT", "SO_SNDBUF", "SO_RCVBUF",
38 "SO_SNDLOWAT", "SO_RCVLOWAT", "SO_SNDTIMEO", "SO_RCVTIMEO",
39 "SO_ERROR", "SO_TYPE", "SOMAXCONN",
41 "MSG_OOB", "MSG_PEEK", "MSG_DONTROUTE", "MSG_EOR",
42 "MSG_TRUNC", "MSG_CTRUNC", "MSG_WAITALL", "MSG_BTAG",
43 "MSG_ETAG",
45 "SOL_SOCKET",
47 "IPPROTO_IP", "IPPROTO_ICMP", "IPPROTO_IGMP",
48 "IPPROTO_GGP", "IPPROTO_TCP", "IPPROTO_EGP",
49 "IPPROTO_PUP", "IPPROTO_UDP", "IPPROTO_IDP",
50 "IPPROTO_HELLO", "IPPROTO_ND", "IPPROTO_TP",
51 "IPPROTO_XTP", "IPPROTO_EON", "IPPROTO_BIP",
52 "IPPROTO_RAW", "IPPROTO_MAX",
54 "IPPORT_RESERVED", "IPPORT_USERRESERVED",
56 "INADDR_ANY", "INADDR_BROADCAST", "INADDR_LOOPBACK",
57 "INADDR_UNSPEC_GROUP", "INADDR_ALLHOSTS_GROUP",
58 "INADDR_MAX_LOCAL_GROUP", "INADDR_NONE",
60 "IP_OPTIONS", "IP_HDRINCL", "IP_TOS", "IP_TTL",
61 "IP_RECVOPTS", "IP_RECVRETOPTS", "IP_RECVDSTADDR",
62 "IP_RETOPTS", "IP_MULTICAST_IF", "IP_MULTICAST_TTL",
63 "IP_MULTICAST_LOOP", "IP_ADD_MEMBERSHIP",
64 "IP_DROP_MEMBERSHIP",
66 missing_ok(optional)
68 socktype = socket.SocketType
69 hostname = socket.gethostname()
70 ip = socket.gethostbyname(hostname)
71 hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
72 all_host_names = [hname] + aliases
74 if verbose:
75 print hostname
76 print ip
77 print hname, aliases, ipaddrs
78 print all_host_names
80 for name in all_host_names:
81 if name.find('.'):
82 break
83 else:
84 print 'FQDN not found'
86 if hasattr(socket, 'getservbyname'):
87 print socket.getservbyname('telnet', 'tcp')
88 try:
89 socket.getservbyname('telnet', 'udp')
90 except socket.error:
91 pass
94 canfork = hasattr(os, 'fork')
95 try:
96 PORT = 50007
97 if not canfork or os.fork():
98 # parent is server
99 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
100 s.bind(("127.0.0.1", PORT))
101 s.listen(1)
102 if verbose:
103 print 'parent accepting'
104 if canfork:
105 conn, addr = s.accept()
106 if verbose:
107 print 'connected by', addr
108 # couple of interesting tests while we've got a live socket
109 f = conn.fileno()
110 if verbose:
111 print 'fileno:', f
112 p = conn.getpeername()
113 if verbose:
114 print 'peer:', p
115 n = conn.getsockname()
116 if verbose:
117 print 'sockname:', n
118 f = conn.makefile()
119 if verbose:
120 print 'file obj:', f
121 while 1:
122 data = conn.recv(1024)
123 if not data:
124 break
125 if verbose:
126 print 'received:', data
127 conn.sendall(data)
128 conn.close()
129 else:
130 try:
131 # child is client
132 time.sleep(5)
133 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
134 if verbose:
135 print 'child connecting'
136 s.connect(("127.0.0.1", PORT))
137 msg = 'socket test'
138 s.send(msg)
139 data = s.recv(1024)
140 if msg != data:
141 print 'parent/client mismatch'
142 s.close()
143 finally:
144 os._exit(1)
145 except socket.error, msg:
146 raise TestFailed, msg