Lib/test/test_socket.py

   1 # Not tested:
   2 #       socket.fromfd()
   3 #       sktobj.getsockopt()
   4 #       sktobj.recvfrom()
   5 #       sktobj.sendto()
   6 #       sktobj.setblocking()
   7 #       sktobj.setsockopt()
   8 #       sktobj.shutdown()
   9
  10
  11 from test_support import verbose, TestFailed
  12 import socket
  13 import os
  14 import time
  15
  16 def missing_ok(str):
  17     try:
  18         getattr(socket, str)
  19     except AttributeError:
  20         pass
  21
  22 try: raise socket.error
  23 except socket.error: print "socket.error"
  24
  25 socket.AF_INET
  26
  27 socket.SOCK_STREAM
  28 socket.SOCK_DGRAM
  29 socket.SOCK_RAW
  30 socket.SOCK_RDM
  31 socket.SOCK_SEQPACKET
  32
  33 for optional in ("AF_UNIX",
  34
  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",
  40
  41                  "MSG_OOB", "MSG_PEEK", "MSG_DONTROUTE", "MSG_EOR",
  42                  "MSG_TRUNC", "MSG_CTRUNC", "MSG_WAITALL", "MSG_BTAG",
  43                  "MSG_ETAG",
  44
  45                  "SOL_SOCKET",
  46
  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",
  53
  54                  "IPPORT_RESERVED", "IPPORT_USERRESERVED",
  55
  56                  "INADDR_ANY", "INADDR_BROADCAST", "INADDR_LOOPBACK",
  57                  "INADDR_UNSPEC_GROUP", "INADDR_ALLHOSTS_GROUP",
  58                  "INADDR_MAX_LOCAL_GROUP", "INADDR_NONE",
  59
  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",
  65                  ):
  66     missing_ok(optional)
  67
  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
  73
  74 if verbose:
  75     print hostname
  76     print ip
  77     print hname, aliases, ipaddrs
  78     print all_host_names
  79
  80 for name in all_host_names:
  81     if name.find('.'):
  82         break
  83 else:
  84     print 'FQDN not found'
  85
  86 if hasattr(socket, 'getservbyname'):
  87     print socket.getservbyname('telnet', 'tcp')
  88     try:
  89         socket.getservbyname('telnet', 'udp')
  90     except socket.error:
  91         pass
  92
  93 import sys
  94 if not sys.platform.startswith('java'):
  95     try:
  96         # On some versions, this loses a reference
  97         orig = sys.getrefcount(__name__)
  98         socket.getnameinfo(__name__,0)
  99     except SystemError:
 100         if sys.getrefcount(__name__) <> orig:
 101             raise TestFailed,"socket.getnameinfo loses a reference"
 102
 103 try:
 104     # On some versions, this crashes the interpreter.
 105     socket.getnameinfo(('x', 0, 0, 0), 0)
 106 except socket.error:
 107     pass
 108
 109 canfork = hasattr(os, 'fork')
 110 try:
 111     PORT = 50007
 112     if not canfork or os.fork():
 113         # parent is server
 114         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 115         s.bind(("127.0.0.1", PORT))
 116         s.listen(1)
 117         if verbose:
 118             print 'parent accepting'
 119         if canfork:
 120             conn, addr = s.accept()
 121             if verbose:
 122                 print 'connected by', addr
 123             # couple of interesting tests while we've got a live socket
 124             f = conn.fileno()
 125             if verbose:
 126                 print 'fileno:', f
 127             p = conn.getpeername()
 128             if verbose:
 129                 print 'peer:', p
 130             n = conn.getsockname()
 131             if verbose:
 132                 print 'sockname:', n
 133             f = conn.makefile()
 134             if verbose:
 135                 print 'file obj:', f
 136             while 1:
 137                 data = conn.recv(1024)
 138                 if not data:
 139                     break
 140                 if verbose:
 141                     print 'received:', data
 142                 conn.sendall(data)
 143             conn.close()
 144     else:
 145         try:
 146             # child is client
 147             time.sleep(5)
 148             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 149             if verbose:
 150                 print 'child connecting'
 151             s.connect(("127.0.0.1", PORT))
 152             msg = 'socket test'
 153             s.send(msg)
 154             data = s.recv(1024)
 155             if msg != data:
 156                 print 'parent/client mismatch'
 157             s.close()
 158         finally:
 159             os._exit(1)
 160 except socket.error, msg:
 161     raise TestFailed, msg