Basic coverage of migwsgi.

This PR does the principle things required to allow exercising the
central component responsible for glueing named MiG "functionality"
files to WSGI and have the result execute to completion under Python 3.

Included is a small tactical change to allow the structural force
recursive functions to iterate tuples preserving their type such that
subsequence output behaves correctly under Py3.

Author: albu-diku

mig/shared/base.py

@@ -542,8 +542,9 @@ def force_utf8_rec(input_obj, highlight=''):
     if isinstance(input_obj, dict):
         return {force_utf8_rec(i, highlight): force_utf8_rec(j, highlight) for (i, j) in
                 input_obj.items()}
-    elif isinstance(input_obj, list):
-        return [force_utf8_rec(i, highlight) for i in input_obj]
+    elif isinstance(input_obj, (list, tuple)):
+        thetype = type(input_obj)
+        return thetype(force_utf8_rec(i, highlight) for i in input_obj)
     elif is_unicode(input_obj):
         return force_utf8(input_obj, highlight)
     else:
@@ -570,8 +571,9 @@ def force_unicode_rec(input_obj, highlight=''):
     if isinstance(input_obj, dict):
         return {force_unicode_rec(i, highlight): force_unicode_rec(j, highlight) for (i, j) in
                 input_obj.items()}
-    elif isinstance(input_obj, list):
-        return [force_unicode_rec(i, highlight) for i in input_obj]
+    elif isinstance(input_obj, (list, tuple)):
+        thetype = type(input_obj)
+        return thetype(force_utf8_rec(i, highlight) for i in input_obj)
     elif not is_unicode(input_obj):
         return force_unicode(input_obj, highlight)
     else:

mig/shared/compat.py

@@ -46,6 +46,19 @@
 _TYPE_UNICODE = type(u"")
 
 
+if PY2:
+    class SimpleNamespace(dict):
+        """Bare minimum SimpleNamespace for Python 2."""
+
+        def __getattribute__(self, name):
+            if name == '__dict__':
+                return dict(**self)
+
+            return self[name]
+else:
+    from types import SimpleNamespace
+
+
 def _is_unicode(val):
     """Return boolean indicating if the value is a unicode string.
 

mig/wsgi-bin/migwsgi.py

@@ -34,7 +34,7 @@
 from mig.shared import returnvalues
 from mig.shared.bailout import bailout_helper, crash_helper, compact_string
 from mig.shared.base import requested_backend, allow_script, \
-    is_default_str_coding, force_default_str_coding_rec
+    is_default_str_coding, force_default_str_coding_rec, force_utf8
 from mig.shared.defaults import download_block_size, default_fs_coding
 from mig.shared.conf import get_configuration_object
 from mig.shared.objecttypes import get_object_type_info
@@ -50,7 +50,7 @@ def object_type_info(object_type):
 
 
 def stub(configuration, client_id, import_path, backend, user_arguments_dict,
-         environ):
+         environ, _import_module):
     """Run backend on behalf of client_id with supplied user_arguments_dict.
     I.e. import main from import_path and execute it with supplied arguments.
     """
@@ -88,7 +88,7 @@ def stub(configuration, client_id, import_path, backend, user_arguments_dict,
 
         # _logger.debug("import main from %r" % import_path)
         # NOTE: dynamic module loading to find corresponding main function
-        module_handle = importlib.import_module(import_path)
+        module_handle = _import_module(import_path)
         main = module_handle.main
     except Exception as err:
         _logger.error("%s could not import %r (%s): %s" %
@@ -184,7 +184,8 @@ def _flush_wsgi_errors():
         environ['wsgi.errors'].close()
 
 
-def application(environ, start_response):
+def application(environ, start_response, configuration=None,
+        _import_module=importlib.import_module, _set_os_environ=True):
     """MiG app called automatically by WSGI.
 
     *environ* is a dictionary populated by the server with CGI-like variables
@@ -235,14 +236,17 @@ def application(environ, start_response):
                                                              os_env_value))
 
     # Assign updated environ to LOCAL os.environ for the rest of this session
-    os.environ = environ
+    if _set_os_environ:
+        os.environ(environ)
 
     # NOTE: redirect stdout to stderr in python 2 only. It breaks logger in 3
     #       and stdout redirection apparently is already handled there.
     if sys.version_info[0] < 3:
         sys.stdout = sys.stderr
 
-    configuration = get_configuration_object()
+    if configuration is None:
+        configuration = get_configuration_object()
+
     _logger = configuration.logger
 
     # NOTE: replace default wsgi errors to apache error log with our own logs
@@ -313,7 +317,8 @@ def application(environ, start_response):
             # _logger.debug("wsgi handling script: %s" % script_name)
             (output_objs, ret_val) = stub(configuration, client_id,
                                           module_path, backend,
-                                          user_arguments_dict, environ)
+                                          user_arguments_dict, environ,
+                                          _import_module=_import_module)
         else:
             _logger.warning("wsgi handling refused script:%s" % script_name)
             (output_objs, ret_val) = reject_main(client_id,
@@ -410,12 +415,12 @@ def application(environ, start_response):
             _logger.info("WSGI %s yielding %d output parts (%db)" %
                          (backend, chunk_parts, content_length))
         # _logger.debug("send chunked %r response to client" % backend)
-        for i in xrange(chunk_parts):
+        for i in list(range(chunk_parts)):
             # _logger.debug("WSGI %s yielding part %d / %d output parts" %
             #              (backend, i+1, chunk_parts))
             # end index may be after end of content - but no problem
             part = output[i*download_block_size:(i+1)*download_block_size]
-            yield part
+            yield force_utf8(part)
         if chunk_parts > 1:
             _logger.info("WSGI %s finished yielding all %d output parts" %
                          (backend, chunk_parts))

tests/support/htmlsupp.py

@@ -0,0 +1,84 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# htmlsupp - test support library for HTML
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+# -- END_HEADER ---
+#
+
+"""Test support library for HTML."""
+
+
+class HtmlAssertMixin:
+    """Custom assertions for HTML containing strings."""
+
+    def assertHtmlElement(self, value, tag_name):
+        """Check that an occurrence of the specifid tag within an HTML input
+        string can be found. Returns the textual content of the first match.
+        """
+
+        self.assertIsValidHtmlDocument(value)
+
+        # TODO: this is a definitively stop-gap way of finding a tag within the HTML
+        #       and is used purely to keep this initial change to a reasonable size.
+
+        tag_open = ''.join(['<', tag_name, '>'])
+        tag_open_index = value.index(tag_open)
+        tag_open_index_after = tag_open_index + len(tag_open)
+
+        tag_close = ''.join(['</', tag_name, '>'])
+        tag_close_index = value.index(tag_close, tag_open_index_after)
+
+        return value[tag_open_index_after:tag_close_index]
+
+    def assertHtmlElementTextContent(self, value, tag_name, expected_text, trim_newlines=True):
+        """Check there is an occurrence of a tag within an HTML input string
+        and check the text it encloses equals exactly the expecatation.
+        """
+
+        self.assertIsValidHtmlDocument(value)
+
+        # TODO: this is a definitively stop-gap way of finding a tag within the HTML
+        #       and is used purely to keep this initial change to a reasonable size.
+
+        actual_text = self.assertHtmlElement(value, tag_name)
+        if trim_newlines:
+            actual_text = actual_text.strip('\n')
+        self.assertEqual(actual_text, expected_text)
+
+    def assertIsValidHtmlDocument(self, value):
+        """Check that the input string contains a valid HTML document.
+        """
+
+        assert isinstance(value, type(u"")), "input string was not utf8"
+
+        error = None
+        try:
+            has_doctype = value.startswith("<!DOCTYPE html")
+            assert has_doctype, "no valid document opener"
+            end_html_tag_idx = value.rfind('</html>')
+            maybe_document_end = value[end_html_tag_idx:].rstrip()
+            assert maybe_document_end == '</html>', "no valid document closer"
+        except Exception as exc:
+            error = exc
+        if error:
+            raise AssertionError("failed to verify input string as HTML: %s", str(error))

tests/test_mig_wsgi-bin.py

@@ -0,0 +1,139 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+#
+# --- END_HEADER ---
+#
+
+"""Unit tests for the MiG WSGI glue."""
+
+import codecs
+from configparser import ConfigParser
+import importlib
+import os
+import stat
+import sys
+
+from tests.support import PY2, MIG_BASE, MigTestCase, testmain, is_path_within
+from tests.support.htmlsupp import HtmlAssertMixin
+from tests.support.wsgisupp import prepare_wsgi, WsgiAssertMixin
+
+from mig.shared.base import client_id_dir, client_dir_id, get_short_id, \
+    invisible_path, allow_script, brief_list
+from mig.shared.compat import SimpleNamespace
+import mig.shared.returnvalues as returnvalues
+
+# workaround for files within non-module directories
+
+
+def _import_forcibly(module_name, relative_module_dir=None):
+    module_path = os.path.join(MIG_BASE, 'mig')
+    if relative_module_dir is not None:
+        module_path = os.path.join(module_path, relative_module_dir)
+    sys.path.append(module_path)
+    mod = importlib.import_module(module_name)
+    sys.path.pop(-1)  # do not leave the forced module path
+    return mod
+
+
+migwsgi = _import_forcibly('migwsgi', relative_module_dir='wsgi-bin')
+
+
+def noop(*args):
+    pass
+
+
+class FakeBackend:
+    def __init__(self):
+        self.output_objects = [
+            { 'object_type': 'start' },
+            { 'object_type': 'title', 'text': 'ERROR' },
+        ]
+        self.return_value = returnvalues.ERROR
+
+    def main(self, client_id, user_arguments_dict):
+        return self.output_objects, self.return_value
+
+    def set_response(self, output_objects, returnvalue):
+        self.output_objects = output_objects
+        self.return_value = returnvalue
+
+    def to_import_module(self):
+        def _import_module(module_path):
+            return self
+        return _import_module
+
+
+class MigWsgibin(MigTestCase, HtmlAssertMixin, WsgiAssertMixin):
+
+    def _provide_configuration(self):
+        return 'testconfig'
+
+    def before_each(self):
+        self.fake_backend = FakeBackend()
+        self.fake_wsgi = prepare_wsgi(self.configuration, 'http://localhost/')
+
+        self.application_args = (
+            self.fake_wsgi.environ,
+            self.fake_wsgi.start_response,
+        )
+        self.application_kwargs = dict(
+            configuration=self.configuration,
+            _import_module=self.fake_backend.to_import_module(),
+            _set_os_environ=False,
+        )
+
+    def test_return_value_ok_returns_status_200(self):
+        wsgi_result = migwsgi.application(
+            *self.application_args,
+            **self.application_kwargs
+        )
+
+        self.assertWsgiResponse(wsgi_result, self.fake_wsgi, 200)
+
+    def test_return_value_ok_returns_valid_html_page(self):
+        wsgi_result = migwsgi.application(
+            *self.application_args,
+            **self.application_kwargs
+        )
+
+        output, _ = self.assertWsgiResponse(wsgi_result, self.fake_wsgi, 200)
+        self.assertIsValidHtmlDocument(output)
+
+    def test_return_value_ok_returns_expected_title(self):
+        output_objects = [
+            { 'object_type': 'title', 'text': 'TEST' }
+        ]
+        self.fake_backend.set_response(output_objects, returnvalues.OK)
+
+        wsgi_result = migwsgi.application(
+            *self.application_args,
+            **self.application_kwargs
+        )
+
+        output, _ = self.assertWsgiResponse(wsgi_result, self.fake_wsgi, 200)
+        self.assertHtmlElementTextContent(
+            output, 'title', 'TEST', trim_newlines=True)
+
+
+if __name__ == '__main__':
+    testmain()