• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
9import logging, os, sys, threading, urllib, unittest
10
11from httpd import HTTPListener, SecureHTTPListener
12
13GET_TEST_PATH = '/get_test'
14
15def 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
31def 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
78def 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
93if __name__ == '__main__':
94    if len(sys.argv) > 1:
95        run_server()
96    else:
97        test()
98
99
100if __name__ == '__main__':
101    unittest.main()
102