Home | History | Annotate | Download | only in cros 
      1 #!/usr/bin/python
      2 
      3 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
      4 # Use of this source code is governed by a BSD-style license that can be
      5 # found in the LICENSE file.
      6 
      7 """HTTPlistener unittest."""
      8 
      9 import logging, os, sys, threading, urllib, unittest
     10 
     11 from httpd import HTTPListener, SecureHTTPListener
     12 
     13 GET_TEST_PATH = '/get_test'
     14 
     15 def run_get_test(test_server, url):
     16     err = 1
     17     get_done = test_server.add_wait_url(GET_TEST_PATH)
     18     get_resp = ''
     19     try:
     20         get_resp = urllib.urlopen(url).read()
     21     except IOError, e:
     22         pass
     23     if not (get_done.is_set() and get_resp):
     24         print 'FAILED'
     25     else:
     26         print 'PASSED'
     27         err = 0
     28     return err
     29 
     30 
     31 def test():
     32     test_server = HTTPListener(8000, docroot='/tmp')
     33     post_done = test_server.add_wait_url("/post_test",
     34                                          matchParams={'test': 'passed'})
     35     def _Spam():
     36         while not post_done.is_set():
     37             print 'TEST: server running'
     38             post_done.wait()
     39         return
     40     test_server.run()
     41     t = threading.Thread(target=_Spam).start()
     42     params = urllib.urlencode({'test': 'passed'})
     43     err = 1
     44 
     45     # TODO(seano): This test doesn't seem to pass.
     46     post_resp = ''
     47     try:
     48         post_resp = urllib.urlopen('http://localhost:8000/post_test',
     49                                    params).read()
     50     except IOError, e:
     51         pass
     52     if not (post_done.is_set() and
     53             test_server.get_form_entries()['test'] != 'passed'):
     54         print 'FAILED'
     55     else:
     56         print 'PASSED'
     57         err = 0
     58 
     59 
     60     err = run_get_test(test_server, 'http://localhost:8000' + GET_TEST_PATH)
     61     test_server.stop()
     62     if err != 0:
     63         return err
     64 
     65     creds_path = (os.path.dirname(os.path.realpath( __file__)) +
     66                   '/httpd_unittest_server')
     67     ssl_port=50000
     68     test_server = SecureHTTPListener(port=ssl_port,
     69                                      cert_path=(creds_path+'.pem'),
     70                                      key_path=(creds_path+'.key'))
     71     test_server.run()
     72     err = run_get_test(test_server,
     73                        'https://localhost:%d%s' % (ssl_port, GET_TEST_PATH))
     74     test_server.stop()
     75     return err
     76 
     77 
     78 def run_server():
     79     """Example method showing how to start a HTTPListener."""
     80     test_server = HTTPListener(8000, docroot='/tmp')
     81     latch = test_server.add_wait_url('/quitquitquit')
     82     test_server.run()
     83     logging.info('server started')
     84     while not latch.is_set():
     85         try:
     86             latch.wait(1)
     87         except KeyboardInterrupt:
     88             sys.exit()
     89     test_server.stop()
     90     return
     91 
     92 
     93 if __name__ == '__main__':
     94     if len(sys.argv) > 1:
     95         run_server()
     96     else:
     97         test()
     98 
     99 
    100 if __name__ == '__main__':
    101     unittest.main()
    102