client/cros/httpd_unittest.py

#!/usr/bin/python

# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""HTTPlistener unittest."""

import logging, os, sys, threading, urllib

from httpd import HTTPListener, SecureHTTPListener

GET_TEST_PATH = '/get_test'

def run_get_test(test_server, url):
    err = 1
    get_done = test_server.add_wait_url(GET_TEST_PATH)
    get_resp = ''
    try:
        get_resp = urllib.urlopen(url).read()
    except IOError, e:
        pass
    if not (get_done.is_set() and get_resp):
        print 'FAILED'
    else:
        print 'PASSED'
        err = 0
    return err


def test():
    test_server = HTTPListener(8000, docroot='/tmp')
    post_done = test_server.add_wait_url("/post_test",
                                         matchParams={'test': 'passed'})
    def _Spam():
        while not post_done.is_set():
            print 'TEST: server running'
            post_done.wait()
        return
    test_server.run()
    t = threading.Thread(target=_Spam).start()
    params = urllib.urlencode({'test': 'passed'})
    err = 1

    # TODO(seano): This test doesn't seem to pass.
    post_resp = ''
    try:
        post_resp = urllib.urlopen('http://localhost:8000/post_test',
                                   params).read()
    except IOError, e:
        pass
    if not (post_done.is_set() and
            test_server.get_form_entries()['test'] != 'passed'):
        print 'FAILED'
    else:
        print 'PASSED'
        err = 0


    err = run_get_test(test_server, 'http://localhost:8000' + GET_TEST_PATH)
    test_server.stop()
    if err != 0:
        return err

    creds_path = (os.path.dirname(os.path.realpath( __file__)) +
                  '/httpd_unittest_server')
    ssl_port=50000
    test_server = SecureHTTPListener(port=ssl_port,
                                     cert_path=(creds_path+'.pem'),
                                     key_path=(creds_path+'.key'))
    test_server.run()
    err = run_get_test(test_server,
                       'https://localhost:%d%s' % (ssl_port, GET_TEST_PATH))
    test_server.stop()
    return err


def run_server():
    """Example method showing how to start a HTTPListener."""
    test_server = HTTPListener(8000, docroot='/tmp')
    latch = test_server.add_wait_url('/quitquitquit')
    test_server.run()
    logging.info('server started')
    while not latch.is_set():
        try:
            latch.wait(1)
        except KeyboardInterrupt:
            sys.exit()
    test_server.stop()
    return


if __name__ == '__main__':
    if len(sys.argv) > 1:
        run_server()
    else:
        test()