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