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