src/native_client_sdk/src/tools/tests/httpd_test.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 import os
   7 import Queue
   8 import sys
   9 import subprocess
  10 import threading
  11 import unittest
  12 import urllib2
  13
  14 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
  15 PARENT_DIR = os.path.dirname(SCRIPT_DIR)
  16
  17 sys.path.append(PARENT_DIR)
  18
  19 import httpd
  20
  21
  22 class HTTPDTest(unittest.TestCase):
  23   def setUp(self):
  24     self.server = httpd.LocalHTTPServer('.', 0, False)
  25
  26   def tearDown(self):
  27     self.server.Shutdown()
  28
  29   def testQuit(self):
  30     urllib2.urlopen(self.server.GetURL('?quit=1'))
  31     self.server.process.join(10)  # Wait 10 seconds for the process to finish.
  32     self.assertFalse(self.server.process.is_alive())
  33
  34
  35 class RunTest(unittest.TestCase):
  36   def setUp(self):
  37     self.process = None
  38
  39   def tearDown(self):
  40     if self.process and self.process.returncode is None:
  41       self.process.kill()
  42
  43   @staticmethod
  44   def _SubprocessThread(process, queue):
  45     stdout, stderr = process.communicate()
  46     queue.put((process.returncode, stdout, stderr))
  47
  48   def _Run(self, args=None, timeout=None):
  49     args = args or []
  50     run_py = os.path.join(PARENT_DIR, 'run.py')
  51     cmd = [sys.executable, run_py]
  52     cmd.extend(args)
  53     self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
  54                                     stderr=subprocess.PIPE)
  55     queue = Queue.Queue()
  56     thread = threading.Thread(target=RunTest._SubprocessThread,
  57                               args=(self.process, queue))
  58     thread.daemon = True
  59     thread.start()
  60     thread.join(timeout)
  61     if not thread.is_alive():
  62       returncode, stdout, stderr = queue.get(False)
  63       return returncode, stdout, stderr
  64
  65     return -1, None, None
  66
  67   def _GetChromeMockArgs(self, page, http_request_type, sleep,
  68                          expect_to_be_killed=True):
  69     args = ['--test-mode']
  70     if page:
  71       args.extend(['-P', page])
  72     args.append('--')
  73     args.extend([sys.executable, os.path.join(SCRIPT_DIR, 'chrome_mock.py')])
  74     if http_request_type:
  75       args.append('--' + http_request_type)
  76     if sleep:
  77       args.extend(['--sleep', str(sleep)])
  78     if expect_to_be_killed:
  79       args.append('--expect-to-be-killed')
  80     return args
  81
  82   def testQuit(self):
  83     args = self._GetChromeMockArgs('?quit=1', 'get', sleep=10)
  84     _, stdout, _ = self._Run(args, timeout=20)
  85     self.assertTrue('Starting' in stdout)
  86     self.assertTrue('Expected to be killed' not in stdout)
  87
  88   def testSubprocessDies(self):
  89     args = self._GetChromeMockArgs(page=None, http_request_type=None, sleep=0,
  90                                    expect_to_be_killed=False)
  91     returncode, stdout, _ = self._Run(args, timeout=10)
  92     self.assertNotEqual(-1, returncode)
  93     self.assertTrue('Starting' in stdout)
  94
  95   def testPostOk(self):
  96     args = self._GetChromeMockArgs('ok', 'post', sleep=10)
  97     returncode, stdout, _ = self._Run(args, timeout=20)
  98     self.assertEqual(0, returncode)
  99     self.assertTrue('Starting' in stdout)
 100     self.assertTrue('Expected to be killed' not in stdout)
 101
 102   def testPostFail(self):
 103     args = self._GetChromeMockArgs('fail', 'post', sleep=10)
 104     returncode, stdout, _ = self._Run(args, timeout=20)
 105     self.assertEqual(1, returncode)
 106     self.assertTrue('Starting' in stdout)
 107     self.assertTrue('Expected to be killed' not in stdout)
 108
 109
 110 if __name__ == '__main__':
 111   unittest.main()