tests/test-l7.py

   1 # Copyright (c) 2015, 2016 Nicira, Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import argparse
  16 import socket
  17
  18 from BaseHTTPServer import HTTPServer
  19 from SimpleHTTPServer import SimpleHTTPRequestHandler
  20 from SocketServer import TCPServer
  21
  22
  23 class TCPServerV6(HTTPServer):
  24     address_family = socket.AF_INET6
  25
  26
  27 def get_ftpd():
  28     try:
  29         from pyftpdlib.authorizers import DummyAuthorizer
  30         from pyftpdlib.handlers import FTPHandler
  31         from pyftpdlib.servers import FTPServer
  32
  33         import logging
  34         import pyftpdlib.log
  35         pyftpdlib.log.LEVEL = logging.DEBUG
  36
  37         class OVSFTPHandler(FTPHandler):
  38             authorizer = DummyAuthorizer()
  39             authorizer.add_anonymous("/tmp")
  40             # Hack around a bug in pyftpdlib, which rejects EPRT
  41             # connection due to mismatching textual representation of
  42             # the IPv6 address.
  43             permit_foreign_addresses = True
  44         server = [FTPServer, OVSFTPHandler, 21]
  45     except ImportError:
  46         server = None
  47         pass
  48     return server
  49
  50
  51 def get_tftpd():
  52     try:
  53         from tftpy import TftpServer, TftpShared
  54
  55         class OVSTFTPServer(TftpServer):
  56             def __init__(self, listen, handler=None):
  57                 (ip, port) = listen
  58                 self.ip = ip
  59                 self.port = port
  60                 TftpServer.__init__(self, tftproot='./')
  61
  62             def serve_forever(self):
  63                 self.listen(self.ip, self.port)
  64         server = [OVSTFTPServer, None, TftpShared.DEF_TFTP_PORT]
  65     except ImportError:
  66         server = None
  67         pass
  68     return server
  69
  70
  71 def main():
  72     SERVERS = {
  73         'http': [TCPServer, SimpleHTTPRequestHandler, 80],
  74         'http6': [TCPServerV6, SimpleHTTPRequestHandler, 80],
  75         'ftp': get_ftpd(),
  76         'tftp': get_tftpd(),
  77     }
  78
  79     protocols = [srv for srv in SERVERS if SERVERS[srv] is not None]
  80     parser = argparse.ArgumentParser(
  81             description='Run basic application servers.')
  82     parser.add_argument('proto', default='http', nargs='?',
  83             help='protocol to serve (%s)' % protocols)
  84     args = parser.parse_args()
  85
  86     if args.proto not in protocols:
  87         parser.print_help()
  88         exit(1)
  89
  90     constructor = SERVERS[args.proto][0]
  91     handler = SERVERS[args.proto][1]
  92     port = SERVERS[args.proto][2]
  93     srv = constructor(('', port), handler)
  94     srv.serve_forever()
  95
  96
  97 if __name__ == '__main__':
  98     main()