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
