diff --git a/mig/shared/functional.py b/mig/shared/functional.py index 23706ffb5..2dcff965e 100644 --- a/mig/shared/functional.py +++ b/mig/shared/functional.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # functional - functionality backend helpers -# Copyright (C) 2003-2023 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -33,21 +34,21 @@ from past.builtins import basestring import os -import time - -# REJECT_UNSET is not used directly but exposed to functionality from mig.shared.accountstate import check_account_status, \ check_update_account_expire from mig.shared.base import requested_backend, force_native_str, get_site_base_url from mig.shared.defaults import csrf_field, auth_openid_ext_db from mig.shared.findtype import is_user -from mig.shared.httpsclient import extract_client_cert, extract_client_openid, \ +from mig.shared.httpsclient import extract_client_openid, \ extract_base_url from mig.shared.init import find_entry, make_title_entry, make_header_entry from mig.shared.safeinput import validated_input, REJECT_UNSET from mig.shared.useradm import expire_oid_sessions +# REJECT_UNSET is not used directly but exposed to functionality +__exports = [REJECT_UNSET] + def warn_on_rejects(rejects, output_objects): """Helper to fill in output_objects in case of rejects""" @@ -89,6 +90,7 @@ def validate_input( output_objects, allow_rejects, prefilter_map=None, + environ=None, typecheck_overrides={} ): """A wrapper used by most back end functionality. @@ -154,7 +156,6 @@ def validate_input_and_cert( creds_error = '' pending_expire, account_expire = True, 0 account_accessible, account_status = True, 'active' - user_dict = None if not client_id: creds_error = "Invalid or missing user credentials" elif not is_user(client_id, configuration): @@ -176,7 +177,7 @@ def validate_input_and_cert( # Expired users can still log out or use their login to access the # (unprivileged) account request pages to renew their account with # auto-fill of fields. - if creds_error and not requested_backend(environ) in \ + if creds_error and requested_backend(environ) not in \ ['logout', 'autologout', 'reqoid', 'reqcert', 'extcert']: # Simple init to get page preamble even where initialize_main_variables # was called with most things disabled because no or limited direct @@ -282,7 +283,8 @@ def validate_input_and_cert( (status, retval) = validate_input(user_arguments_dict, defaults, output_objects, allow_rejects, - filter_values, + prefilter_map=filter_values, + environ=environ, typecheck_overrides=typecheck_overrides) return (status, retval) diff --git a/mig/shared/functionality/cat.py b/mig/shared/functionality/cat.py index 5116ed0b0..798ff55b2 100755 --- a/mig/shared/functionality/cat.py +++ b/mig/shared/functionality/cat.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # cat - show lines of one or more files -# Copyright (C) 2003-2024 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -41,8 +42,8 @@ write_file_lines from mig.shared.functional import validate_input_and_cert, REJECT_UNSET from mig.shared.handlers import safe_handler, get_csrf_limit -from mig.shared.init import initialize_main_variables, find_entry, \ - make_start_entry, start_error, start_download +from mig.shared.init import find_entry, lazy_init_backend, start_download, \ + start_error from mig.shared.parseflags import verbose, binary from mig.shared.userio import GDPIOLogError, gdp_iolog from mig.shared.safeinput import valid_path_pattern @@ -90,32 +91,12 @@ def signature(): return ['file_output', defaults] -def main(client_id, user_arguments_dict, environ=None): +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs=None): """Main function wrapper used by front end""" - if environ is None: - environ = os.environ - - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id) - - return _main(configuration, logger, environ, op_name=op_name, - output_objects=output_objects, client_id=client_id, - user_arguments_dict=user_arguments_dict) - - -def _main(configuration, logger, environ, op_name='', output_objects=None, client_id=None, - user_arguments_dict=None): - """Actual main function to generate contents for the front end""" - - assert environ is not None, "required arg: environ" - - if logger is None: - logger = configuration.logger - - # Create new output_objects list with start entry if None was supplied - if output_objects is None: - output_objects = [make_start_entry()] + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) client_dir = client_id_dir(client_id) defaults = signature()[1] diff --git a/mig/shared/functionality/datatransfer.py b/mig/shared/functionality/datatransfer.py index 713f4b7a8..0b43343e9 100755 --- a/mig/shared/functionality/datatransfer.py +++ b/mig/shared/functionality/datatransfer.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # datatransfer - import and export data in the backgroud -# Copyright (C) 2003-2023 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -34,13 +35,13 @@ import time from mig.shared import returnvalues -from mig.shared.base import client_id_dir, mask_creds, hexlify, requested_backend +from mig.shared.base import client_id_dir, mask_creds, hexlify from mig.shared.defaults import default_pager_entries, csrf_field, protocol_aliases from mig.shared.fileio import read_tail_lines from mig.shared.functional import validate_input_and_cert from mig.shared.handlers import safe_handler, get_csrf_limit, make_csrf_token from mig.shared.htmlgen import man_base_js, man_base_html, html_post_helper -from mig.shared.init import initialize_main_variables, find_entry, make_title_entry, make_start_entry +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.parseflags import quiet from mig.shared.pwcrypto import make_digest, make_encrypt from mig.shared.transferfunctions import build_transferitem_object, \ @@ -56,8 +57,8 @@ get_actions = ['show', 'fillimport', 'fillexport'] transfer_actions = ['import', 'export', 'deltransfer', 'redotransfer'] # TODO: add these internal data shuffling targets on a separate tab without -#address and creds -#shuffling_actions = ['move', 'copy', 'unpack', 'pack', 'remove'] +# address and creds +# shuffling_actions = ['move', 'copy', 'unpack', 'pack', 'remove'] shuffling_actions = [] key_actions = ['generatekey', 'delkey'] post_actions = transfer_actions + shuffling_actions + key_actions @@ -88,35 +89,12 @@ def signature(): return ['text', defaults] -def main(client_id, user_arguments_dict, environ=None): +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs=None): """Main function wrapper used by front end""" - if environ is None: - environ = os.environ - - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id) - - return _main(configuration, logger, environ, op_name=op_name, - output_objects=output_objects, client_id=client_id, - user_arguments_dict=user_arguments_dict) - - -def _main(configuration, logger, environ, op_name='', output_objects=None, client_id=None, - user_arguments_dict=None): - """Actual main function to generate contents for the front end""" - - assert environ is not None, "required arg: environ" - - if logger is None: - logger = configuration.logger - - # Create new output_objects list with start entry if None was supplied - if output_objects is None: - output_objects = [make_start_entry()] - if not op_name: - op_name = requested_backend() - output_objects.append(make_title_entry('%s' % op_name)) + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) defaults = signature()[1] (validate_status, accepted) = validate_input_and_cert( @@ -836,7 +814,7 @@ def _main(configuration, logger, environ, op_name='', output_objects=None, clien try: password_encrypted = make_encrypt(configuration, password) password_digest = '' - except: + except Exception: password_encrypted = '' password_digest = make_digest( 'datatransfer', client_id, password, diff --git a/mig/shared/functionality/docs.py b/mig/shared/functionality/docs.py index 05078ce73..cdbca012f 100755 --- a/mig/shared/functionality/docs.py +++ b/mig/shared/functionality/docs.py @@ -40,8 +40,9 @@ from mig.shared import mrslkeywords from mig.shared import resconfkeywords from mig.shared import returnvalues +from mig.shared.defaults import keyword_auto from mig.shared.functional import validate_input -from mig.shared.init import initialize_main_variables +from mig.shared.init import lazy_init_backend from mig.shared.output import get_valid_outputformats @@ -606,18 +607,20 @@ def license_information(configuration, output_objects): 'text': 'sshfs client (GNU v2.0)'}) -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_menu': keyword_auto}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False, - op_menu=client_id) defaults = signature()[1] (validate_status, accepted) = validate_input( user_arguments_dict, defaults, output_objects, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) diff --git a/mig/shared/functionality/fileman.py b/mig/shared/functionality/fileman.py index 3bff7c02d..7ad71706b 100644 --- a/mig/shared/functionality/fileman.py +++ b/mig/shared/functionality/fileman.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # fileman - File manager UI for browsing and manipulating files and folders -# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -31,10 +32,8 @@ from __future__ import absolute_import -import sys from mig.shared import returnvalues -from mig.shared.base import client_id_dir from mig.shared.defaults import trash_linkname, csrf_backends, csrf_field, \ default_max_chunks from mig.shared.freezefunctions import import_freeze_form @@ -44,7 +43,7 @@ from mig.shared.gdp.all import get_project_from_client_id from mig.shared.handlers import get_csrf_limit, make_csrf_token from mig.shared.htmlgen import themed_styles, legacy_user_interface -from mig.shared.init import initialize_main_variables, find_entry, extract_menu +from mig.shared.init import extract_menu, find_entry, lazy_init_backend from mig.shared.pwcrypto import sorted_hash_algos, default_algo from mig.shared.sharelinks import create_share_link_form, import_share_link_form @@ -463,7 +462,7 @@ def js_tmpl_parts(configuration, ('%s' % (configuration.site_enable_transfers and legacy_buttons)).lower(), 'enable_gdp': ('%s' % configuration.site_enable_gdp).lower(), - 'max_stream_size': 64*1024*1024 + 'max_stream_size': 64 * 1024 * 1024 } js_import = ''' @@ -590,7 +589,7 @@ def js_tmpl_parts(configuration, /* jquery-ui-1.7.x option format */ $.ui.dialog.defaults.bgiframe = true; } - ''' % fill_entries + ''' # no use for fill_entries here js_ready = ''' /* wrap in try/catch for debugging - disabled in prodution */ /* @@ -661,12 +660,13 @@ def signature(): return ['', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) - client_dir = client_id_dir(client_id) defaults = signature()[1] (validate_status, accepted) = validate_input_and_cert( user_arguments_dict, @@ -675,6 +675,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, # NOTE: path cannot use wildcards here typecheck_overrides={}, ) diff --git a/mig/shared/functionality/freezedb.py b/mig/shared/functionality/freezedb.py index 7741f8e00..8fb9403a7 100755 --- a/mig/shared/functionality/freezedb.py +++ b/mig/shared/functionality/freezedb.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # freezedb - manage frozen archives -# Copyright (C) 2003-2023 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,23 +20,25 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # """Manage all owned frozen archives""" + from __future__ import absolute_import from mig.shared import returnvalues from mig.shared.defaults import default_pager_entries, csrf_field, keyword_final from mig.shared.freezefunctions import build_freezeitem_object, \ - list_frozen_archives, get_frozen_meta, get_frozen_archive, \ + list_frozen_archives, get_frozen_archive, \ pending_archives_update, TARGET_ARCHIVE from mig.shared.functional import validate_input_and_cert from mig.shared.handlers import get_csrf_limit, make_csrf_token from mig.shared.htmlgen import man_base_js, man_base_html, html_post_helper -from mig.shared.init import initialize_main_variables, find_entry +from mig.shared.init import find_entry, lazy_init_backend list_operations = ['showlist', 'list'] show_operations = ['show', 'showlist'] @@ -51,11 +53,13 @@ def signature(): return ['frozenarchives', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) defaults = signature()[1] title_entry = find_entry(output_objects, 'title') title_entry['text'] = 'Frozen Archives' @@ -66,6 +70,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) @@ -80,7 +85,7 @@ def main(client_id, user_arguments_dict): """ % (configuration.short_title, configuration.support_email)}) return (output_objects, returnvalues.OK) - if not operation in allowed_operations: + if operation not in allowed_operations: output_objects.append({'object_type': 'text', 'text': '''Operation must be one of %s.''' % ', '.join(allowed_operations)}) diff --git a/mig/shared/functionality/home.py b/mig/shared/functionality/home.py index 9eb1f8c17..40b3d9cdc 100755 --- a/mig/shared/functionality/home.py +++ b/mig/shared/functionality/home.py @@ -20,23 +20,21 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # - """Home page generator with dynamic app selection""" from __future__ import absolute_import -import os - from mig.shared import returnvalues from mig.shared.defaults import csrf_field, user_home_label from mig.shared.findtype import is_admin from mig.shared.functional import validate_input_and_cert -from mig.shared.init import initialize_main_variables, find_entry, extract_menu +from mig.shared.init import extract_menu, find_entry, lazy_init_backend from mig.shared.handlers import get_csrf_limit, make_csrf_token from mig.shared.htmlgen import save_settings_js, save_settings_html, render_apps, \ menu_items, legacy_user_interface, html_user_messages @@ -112,8 +110,7 @@ def html_tmpl(configuration, client_id, title_entry, csrf_map={}, chroot=''): apps_field = 'SITE_USER_MENU' # NOTE: build list of all default and user selectable apps in that order app_list = [app_id for app_id in configuration.site_default_menu] - app_list += [app_id for app_id in configuration.site_user_menu if not - app_id in app_list] + app_list += [app_id for app_id in configuration.site_user_menu if app_id not in app_list] mandatory_apps = [] for app_name in configuration.site_default_menu: @@ -221,12 +218,13 @@ def signature(): return ['text', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False, - op_menu=client_id) defaults = signature()[1] (validate_status, accepted) = validate_input_and_cert( user_arguments_dict, @@ -235,6 +233,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) diff --git a/mig/shared/functionality/jupyter.py b/mig/shared/functionality/jupyter.py index fcbdfeb9c..ee7b52d33 100755 --- a/mig/shared/functionality/jupyter.py +++ b/mig/shared/functionality/jupyter.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # jupyter - User menu over the available jupyter services -# Copyright (C) 2003-2019 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -30,13 +31,13 @@ provides a list of buttons based on services defined in the configuration.jupyter_services """ + from __future__ import absolute_import from mig.shared import returnvalues - -from mig.shared.init import find_entry, initialize_main_variables from mig.shared.functional import validate_input_and_cert from mig.shared.htmlgen import man_base_js +from mig.shared.init import find_entry, lazy_init_backend def signature(): @@ -46,10 +47,13 @@ def signature(): return ['jupyter', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) + defaults = signature()[1] (validate_status, accepted) = validate_input_and_cert( user_arguments_dict, @@ -58,12 +62,13 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) - logger.debug("User: %s executing %s", client_id, op_name) + logger.debug("User: %s executing %s" % (client_id, op_name)) if not configuration.site_enable_jupyter: output_objects.append( {'object_type': 'error_text', 'text': diff --git a/mig/shared/functionality/reqcert.py b/mig/shared/functionality/reqcert.py index bfbb68b27..f2a1ca5f9 100644 --- a/mig/shared/functionality/reqcert.py +++ b/mig/shared/functionality/reqcert.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # reqcert - Local certificate request and account sign up backend -# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -40,7 +41,7 @@ from mig.shared.defaults import csrf_field, keyword_auto from mig.shared.functional import validate_input from mig.shared.handlers import get_csrf_limit, make_csrf_token -from mig.shared.init import initialize_main_variables, find_entry +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.pwcrypto import parse_password_policy from mig.shared.safeinput import html_escape @@ -63,20 +64,23 @@ def signature(configuration): return ['html_form', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False, 'op_menu': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False, op_menu=False) client_dir = client_id_dir(client_id) defaults = signature(configuration)[1] (validate_status, accepted) = validate_input(user_arguments_dict, defaults, output_objects, - allow_rejects=False) + allow_rejects=False, + environ=environ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) - if not 'migcert' in configuration.site_signup_methods: + if 'migcert' not in configuration.site_signup_methods: output_objects.append( {'object_type': 'error_text', 'text': '''X.509 certificate login is not enabled on this site'''}) @@ -159,7 +163,7 @@ def main(client_id, user_arguments_dict): # Override with arg values if set for field in user_fields: - if not field in accepted: + if field not in accepted: continue override_val = accepted[field][-1].strip() if override_val: @@ -198,7 +202,7 @@ def main(client_id, user_arguments_dict): list(cert_field_map) + given_peers] # Write-protect ID fields in auto-mode or if already logged in if keyword_auto in accepted['ro_fields'] or client_id: - ro_fields += [i for i in list(cert_field_map) if not i in ro_fields] + ro_fields += [i for i in list(cert_field_map) if i not in ro_fields] if reset_token: user_fields['reset_token'] = reset_token lock_fields = given_peers + ['comment'] diff --git a/mig/shared/functionality/reqoid.py b/mig/shared/functionality/reqoid.py index 578739312..1131ce06f 100644 --- a/mig/shared/functionality/reqoid.py +++ b/mig/shared/functionality/reqoid.py @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -40,7 +41,7 @@ from mig.shared.defaults import csrf_field, keyword_auto from mig.shared.functional import validate_input from mig.shared.handlers import get_csrf_limit, make_csrf_token -from mig.shared.init import find_entry, initialize_main_variables +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.pwcrypto import parse_password_policy from mig.shared.safeinput import html_escape from mig.shared.useradm import get_full_user_map @@ -68,21 +69,24 @@ def signature(configuration): return ['html_form', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False, 'op_menu': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False, op_menu=False) client_dir = client_id_dir(client_id) defaults = signature(configuration)[1] (validate_status, accepted) = validate_input(user_arguments_dict, defaults, output_objects, - allow_rejects=False) + allow_rejects=False, + environ=environ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) if not configuration.site_enable_openid or \ - not 'migoid' in configuration.site_signup_methods: + 'migoid' not in configuration.site_signup_methods: output_objects.append( {'object_type': 'error_text', 'text': '''Local OpenID login is not enabled on this site'''}) @@ -163,7 +167,7 @@ def main(client_id, user_arguments_dict): user_map = get_full_user_map(configuration) user_dict = user_map.get(client_id, {}) peers_fields = ['peers_%s' % field for field in - configuration.site_peers_explicit_fields] + configuration.site_peers_explicit_fields] for peers_field in peers_fields: peers_value = user_dict.get(peers_field, '') if peers_value: @@ -171,7 +175,7 @@ def main(client_id, user_arguments_dict): # Override with arg values if set for field in user_fields: - if not field in accepted: + if field not in accepted: continue override_val = accepted[field][-1].strip() if override_val: @@ -210,7 +214,7 @@ def main(client_id, user_arguments_dict): list(cert_field_map) + given_peers] # Write-protect ID fields in auto-mode or if already logged in if keyword_auto in accepted['ro_fields'] or client_id: - ro_fields += [i for i in list(cert_field_map) if not i in ro_fields] + ro_fields += [i for i in list(cert_field_map) if i not in ro_fields] if reset_token: user_fields['reset_token'] = reset_token lock_fields = given_peers + ['comment'] diff --git a/mig/shared/functionality/resedit.py b/mig/shared/functionality/resedit.py index 236f2aba6..45b4cfa0a 100644 --- a/mig/shared/functionality/resedit.py +++ b/mig/shared/functionality/resedit.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # resedit - Resource editor back end -# Copyright (C) 2003-2019 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,14 +20,14 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # -# Martin Rehr martin@rehr.dk August 2005 - """Display resource editor""" + from __future__ import print_function from __future__ import absolute_import @@ -39,7 +39,7 @@ from mig.shared.defaults import csrf_field from mig.shared.functional import validate_input_and_cert from mig.shared.handlers import get_csrf_limit, make_csrf_token -from mig.shared.init import initialize_main_variables, find_entry +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.refunctions import list_runtime_environments from mig.shared.resource import init_conf, empty_resource_config from mig.shared.vgridaccess import res_vgrid_access @@ -87,11 +87,13 @@ def available_choices(configuration, client_id, resource_id, field, spec): return choices -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs=None): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) defaults = signature()[1] (validate_status, accepted) = validate_input_and_cert( user_arguments_dict, @@ -100,6 +102,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) @@ -187,7 +190,7 @@ def main(client_id, user_arguments_dict): if hosturl: try: hostip = conf.get('HOSTIP', socket.gethostbyname(hosturl)) - except: + except Exception: hostip = '' output_objects.append({'object_type': 'html_form', 'text': """
%s: help
diff --git a/mig/shared/functionality/vgridman.py b/mig/shared/functionality/vgridman.py index 88b72355e..09e77ea39 100644 --- a/mig/shared/functionality/vgridman.py +++ b/mig/shared/functionality/vgridman.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # vgridman - backend to manage vgrids -# Copyright (C) 2003-2025 The MiG Project lead by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -20,7 +20,8 @@ # # 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. +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, +# USA. # # -- END_HEADER --- # @@ -36,7 +37,7 @@ from mig.shared.functional import validate_input_and_cert from mig.shared.handlers import get_csrf_limit, make_csrf_token from mig.shared.htmlgen import man_base_js, man_base_html, html_post_helper -from mig.shared.init import initialize_main_variables, find_entry +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.modified import pending_vgrids_update from mig.shared.useradm import get_full_user_map from mig.shared.vgrid import vgrid_create_allowed @@ -55,11 +56,13 @@ def signature(): return ['vgrids', defaults] -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs={'op_header': False}): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) status = returnvalues.OK defaults = signature()[1] title_entry = find_entry(output_objects, 'title') @@ -73,6 +76,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) @@ -80,7 +84,7 @@ def main(client_id, user_arguments_dict): operation = accepted['operation'][-1] caching = (accepted['caching'][-1].lower() in ('true', 'yes')) - if not operation in allowed_operations: + if operation not in allowed_operations: output_objects.append({'object_type': 'error_text', 'text': '''Operation must be one of %s.''' % ', '.join(allowed_operations)}) @@ -96,7 +100,7 @@ def main(client_id, user_arguments_dict): user_settings = title_entry.get('user_settings', {}) collaboration_links = user_settings.get( 'SITE_COLLABORATION_LINKS', 'default') - if not collaboration_links in configuration.site_collaboration_links or \ + if collaboration_links not in configuration.site_collaboration_links or \ collaboration_links == 'default': active_vgrid_links += configuration.site_default_vgrid_links elif collaboration_links == 'advanced': diff --git a/mig/shared/functionality/vgridworkflows.py b/mig/shared/functionality/vgridworkflows.py index abc429fdd..6fb7ed5d9 100644 --- a/mig/shared/functionality/vgridworkflows.py +++ b/mig/shared/functionality/vgridworkflows.py @@ -52,10 +52,9 @@ from mig.shared.fileio import unpickle, makedirs_rec, move_file from mig.shared.functional import validate_input_and_cert, REJECT_UNSET from mig.shared.htmlgen import man_base_js, man_base_html -from mig.shared.init import initialize_main_variables, find_entry +from mig.shared.init import find_entry, lazy_init_backend from mig.shared.parseflags import verbose -from mig.shared.vgrid import vgrid_add_remove_table, vgrid_is_owner_or_member, \ - vgrid_triggers, vgrid_set_triggers +from mig.shared.vgrid import vgrid_add_remove_table, vgrid_is_owner_or_member default_pager_entries = 20 @@ -107,11 +106,13 @@ def read_trigger_log(configuration, vgrid_name, flags): return log_content -def main(client_id, user_arguments_dict): - """Main function used by front end""" +def main(client_id, user_arguments_dict, environ=None, init_main_res=None, + init_kwargs=None): + """Main function wrapper used by front end""" + + (configuration, logger, output_objects, op_name, environ) = \ + lazy_init_backend(client_id, environ, init_main_res, init_kwargs) - (configuration, logger, output_objects, op_name) = \ - initialize_main_variables(client_id, op_header=False) defaults = signature()[1] title_entry = find_entry(output_objects, 'title') label = "%s" % configuration.site_vgrid_label @@ -124,6 +125,7 @@ def main(client_id, user_arguments_dict): client_id, configuration, allow_rejects=False, + environ=environ, ) if not validate_status: return (accepted, returnvalues.CLIENT_ERROR) @@ -140,7 +142,7 @@ def main(client_id, user_arguments_dict): % vgrid_name}) return (output_objects, returnvalues.CLIENT_ERROR) - if not operation in allowed_operations: + if operation not in allowed_operations: output_objects.append({'object_type': 'error_text', 'text': '''Operation must be one of %s.''' % ', '.join(allowed_operations)}) diff --git a/mig/shared/init.py b/mig/shared/init.py index d16fc776d..386cf733c 100644 --- a/mig/shared/init.py +++ b/mig/shared/init.py @@ -4,7 +4,7 @@ # --- BEGIN_HEADER --- # # init - shared helpers to init functionality backends -# Copyright (C) 2003-2024 The MiG Project lead by Brian Vinter +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -31,13 +31,17 @@ import mimetypes import os -import time from mig.shared.base import requested_backend, extract_field from mig.shared.conf import get_configuration_object +from mig.shared.defaults import keyword_auto from mig.shared.htmlgen import themed_styles, themed_scripts from mig.shared.settings import load_settings, load_widgets, load_profile +title_style_helpers = {'base': '', 'advanced': '', 'page': '', 'skin': ''} +title_script_helpers = {'base': '', 'advanced': '', 'skin': '', 'page': '', + 'init': '', 'ready': '', 'body': ''} + def make_basic_entry(kind, values): """Create basic entry for output_objects""" @@ -51,10 +55,14 @@ def make_start_entry(headers=[]): return make_basic_entry('start', {'headers': headers}) -def make_title_entry(text, meta='', style={}, script={}, skipmenu=False, +def make_title_entry(text, meta='', style=None, script=None, skipmenu=False, skipwidgets=False, skipuserstyle=False, skipuserprofile=False, backend=''): """Create title entry for output_objects""" + if style is None: + style = title_style_helpers.copy() + if script is None: + script = title_script_helpers.copy() return make_basic_entry('title', {'text': text, 'meta': meta, 'style': style, @@ -139,6 +147,8 @@ def initialize_main_variables(client_id, op_title=True, op_header=True, output_objects.append(start_entry) op_name = requested_backend() + if op_menu == keyword_auto: + op_menu = bool(client_id) if op_title: skipwidgets = not configuration.site_enable_widgets or not client_id skipuserstyle = not configuration.site_enable_styling or not client_id @@ -148,7 +158,7 @@ def initialize_main_variables(client_id, op_title=True, op_header=True, skipuserprofile=(not client_id), backend=op_name,) # Make sure base_menu is always set for extract_menu - # Typicall overriden based on client_id cases below + # Typically overriden based on client_id cases below title_object['base_menu'] = configuration.site_default_menu output_objects.append(title_object) if op_header: @@ -169,7 +179,7 @@ def initialize_main_variables(client_id, op_title=True, op_header=True, if settings: title['user_settings'] = settings base_menu = settings.get('SITE_BASE_MENU', 'default') - if not base_menu in configuration.site_base_menu: + if base_menu not in configuration.site_base_menu: base_menu = 'default' if base_menu == 'simple' and configuration.site_simple_menu: title['base_menu'] = configuration.site_simple_menu @@ -225,3 +235,54 @@ def extract_menu(configuration, title_entry): else: menu_items = configuration.site_default_menu return menu_items + + +def lazy_init_backend(client_id, environ=None, init_main_res=None, + init_kwargs=None): + """Helper to allow direct and lazy init of backend main functions. Can be + passed an existing initialize_main_variables result tuple and environ for + direct use, but if either is left to None they will be populated with the + result of an initialize_main_variables call and as os.environ respectively. + The initialize_main_variables call will include any additional kwars from + init_kwargs if so. + The init_main_res tuple additionally is lazy filled if it is incomplete for + more flexible use e.g. in unit tests. + """ + if environ is None: + environ = os.environ + if init_kwargs is None: + init_kwargs = {} + if init_main_res is None: + (configuration, logger, output_objects, op_name) = \ + initialize_main_variables(client_id, **init_kwargs) + else: + (configuration, logger, output_objects, op_name) = init_main_res + # Lazy fill any partial or missing entries + if configuration is None: + configuration = get_configuration_object() + if logger is None: + logger = configuration.logger + if not op_name: + op_name = requested_backend() + # Create new output_objects list with start entry if None was supplied + if output_objects is None: + output_objects = [make_start_entry()] + # Mimic initialize_main_variables use with init_kwargs + op_menu = init_kwargs.get('op_menu', True) + if op_menu == keyword_auto: + op_menu = bool(client_id) + skipwidgets = not configuration.site_enable_widgets or not client_id + skipuserstyle = not configuration.site_enable_styling or not client_id + title_object = make_title_entry('%s' % op_name, + skipmenu=(not op_menu), + skipwidgets=skipwidgets, + skipuserstyle=skipuserstyle, + skipuserprofile=(not client_id), + backend=op_name, + ) + # Make sure base_menu is always set for extract_menu + # Typically overriden based on client_id cases below + title_object['base_menu'] = configuration.site_default_menu + output_objects.append(title_object) + + return (configuration, logger, output_objects, op_name, environ) diff --git a/mig/shared/output.py b/mig/shared/output.py index cba0d832d..3db4301da 100644 --- a/mig/shared/output.py +++ b/mig/shared/output.py @@ -54,7 +54,8 @@ from mig.shared.defaults import file_dest_sep, keyword_any, keyword_updating from mig.shared.htmlgen import get_xgi_html_header, get_xgi_html_footer, \ vgrid_items, html_post_helper, tablesorter_pager -from mig.shared.init import find_entry, find_entry_index +from mig.shared.init import find_entry, find_entry_index, \ + title_script_helpers, title_style_helpers from mig.shared.objecttypes import validate from mig.shared.prettyprinttable import pprint_table from mig.shared.pwcrypto import sorted_hash_algos @@ -788,15 +789,13 @@ def html_format(configuration, ret_val, ret_msg, out_obj): meta = i.get('meta', '') backend = i.get('backend', '') style_entry = i.get('style', '') - style_helpers = {'base': '', 'advanced': '', 'page': '', - 'skin': ''} + style_helpers = title_style_helpers.copy() if isinstance(style_entry, dict): style_helpers.update(style_entry) else: style_helpers['base'] += style_entry script_entry = i.get('script', '') - script_helpers = {'base': '', 'advanced': '', 'skin': '', - 'page': '', 'init': '', 'ready': '', 'body': ''} + script_helpers = title_script_helpers.copy() if isinstance(script_entry, dict): script_helpers.update(script_entry) else: diff --git a/tests/support/wsgisupp.py b/tests/support/wsgisupp.py index 1105d0db8..acc906ae0 100644 --- a/tests/support/wsgisupp.py +++ b/tests/support/wsgisupp.py @@ -51,6 +51,27 @@ def __call__(self, status, headers, exc=None): self.calls.append((status, headers, exc)) +def create_http_environ(configuration, path_info=''): + """Small helper that can create a minimum viable environ dict suitable + for passing to http-facing code for the supplied configuration. + """ + + environ = {} + environ["MIG_CONF"] = configuration.config_file + environ["HTTP_HOST"] = "localhost" + environ["PATH_INFO"] = path_info + environ["REMOTE_ADDR"] = "127.0.0.1" + environ["SCRIPT_URI"] = "".join( + ("https://", environ["HTTP_HOST"], path_info) + ) + return environ + + +def filter_output_objects(output_objects, with_object_type=None): + """Filter output objects to pick only those of a specific object_type""" + return [o for o in output_objects if o["object_type"] == with_object_type] + + def create_wsgi_environ(configuration, wsgi_url, method='GET', query=None, headers=None, form=None): """Populate the necessary variables that will constitute a valid WSGI environment given a URL to which we will make a requests under test and diff --git a/tests/test_mig_shared_functionality_cat.py b/tests/test_mig_shared_functionality_cat.py index 619c3dd4e..75813c544 100644 --- a/tests/test_mig_shared_functionality_cat.py +++ b/tests/test_mig_shared_functionality_cat.py @@ -3,7 +3,7 @@ # --- BEGIN_HEADER --- # # test_mig_shared_functionality_cat - unit test of the corresponding mig module -# Copyright (C) 2003-2024 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -28,176 +28,206 @@ """Unit tests of the MiG functionality file implementing the cat backend""" from __future__ import print_function -import importlib + import os import shutil -import sys -import unittest - -from tests.support import MIG_BASE, PY2, TEST_DATA_DIR, MigTestCase, testmain, \ - temppath, ensure_dirs_exist - -from mig.shared.base import client_id_dir -from mig.shared.functionality.cat import _main as submain, main as realmain - - -def create_http_environ(configuration): - """Small helper that can create a minimum viable environ dict suitable - for passing to http-facing code for the supplied configuration. - """ - environ = {} - environ['MIG_CONF'] = configuration.config_file - environ['HTTP_HOST'] = 'localhost' - environ['PATH_INFO'] = '/' - environ['REMOTE_ADDR'] = '127.0.0.1' - environ['SCRIPT_URI'] = ''.join(('https://', environ['HTTP_HOST'], - environ['PATH_INFO'])) - return environ +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues +# Imports of the code under test +from mig.shared.functionality.cat import main as backend_main -def _only_output_objects(output_objects, with_object_type=None): - return [o for o in output_objects if o['object_type'] == with_object_type] +# Imports required for the unit tests themselves +from tests.support import ( + TEST_DATA_DIR, + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects -class MigSharedFunctionalityCat(MigTestCase): +class MigSharedFunctionalityCat(MigTestCase, UserAssertMixin): """Wrap unit tests for the corresponding module""" - TEST_CLIENT_ID = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=test@example.com' - def _provide_configuration(self): - return 'testconfig' + return "testconfig" def before_each(self): - self.test_user_dir = self._provision_test_user(self, self.TEST_CLIENT_ID) - self.test_environ = create_http_environ(self.configuration) - - def assertSingleOutputObject(self, output_objects, with_object_type=None): - assert with_object_type is not None - found_objects = _only_output_objects(output_objects, - with_object_type=with_object_type) - self.assertEqual(len(found_objects), 1) - return found_objects[0] + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/cat.py" + ) def test_file_serving_a_single_file_match(self): - with open(os.path.join(self.test_user_dir, 'foobar.txt'), 'w'): - pass + test_data = "424242" + with open(os.path.join(self.test_user_dir, "foobar.txt"), "w") as fd: + fd.write(test_data) payload = { - 'path': ['foobar.txt'], + "path": ["foobar.txt"], } - (output_objects, status) = submain(self.configuration, self.logger, - client_id=self.TEST_CLIENT_ID, - user_arguments_dict=payload, - environ=self.test_environ) - - # NOTE: start entry with headers and actual content - self.assertEqual(len(output_objects), 2) - self.assertSingleOutputObject(output_objects, - with_object_type='file_output') + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # NOTE: start entry with headers, title, header and actual content + self.assertEqual(len(output_objects), 3) + file_output_objects = filter_output_objects( + output_objects, with_object_type="file_output" + ) + self.assertEqual(len(file_output_objects), 1) + relevant_obj = file_output_objects[0] + self.assertEqual(len(relevant_obj["lines"]), 1) + self.assertEqual(relevant_obj["lines"][0], test_data) def test_file_serving_at_limit(self): test_binary_file = os.path.realpath( - os.path.join(TEST_DATA_DIR, 'loading.gif')) + os.path.join(TEST_DATA_DIR, "loading.gif") + ) test_binary_file_size = os.stat(test_binary_file).st_size - with open(test_binary_file, 'rb') as fh_test_file: + with open(test_binary_file, "rb") as fh_test_file: test_binary_file_data = fh_test_file.read() - shutil.copyfile(test_binary_file, os.path.join( - self.test_user_dir, 'loading.gif')) + shutil.copyfile( + test_binary_file, os.path.join(self.test_user_dir, "loading.gif") + ) payload = { - 'output_format': ['file'], - 'path': ['loading.gif'], + "output_format": ["file"], + "path": ["loading.gif"], } self.configuration.wwwserve_max_bytes = test_binary_file_size - (output_objects, status) = submain(self.configuration, self.logger, - client_id=self.TEST_CLIENT_ID, - user_arguments_dict=payload, - environ=self.test_environ) - - self.assertEqual(len(output_objects), 2) - relevant_obj = self.assertSingleOutputObject(output_objects, - with_object_type='file_output') - self.assertEqual(len(relevant_obj['lines']), 1) - self.assertEqual(relevant_obj['lines'][0], test_binary_file_data) + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # NOTE: start entry with headers, title and actual content + self.assertEqual(len(output_objects), 3) + file_output_objects = filter_output_objects( + output_objects, with_object_type="file_output" + ) + self.assertEqual(len(file_output_objects), 1) + relevant_obj = file_output_objects[0] + self.assertEqual(len(relevant_obj["lines"]), 1) + self.assertEqual(relevant_obj["lines"][0], test_binary_file_data) def test_file_serving_over_limit_without_storage_protocols(self): - test_binary_file = os.path.realpath(os.path.join(TEST_DATA_DIR, - 'loading.gif')) + test_binary_file = os.path.realpath( + os.path.join(TEST_DATA_DIR, "loading.gif") + ) test_binary_file_size = os.stat(test_binary_file).st_size - with open(test_binary_file, 'rb') as fh_test_file: - test_binary_file_data = fh_test_file.read() - shutil.copyfile(test_binary_file, os.path.join(self.test_user_dir, - 'loading.gif')) + with open(test_binary_file, "rb") as fh_test_file: + _ = fh_test_file.read() + shutil.copyfile( + test_binary_file, os.path.join(self.test_user_dir, "loading.gif") + ) payload = { - 'output_format': ['file'], - 'path': ['loading.gif'], + "output_format": ["file"], + "path": ["loading.gif"], } # NOTE: override default storage_protocols to empty in this test self.configuration.storage_protocols = [] self.configuration.wwwserve_max_bytes = test_binary_file_size - 1 - (output_objects, status) = submain(self.configuration, self.logger, - client_id=self.TEST_CLIENT_ID, - user_arguments_dict=payload, - environ=self.test_environ) - - # NOTE: start entry with headers and actual error message - self.assertEqual(len(output_objects), 2) - relevant_obj = self.assertSingleOutputObject(output_objects, - with_object_type='error_text') - self.assertEqual(relevant_obj['text'], - "Site configuration prevents web serving contents " - "bigger than 3896 bytes") + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.REJECT_PROCESSING_ERROR) + + # NOTE: start entry with headers, title and actual error message + self.assertEqual(len(output_objects), 3) + error_text_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_text_objects), 1) + self.assertEqual( + error_text_objects[0]["text"], + "Site configuration prevents web serving contents " + "bigger than 3896 bytes", + ) def test_file_serving_over_limit_with_storage_protocols_sftp(self): - test_binary_file = os.path.realpath(os.path.join(TEST_DATA_DIR, - 'loading.gif')) + test_binary_file = os.path.realpath( + os.path.join(TEST_DATA_DIR, "loading.gif") + ) test_binary_file_size = os.stat(test_binary_file).st_size - with open(test_binary_file, 'rb') as fh_test_file: - test_binary_file_data = fh_test_file.read() - shutil.copyfile(test_binary_file, os.path.join(self.test_user_dir, - 'loading.gif')) + with open(test_binary_file, "rb") as fh_test_file: + _ = fh_test_file.read() + shutil.copyfile( + test_binary_file, os.path.join(self.test_user_dir, "loading.gif") + ) payload = { - 'output_format': ['file'], - 'path': ['loading.gif'], + "output_format": ["file"], + "path": ["loading.gif"], } - self.configuration.storage_protocols = ['sftp'] + self.configuration.storage_protocols = ["sftp"] self.configuration.wwwserve_max_bytes = test_binary_file_size - 1 - (output_objects, status) = submain(self.configuration, self.logger, - client_id=self.TEST_CLIENT_ID, - user_arguments_dict=payload, - environ=self.test_environ) - - # NOTE: start entry with headers and actual error message - relevant_obj = self.assertSingleOutputObject(output_objects, - with_object_type='error_text') - self.assertEqual(relevant_obj['text'], - "Site configuration prevents web serving contents " - "bigger than 3896 bytes - please use better " - "alternatives (SFTP) to retrieve large data") + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.REJECT_PROCESSING_ERROR) + + # NOTE: start entry with headers, title and actual error message + self.assertEqual(len(output_objects), 3) + error_text_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_text_objects), 1) + relevant_obj = error_text_objects[0] + self.assertEqual( + relevant_obj["text"], + "Site configuration prevents web serving contents " + "bigger than 3896 bytes - please use better " + "alternatives (SFTP) to retrieve large data", + ) - @unittest.skipIf(PY2, "Python 3 only") def test_main_passes_environ(self): + payload = {} try: - result = realmain(self.TEST_CLIENT_ID, {}, self.test_environ) + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) except Exception as unexpectedexc: raise AssertionError( - "saw unexpected exception: %s" % (unexpectedexc,)) + "saw unexpected exception: %s" % (unexpectedexc,) + ) - (output_objects, status) = result - self.assertEqual(status[1], 'Client error') + self.assertEqual(status, returnvalues.CLIENT_ERROR) - error_text_objects = _only_output_objects(output_objects, - with_object_type='error_text') + # NOTE: start entry with headers, title, header, three actual error + # messages and finally a Go back link. + self.assertEqual(len(output_objects), 6) + error_text_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) relevant_obj = error_text_objects[2] self.assertEqual( - relevant_obj['text'], 'Input arguments were rejected - not allowed for this script!') + relevant_obj["text"], + "Input arguments were rejected - not allowed for this script!", + ) -if __name__ == '__main__': +if __name__ == "__main__": testmain() diff --git a/tests/test_mig_shared_functionality_datatransfer.py b/tests/test_mig_shared_functionality_datatransfer.py index 3fe9e5fdc..10dde675e 100644 --- a/tests/test_mig_shared_functionality_datatransfer.py +++ b/tests/test_mig_shared_functionality_datatransfer.py @@ -3,7 +3,7 @@ # --- BEGIN_HEADER --- # # test_mig_shared_functionality_datatransfer - unit test of the corresponding mig module -# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH +# Copyright (C) 2003-2026 The MiG Project by the Science HPC Center at UCPH # # This file is part of MiG. # @@ -25,66 +25,69 @@ # --- END_HEADER --- # -"""Unit tests of the MiG functionality file implementing the datatransfer backend""" +"""Unit tests of the MiG functionality file implementing the datatransfer +backend. +""" from __future__ import print_function -import os +# Imports required for the unit test wrapping import mig.shared.returnvalues as returnvalues from mig.shared.defaults import CSRF_MINIMAL -from mig.shared.base import client_id_dir -from mig.shared.functionality.datatransfer import _main as submain, main as realmain +# Imports of the code under test +from mig.shared.functionality.datatransfer import main as backend_main + +# Imports required for the unit tests themselves from tests.support import ( MigTestCase, testmain, - temppath, - ensure_dirs_exist, ) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects -def create_http_environ(configuration): - """Small helper that can create a minimum viable environ dict suitable - for passing to http-facing code for the supplied configuration. - """ - - environ = {} - environ["MIG_CONF"] = configuration.config_file - environ["HTTP_HOST"] = "localhost" - environ["PATH_INFO"] = "/" - environ["REMOTE_ADDR"] = "127.0.0.1" - environ["SCRIPT_URI"] = "".join( - ("https://", environ["HTTP_HOST"], environ["PATH_INFO"]) - ) - return environ - - -def _only_output_objects(output_objects, with_object_type=None): - return [o for o in output_objects if o["object_type"] == with_object_type] - - -class MigSharedFunctionalityDataTransfer(MigTestCase): +class MigSharedFunctionalityDataTransfer(MigTestCase, UserAssertMixin): """Wrap unit tests for the corresponding module""" - TEST_CLIENT_ID = ( - "/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=test@example.com" - ) - def _provide_configuration(self): return "testconfig" def before_each(self): - self.test_user_dir = self._provision_test_user(self, self.TEST_CLIENT_ID) - self.test_environ = create_http_environ(self.configuration) + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/datatransfer.py" + ) + self.configuration.site_enable_transfers = True def test_default_disabled_site_transfer(self): - self.assertFalse(self.configuration.site_enable_transfers) + self.configuration.site_enable_transfers = False + payload = {} - result = realmain(self.TEST_CLIENT_ID, {}, self.test_environ) - (output_objects, status) = result + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) self.assertEqual(status, returnvalues.OK) - text_objects = _only_output_objects(output_objects, with_object_type="text") + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) self.assertEqual(len(text_objects), 1) self.assertIn("text", text_objects[0]) text_object = text_objects[0]["text"] @@ -95,41 +98,57 @@ def test_show_action_enabled_site_transfer(self): payload = {"action": ["show"]} self.configuration.site_enable_transfers = True - (output_objects, status) = submain( - self.configuration, - self.logger, - client_id=self.TEST_CLIENT_ID, + output_objects, status = backend_main( + client_id=TEST_USER_DN, user_arguments_dict=payload, environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), ) self.assertEqual(status, returnvalues.OK) - # We don't expect any text messages here - text_objects = _only_output_objects(output_objects, with_object_type="text") + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) self.assertEqual(len(text_objects), 0) + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 10) + def test_deltransfer_without_transfer_id(self): non_existing_transfer_id = "non-existing-transfer-id" - payload = {"action": ["deltransfer"], "transfer_id": [non_existing_transfer_id]} + payload = { + "action": ["deltransfer"], + "transfer_id": [non_existing_transfer_id], + } self.configuration.site_enable_transfers = True self.configuration.site_csrf_protection = CSRF_MINIMAL self.test_environ["REQUEST_METHOD"] = "post" - (output_objects, status) = submain( - self.configuration, - self.logger, - client_id=self.TEST_CLIENT_ID, + output_objects, status = backend_main( + client_id=TEST_USER_DN, user_arguments_dict=payload, environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), ) self.assertEqual(status, returnvalues.CLIENT_ERROR) - error_text_objects = _only_output_objects( + error_text_objects = filter_output_objects( output_objects, with_object_type="error_text" ) self.assertEqual(len(error_text_objects), 1) self.assertEqual( - error_text_objects[0]["text"], "existing transfer_id is required for delete" + error_text_objects[0]["text"], + "existing transfer_id is required for delete", ) def test_redotransfer_without_transfer_id(self): @@ -142,16 +161,15 @@ def test_redotransfer_without_transfer_id(self): self.configuration.site_csrf_protection = CSRF_MINIMAL self.test_environ["REQUEST_METHOD"] = "post" - (output_objects, status) = submain( - self.configuration, - self.logger, - client_id=self.TEST_CLIENT_ID, + output_objects, status = backend_main( + client_id=TEST_USER_DN, user_arguments_dict=payload, environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), ) self.assertEqual(status, returnvalues.CLIENT_ERROR) - error_text_objects = _only_output_objects( + error_text_objects = filter_output_objects( output_objects, with_object_type="error_text" ) self.assertEqual(len(error_text_objects), 1) @@ -161,7 +179,7 @@ def test_redotransfer_without_transfer_id(self): ) -# TODO, add additional tests that succesfully makes data transfers across a range of protocols +# TODO: extend tests to cover data transfers across a range of protocols if __name__ == "__main__": testmain() diff --git a/tests/test_mig_shared_functionality_docs.py b/tests/test_mig_shared_functionality_docs.py new file mode 100644 index 000000000..59538035b --- /dev/null +++ b/tests/test_mig_shared_functionality_docs.py @@ -0,0 +1,391 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_docs - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the docs backend""" + +from __future__ import print_function + +import unittest + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.docs import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityDocs(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/docs.py" + ) + self.configuration.site_enable_styling = False + self.configuration.site_enable_widgets = False + + def test_show_default_anonymous_site_docs(self): + payload = {"show": [""]} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + self.assertTrue(title_objects[0]["skipuserprofile"]) + # No bling here for anon users + self.assertTrue(title_objects[0]["skipuserstyle"]) + self.assertTrue(title_objects[0]["skipwidgets"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 2) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 6) + + def test_show_default_anonymous_site_docs_with_bling(self): + payload = {"show": [""]} + self.configuration.site_enable_styling = True + self.configuration.site_enable_widgets = True + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + self.assertTrue(title_objects[0]["skipuserprofile"]) + # No bling here for anon users + self.assertTrue(title_objects[0]["skipuserstyle"]) + self.assertTrue(title_objects[0]["skipwidgets"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 2) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 6) + + def test_show_default_site_docs(self): + payload = {"show": [""]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertFalse(title_objects[0]["skipmenu"]) + self.assertFalse(title_objects[0]["skipuserprofile"]) + # Only bling here if enabled in conf + self.assertNotEqual( + title_objects[0]["skipuserstyle"], + self.configuration.site_enable_styling, + ) + self.assertNotEqual( + title_objects[0]["skipwidgets"], + self.configuration.site_enable_widgets, + ) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 2) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 6) + + def test_show_default_site_docs_with_bling(self): + self.configuration.site_enable_styling = True + self.configuration.site_enable_widgets = True + payload = {"show": [""]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertFalse(title_objects[0]["skipmenu"]) + self.assertFalse(title_objects[0]["skipuserprofile"]) + # Only bling here if enabled in conf + self.assertNotEqual( + title_objects[0]["skipuserstyle"], + self.configuration.site_enable_styling, + ) + self.assertNotEqual( + title_objects[0]["skipwidgets"], + self.configuration.site_enable_widgets, + ) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 2) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 6) + + def test_show_default_credits(self): + payload = {"show": ["credits"]} + + output_objects, status = backend_main( + client_id=TEST_USER_DN, + user_arguments_dict=payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertTrue(len(text_objects) >= 20) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 3) + + # Check expected links + link_objects = filter_output_objects( + output_objects, with_object_type="link" + ) + self.assertTrue(len(link_objects) >= 20) + + def test_show_cracklib_credits_when_enabled(self): + self.configuration.site_password_cracklib = True + # Needs one service with password login to trigger conditional + self.configuration.site_enable_sftp = True + payload = {"show": ["credits"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected links + link_objects = filter_output_objects( + output_objects, with_object_type="link" + ) + self.assertTrue(len(link_objects) >= 20) + self.assertTrue( + [i for i in link_objects if "cracklib" in i["title"].lower()] + ) + + def test_hide_cracklib_credits_when_disabled(self): + self.configuration.site_password_cracklib = False + # Needs one service with password login to trigger conditional + self.configuration.site_enable_sftp = True + payload = {"show": ["credits"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected links + link_objects = filter_output_objects( + output_objects, with_object_type="link" + ) + self.assertTrue(len(link_objects) >= 20) + self.assertFalse( + [i for i in link_objects if "cracklib" in i["title"].lower()] + ) + + @unittest.skip("TODO: fix unused detection in backend and re-enable") + def test_hide_cracklib_credits_when_unused(self): + # Hidden if no password login service enabled + self.configuration.site_password_cracklib = True + self.configuration.site_enable_sftp = False + payload = {"show": ["credits"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected links + link_objects = filter_output_objects( + output_objects, with_object_type="link" + ) + self.assertTrue(len(link_objects) >= 20) + self.assertFalse( + [i for i in link_objects if "cracklib" in i["title"].lower()] + ) + + @unittest.skip("TODO: fix broken br tag and re-enable") + def test_show_default_credits_no_longer_has_broken_html_br_tag(self): + payload = {"show": ["credits"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 3) + self.assertFalse([i for i in html_objects if "
" in i["text"]]) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_fileman.py b/tests/test_mig_shared_functionality_fileman.py new file mode 100644 index 000000000..f6bb00747 --- /dev/null +++ b/tests/test_mig_shared_functionality_fileman.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_fileman - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the fileman backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.fileman import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityFileman(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/fileman.py" + ) + + def test_show_default_user_fileman(self): + payload = {} + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + self.assertEqual(header_objects[0]["text"], "File Manager") + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 1) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_freezedb.py b/tests/test_mig_shared_functionality_freezedb.py new file mode 100644 index 000000000..eb11823e2 --- /dev/null +++ b/tests/test_mig_shared_functionality_freezedb.py @@ -0,0 +1,138 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_freezedb - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the freezedb +backend. +""" + +from __future__ import print_function + +import unittest + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.freezedb import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityFreezedb(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/freezedb.py" + ) + self.configuration.site_enable_freeze = True + + @unittest.skip("TODO: fix error response in backend and re-enable") + def test_freezedb_disabled_site_freeze(self): + self.configuration.site_enable_freeze = False + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = "Freezing archives is disabled on this site" + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_user_freezedb(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 2) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 5) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_home.py b/tests/test_mig_shared_functionality_home.py new file mode 100644 index 000000000..16bc7915c --- /dev/null +++ b/tests/test_mig_shared_functionality_home.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_home - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the home backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.home import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityHome(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/home.py" + ) + + def test_show_default_user_home(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 0) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 1) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_jupyter.py b/tests/test_mig_shared_functionality_jupyter.py new file mode 100644 index 000000000..faeef1187 --- /dev/null +++ b/tests/test_mig_shared_functionality_jupyter.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_jupyter - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the jupyter backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.jupyter import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityJupyter(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/jupyter.py" + ) + self.configuration.site_enable_jupyter = True + self.configuration.site_enable_sftp = True + + def test_jupyter_disabled_site_jupyter(self): + self.configuration.site_enable_jupyter = False + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = ( + "The Jupyter service is not enabled on the system" + ) + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_user_jupyter(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 2) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_reqcert.py b/tests/test_mig_shared_functionality_reqcert.py new file mode 100644 index 000000000..4ee4e81f8 --- /dev/null +++ b/tests/test_mig_shared_functionality_reqcert.py @@ -0,0 +1,274 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_reqcert - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the reqcert backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.reqcert import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + ensure_dirs_exist, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + +TEST_USER_EMAIL = TEST_USER_DN.split("/emailAddress=", 1)[-1] + + +class MigSharedFunctionalityReqcert(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + ensure_dirs_exist(self.configuration.resource_home) + ensure_dirs_exist(self.configuration.vgrid_home) + ensure_dirs_exist(self.configuration.mig_system_files) + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/reqcert.py" + ) + self.configuration.ca_fqdn = "ca.migrid.org" + self.configuration.ca_user = "mig-ca" + self.configuration.site_signup_methods = ["migcert"] + + def test_reqcert_disabled_site_ca(self): + self.configuration.ca_fqdn = "" + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.CLIENT_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = ( + "User certificate requests are not supported on this site" + ) + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_reqcert_disabled_site_cert_signup(self): + self.configuration.ca_fqdn = "ca.migrid.org" + self.configuration.site_signup_methods = [] + payload = {} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = ( + "X.509 certificate login is not enabled on this site" + ) + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html messages + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_anonymous_user_reqcert(self): + payload = {} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 1) + relevant_obj = html_objects[0] + # TODO: polish reqcert and reqoid to be consistent here + plain_text = relevant_obj["text"].replace('\n', ' ') + self.assertIn("Please enter your information", plain_text) + self.assertNotIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + def test_show_url_prefill_user_reqcert(self): + payload = {"email": [TEST_USER_EMAIL]} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 1) + relevant_obj = html_objects[0] + # TODO: polish reqcert and reqoid to be consistent here + plain_text = relevant_obj["text"].replace('\n', ' ') + self.assertIn("Please enter your information", plain_text) + self.assertIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + def test_show_default_authenticated_user_reqcert(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 2) + relevant_obj = html_objects[0] + self.assertIn( + "you already have a valid MiG certificate", relevant_obj["text"] + ) + relevant_obj = html_objects[1] + self.assertIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + +# TODO: add additional tests to cover other uses +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_reqoid.py b/tests/test_mig_shared_functionality_reqoid.py new file mode 100644 index 000000000..cb97e25b0 --- /dev/null +++ b/tests/test_mig_shared_functionality_reqoid.py @@ -0,0 +1,268 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_reqoid - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the reqoid backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.reqoid import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + ensure_dirs_exist, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + +TEST_USER_EMAIL = TEST_USER_DN.split("/emailAddress=", 1)[-1] + + +class MigSharedFunctionalityReqoid(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + ensure_dirs_exist(self.configuration.resource_home) + ensure_dirs_exist(self.configuration.vgrid_home) + ensure_dirs_exist(self.configuration.mig_system_files) + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/reqoid.py" + ) + self.configuration.site_enable_openid = True + self.configuration.site_signup_methods = ["migoid"] + + def test_reqoid_disabled_site_openid(self): + self.configuration.site_enable_openid = False + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = "Local OpenID login is not enabled on this site" + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_reqoid_disabled_site_oid_signup(self): + self.configuration.site_signup_methods = [] + payload = {} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = "Local OpenID login is not enabled on this site" + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html messages + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_anonymous_user_reqoid(self): + payload = {} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 2) + relevant_obj = html_objects[1] + # TODO: polish reqcert and reqoid to be consistent here + plain_text = relevant_obj["text"].replace('\n', ' ') + self.assertIn("Please enter your information", plain_text) + self.assertNotIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + def test_show_url_prefill_user_reqoid(self): + payload = {"email": [TEST_USER_EMAIL]} + + output_objects, status = backend_main( + "", + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 2) + relevant_obj = html_objects[1] + # TODO: polish reqcert and reqoid to be consistent here + plain_text = relevant_obj["text"].replace('\n', ' ') + self.assertIn("Please enter your information", plain_text) + self.assertIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + def test_show_default_authenticated_user_reqoid(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertTrue(title_objects[0]["skipmenu"]) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 3) + relevant_obj = html_objects[1] + self.assertIn( + "you already have valid MiG credentials", relevant_obj["text"] + ) + relevant_obj = html_objects[2] + self.assertIn("value='%s'" % TEST_USER_EMAIL, relevant_obj["text"]) + + +# TODO: add additional tests to cover other uses +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_resedit.py b/tests/test_mig_shared_functionality_resedit.py new file mode 100644 index 000000000..50b31175e --- /dev/null +++ b/tests/test_mig_shared_functionality_resedit.py @@ -0,0 +1,137 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_resedit - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the resedit backend""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.resedit import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + ensure_dirs_exist, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityResedit(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + ensure_dirs_exist(self.configuration.resource_home) + ensure_dirs_exist(self.configuration.vgrid_home) + ensure_dirs_exist(self.configuration.mig_system_files) + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/resedit.py" + ) + self.configuration.site_enable_resources = True + + def test_resedit_disabled_site_resources(self): + self.configuration.site_enable_resources = False + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = "Resources are not enabled on this system" + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_user_resedit(self): + payload = {} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 4) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 54) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_vgridman.py b/tests/test_mig_shared_functionality_vgridman.py new file mode 100644 index 000000000..7c335c6e0 --- /dev/null +++ b/tests/test_mig_shared_functionality_vgridman.py @@ -0,0 +1,107 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_vgridman - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the vgridman +backend. +""" + +from __future__ import print_function + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.vgridman import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityVgridman(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/vgridman.py" + ) + + def test_show_default_user_vgridman(self): + payload = {} + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + self.assertEqual(header_objects[0]["text"], "VGrids") + + # Check expected title contents + title_objects = filter_output_objects( + output_objects, with_object_type="title" + ) + self.assertEqual(len(title_objects), 1) + self.assertEqual(title_objects[0]["text"], "VGrid Management") + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 3) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 7) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain() diff --git a/tests/test_mig_shared_functionality_vgridworkflows.py b/tests/test_mig_shared_functionality_vgridworkflows.py new file mode 100644 index 000000000..cc53d84c3 --- /dev/null +++ b/tests/test_mig_shared_functionality_vgridworkflows.py @@ -0,0 +1,142 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_functionality_vgridworkflows - unit test of the corresponding mig module +# Copyright (C) 2003-2026 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 of the MiG functionality file implementing the vgridworkflows +backend. +""" + +from __future__ import print_function + +import unittest + +# Imports required for the unit test wrapping +import mig.shared.returnvalues as returnvalues + +# Imports of the code under test +from mig.shared.functionality.vgridworkflows import main as backend_main + +# Imports required for the unit tests themselves +from tests.support import ( + MigTestCase, + ensure_dirs_exist, + testmain, +) +from tests.support.usersupp import TEST_USER_DN, UserAssertMixin +from tests.support.wsgisupp import create_http_environ, filter_output_objects + + +class MigSharedFunctionalityVgridworkflows(MigTestCase, UserAssertMixin): + """Wrap unit tests for the corresponding module""" + + def _provide_configuration(self): + return "testconfig" + + def before_each(self): + ensure_dirs_exist(self.configuration.resource_home) + ensure_dirs_exist(self.configuration.vgrid_home) + ensure_dirs_exist(self.configuration.mig_system_files) + self.test_user_dir = self._provision_test_user(self, TEST_USER_DN) + self.test_environ = create_http_environ( + self.configuration, "wsgi-bin/vgridworkflows.py" + ) + self.configuration.site_enable_workflows = True + + @unittest.skip("TODO: fix missing enabled check in backend and re-enable") + def test_vgridworkflows_disabled_site_workflows(self): + self.configuration.site_enable_workflows = False + payload = {"vgrid_name": ["Generic"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.SYSTEM_ERROR) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 1) + self.assertIn("text", error_objects[0]) + text_object = error_objects[0]["text"] + expected_response_msg = "Workflows are not enabled on this system" + self.assertIn(expected_response_msg, text_object) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 0) + + def test_show_default_user_vgridworkflows(self): + payload = {"vgrid_name": ["Generic"]} + + output_objects, status = backend_main( + TEST_USER_DN, + payload, + environ=self.test_environ, + init_main_res=(self.configuration, self.logger, None, None), + ) + self.assertEqual(status, returnvalues.OK) + + # Check expected error messages + error_objects = filter_output_objects( + output_objects, with_object_type="error_text" + ) + self.assertEqual(len(error_objects), 0) + + # Check expected header messages + header_objects = filter_output_objects( + output_objects, with_object_type="header" + ) + self.assertEqual(len(header_objects), 1) + + # Check expected text messages + text_objects = filter_output_objects( + output_objects, with_object_type="text" + ) + self.assertEqual(len(text_objects), 0) + + # Check expected html snippets + html_objects = filter_output_objects( + output_objects, with_object_type="html_form" + ) + self.assertEqual(len(html_objects), 11) + + +# TODO: add additional tests to cover other uses + +if __name__ == "__main__": + testmain()