pbr/tests/test_wsgi.py

   1 # Copyright (c) 2015 Hewlett-Packard Development Company, L.P. (HP)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 # not use this file except in compliance with the License. You may obtain
   5 # a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 # License for the specific language governing permissions and limitations
  13 # under the License.
  14
  15 import os
  16 import re
  17 import subprocess
  18 import sys
  19 try:
  20     # python 2
  21     from urllib2 import urlopen
  22 except ImportError:
  23     # python 3
  24     from urllib.request import urlopen
  25
  26 from pbr.tests import base
  27
  28
  29 class TestWsgiScripts(base.BaseTestCase):
  30
  31     cmd_names = ('pbr_test_wsgi', 'pbr_test_wsgi_with_class')
  32
  33     def _get_path(self):
  34         if os.path.isdir("%s/lib64" % self.temp_dir):
  35             path = "%s/lib64" % self.temp_dir
  36         elif os.path.isdir("%s/lib" % self.temp_dir):
  37             path = "%s/lib" % self.temp_dir
  38         elif os.path.isdir("%s/site-packages" % self.temp_dir):
  39             return ".:%s/site-packages" % self.temp_dir
  40         else:
  41             raise Exception("Could not determine path for test")
  42         return ".:%s/python%s.%s/site-packages" % (
  43             path,
  44             sys.version_info[0],
  45             sys.version_info[1])
  46
  47     def test_wsgi_script_install(self):
  48         """Test that we install a non-pkg-resources wsgi script."""
  49         if os.name == 'nt':
  50             self.skipTest('Windows support is passthrough')
  51
  52         stdout, _, return_code = self.run_setup(
  53             'install', '--prefix=%s' % self.temp_dir)
  54
  55         self._check_wsgi_install_content(stdout)
  56
  57     def test_wsgi_script_run(self):
  58         """Test that we install a runnable wsgi script.
  59
  60         This test actually attempts to start and interact with the
  61         wsgi script in question to demonstrate that it's a working
  62         wsgi script using simple server.
  63
  64         """
  65         if os.name == 'nt':
  66             self.skipTest('Windows support is passthrough')
  67
  68         stdout, _, return_code = self.run_setup(
  69             'install', '--prefix=%s' % self.temp_dir)
  70
  71         self._check_wsgi_install_content(stdout)
  72
  73         # Live test run the scripts and see that they respond to wsgi
  74         # requests.
  75         for cmd_name in self.cmd_names:
  76             self._test_wsgi(cmd_name, b'Hello World')
  77
  78     def _test_wsgi(self, cmd_name, output, extra_args=None):
  79         cmd = os.path.join(self.temp_dir, 'bin', cmd_name)
  80         print("Running %s -p 0 -b 127.0.0.1" % cmd)
  81         popen_cmd = [cmd, '-p', '0', '-b', '127.0.0.1']
  82         if extra_args:
  83             popen_cmd.extend(extra_args)
  84
  85         env = {'PYTHONPATH': self._get_path()}
  86
  87         p = subprocess.Popen(popen_cmd, stdout=subprocess.PIPE,
  88                              stderr=subprocess.PIPE, cwd=self.temp_dir,
  89                              env=env)
  90         self.addCleanup(p.kill)
  91
  92         stdoutdata = p.stdout.readline()  # ****...
  93
  94         stdoutdata = p.stdout.readline()  # STARTING test server...
  95         self.assertIn(
  96             b"STARTING test server pbr_testpackage.wsgi",
  97             stdoutdata)
  98
  99         stdoutdata = p.stdout.readline()  # Available at ...
 100         print(stdoutdata)
 101         m = re.search(b'(http://[^:]+:\d+)/', stdoutdata)
 102         self.assertIsNotNone(m, "Regex failed to match on %s" % stdoutdata)
 103
 104         stdoutdata = p.stdout.readline()  # DANGER! ...
 105         self.assertIn(
 106             b"DANGER! For testing only, do not use in production",
 107             stdoutdata)
 108
 109         stdoutdata = p.stdout.readline()  # ***...
 110
 111         f = urlopen(m.group(1).decode('utf-8'))
 112         self.assertEqual(output, f.read())
 113
 114         # Request again so that the application can force stderr.flush(),
 115         # otherwise the log is buffered and the next readline() will hang.
 116         urlopen(m.group(1).decode('utf-8'))
 117
 118         stdoutdata = p.stderr.readline()
 119         # we should have logged an HTTP request, return code 200, that
 120         # returned the right amount of bytes
 121         status = '"GET / HTTP/1.1" 200 %d' % len(output)
 122         self.assertIn(status.encode('utf-8'), stdoutdata)
 123
 124     def _check_wsgi_install_content(self, install_stdout):
 125         for cmd_name in self.cmd_names:
 126             install_txt = 'Installing %s script to %s' % (cmd_name,
 127                                                           self.temp_dir)
 128             self.assertIn(install_txt, install_stdout)
 129
 130             cmd_filename = os.path.join(self.temp_dir, 'bin', cmd_name)
 131
 132             script_txt = open(cmd_filename, 'r').read()
 133             self.assertNotIn('pkg_resources', script_txt)
 134
 135             main_block = """if __name__ == "__main__":
 136     import argparse
 137     import socket
 138     import sys
 139     import wsgiref.simple_server as wss"""
 140
 141             if cmd_name == 'pbr_test_wsgi':
 142                 app_name = "main"
 143             else:
 144                 app_name = "WSGI.app"
 145
 146             starting_block = ("STARTING test server pbr_testpackage.wsgi."
 147                               "%s" % app_name)
 148
 149             else_block = """else:
 150     application = None"""
 151
 152             self.assertIn(main_block, script_txt)
 153             self.assertIn(starting_block, script_txt)
 154             self.assertIn(else_block, script_txt)
 155
 156     def test_with_argument(self):
 157         if os.name == 'nt':
 158             self.skipTest('Windows support is passthrough')
 159
 160         stdout, _, return_code = self.run_setup(
 161             'install', '--prefix=%s' % self.temp_dir)
 162
 163         self._test_wsgi('pbr_test_wsgi', b'Foo Bar', ["--", "-c", "Foo Bar"])