Add test covering the useradm create_user function.

Leverage the opportunity to make sure that a call to create_user writes
the same content as has been declared as our on-disk user fixture.

Note, create_user is mostly about writing rather than about the population
of the user dictionary with most of that important work (in terms of the
written values) being done by the createuser calling script. This means
that the we are only comparing written data as of now, meaning it is
something of a loose test, but this towards additional rigour.

Author: albu-diku

envhelp/makeconfig.py

@@ -40,11 +40,10 @@
 
 from mig.shared.conf import get_configuration_object
 from mig.shared.install import MIG_BASE, generate_confs
+from mig.shared.useradm import _ensure_dirs_needed_for_userdb
 
 _LOCAL_ENVHELP_OUTPUT_DIR = os.path.join(_LOCAL_MIG_BASE, "envhelp/output")
 _MAKECONFIG_ALLOWED = ["local", "test"]
-_USERADM_PATH_KEYS = ('user_cache', 'user_db_home', 'user_home',
-                      'user_settings', 'mrsl_files_dir', 'resource_pending')
 
 
 def _at(sequence, index=-1, default=None):
@@ -55,18 +54,6 @@ def _at(sequence, index=-1, default=None):
         return default
 
 
-def _ensure_dirs_needed_for_userdb(configuration):
-    """Provision the basic directories needed for the operation of the
-     userdb deriving paths from the supplied configuration object."""
-
-    for config_key in _USERADM_PATH_KEYS:
-        dir_path = getattr(configuration, config_key).rstrip(os.path.sep)
-        try:
-            os.makedirs(dir_path, exist_ok=True)
-        except OSError as exc:
-            pass
-
-
 def write_testconfig(env_name, is_docker=False):
     is_predefined = env_name == 'test'
     confs_name = '%sconfs' % (env_name,)

mig/shared/useradm.py

@@ -102,6 +102,22 @@
 https_authdigests = user_db_filename
 
 
+_USERADM_PATH_KEYS = ('user_db_home', 'user_home', 'user_settings',
+                      'user_cache', 'mrsl_files_dir', 'resource_pending')
+
+
+def _ensure_dirs_needed_for_userdb(configuration):
+    """Provision the basic directories needed for the operation of the
+     userdb deriving paths from the supplied configuration object."""
+
+    for config_key in _USERADM_PATH_KEYS:
+        dir_path = getattr(configuration, config_key).rstrip(os.path.sep)
+        try:
+            os.makedirs(dir_path, exist_ok=True)
+        except OSError as exc:
+            pass
+
+
 def init_user_adm(dynamic_db_path=True):
     """Shared init function for all user administration scripts.
     The optional dynamic_db_path argument toggles dynamic user db path lookup

tests/fixture/MiG-users.db--example.json.ini

@@ -1,2 +1,2 @@
 [ONWRITE]
-convert_dict_bytes_to_strings_kv = True
+convert_dict_strings_to_bytes_kv = True

tests/support/fixturesupp.py

@@ -101,7 +101,33 @@ def _hints_apply_today_relative(value, modifier):
         raise NotImplementedError("unspported today_relative modifier")
 
 
-def _hints_apply_dict_bytes_to_strings_kv(input_dict):
+def _hints_apply_dict_bytes_to_strings_kv(input_dict, modifier):
+    assert modifier is None
+
+    output_dict = {}
+
+    for k, v in input_dict.items():
+        key_to_use = k
+        if isinstance(k, bytes):
+            key_to_use = str(k, 'utf8')
+
+        if isinstance(v, dict):
+            output_dict[key_to_use] = _hints_apply_dict_bytes_to_strings_kv(v, modifier)
+            continue
+
+        val_to_use = v
+        if isinstance(v, bytes):
+            val_to_use = str(v, 'utf8')
+
+        output_dict[key_to_use] = val_to_use
+
+    return output_dict
+
+
+
+def _hints_apply_dict_strings_to_bytes_kv(input_dict, modifier):
+    assert modifier is None
+
     output_dict = {}
 
     for k, v in input_dict.items():
@@ -110,7 +136,7 @@ def _hints_apply_dict_bytes_to_strings_kv(input_dict):
             key_to_use = bytes(k, 'utf8')
 
         if isinstance(v, dict):
-            output_dict[key_to_use] = _hints_apply_dict_bytes_to_strings_kv(v)
+            output_dict[key_to_use] = _hints_apply_dict_strings_to_bytes_kv(v, modifier)
             continue
 
         val_to_use = v
@@ -122,14 +148,22 @@ def _hints_apply_dict_bytes_to_strings_kv(input_dict):
     return output_dict
 
 
-_FIXTUREFILE_APPLIERS_ATTRIBUTES = {
+_FIXTUREFILE_APPLIERS_ARGLESS = {
     'array_of_tuples': _hints_apply_array_of_tuples,
     'today_relative': _hints_apply_today_relative,
+    'convert_dict_bytes_to_strings_kv': _hints_apply_dict_bytes_to_strings_kv,
+    'convert_dict_strings_to_bytes_kv': _hints_apply_dict_strings_to_bytes_kv,
+}
+
+
+_FIXTUREFILE_APPLIERS_ATTRIBUTES = {
+    'array_of_tuples': _FIXTUREFILE_APPLIERS_ARGLESS['array_of_tuples'],
+    'today_relative': _FIXTUREFILE_APPLIERS_ARGLESS['today_relative'],
 }
 
 
 _FIXTUREFILE_APPLIERS_ONWRITE = {
-    'convert_dict_bytes_to_strings_kv': _hints_apply_dict_bytes_to_strings_kv,
+    'convert_dict_strings_to_bytes_kv': _FIXTUREFILE_APPLIERS_ARGLESS['convert_dict_strings_to_bytes_kv'],
 }
 
 
@@ -301,7 +335,7 @@ def write_to_dir(self, target_dir, output_format=None):
                 continue
 
             apply_conversion = _FIXTUREFILE_APPLIERS_ONWRITE[item_name]
-            output_data = apply_conversion(output_data)
+            output_data = apply_conversion(output_data, None)
 
         if output_format == 'binary':
             with open(fixture_file_target, 'wb') as fixture_outputfile:

tests/support/picklesupp.py

@@ -0,0 +1,59 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# __init__ - package marker and core package functions
+# Copyright (C) 2003-2025  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 ---
+#
+
+"""Pickle related details within the test support library."""
+
+import pickle
+
+from tests.support.suppconst import TEST_OUTPUT_DIR
+from tests.support.fixturesupp import _FIXTUREFILE_APPLIERS_ARGLESS
+
+
+class PickleAssertMixin:
+    """Assertions for working with pickled files to be used as a mixin."""
+
+    def assertPickledFile(self, pickle_file_path, apply_hints=None):
+        """
+        Check a paricular pickled file exists and is loadable.
+
+        Any data contained within it is returned for further assertions
+        having been optionally transformed by any requested hints.
+        """
+
+        with open(pickle_file_path, 'rb') as picklefile:
+            pickled = pickle.load(picklefile)
+
+        if not apply_hints:
+            return pickled
+
+        result = pickled
+        for hint_name in apply_hints:
+            if not hint_name in _FIXTUREFILE_APPLIERS_ARGLESS:
+                raise NotImplementedError("unknown hint %s" % (hint_name,))
+            hint_fn = _FIXTUREFILE_APPLIERS_ARGLESS[hint_name]
+            result = hint_fn(pickled, modifier=None)
+        return result

tests/test_mig_shared_useradm.py

@@ -38,10 +38,13 @@
 
 from tests.support import MIG_BASE, TEST_OUTPUT_DIR, MigTestCase, \
     FakeConfiguration, testmain, cleanpath, is_path_within
+from tests.support.fixturesupp import FixtureAssertMixin
+from tests.support.picklesupp import PickleAssertMixin
 
 from mig.shared.defaults import keyword_auto, htaccess_filename, \
     DEFAULT_USER_ID_FORMAT
-from mig.shared.useradm import assure_current_htaccess
+from mig.shared.useradm import assure_current_htaccess, create_user, \
+    _ensure_dirs_needed_for_userdb
 
 DUMMY_USER = 'dummy-user'
 DUMMY_STALE_USER = 'dummy-stale-user'
@@ -80,8 +83,87 @@
                                )
 
 
+class TestMigSharedUsedadm_create_user(MigTestCase,
+                                       FixtureAssertMixin,
+                                       PickleAssertMixin):
+    """Coverage of useradm create_user function."""
+
+    def before_each(self):
+        configuration = self.configuration
+
+        _ensure_dirs_needed_for_userdb(self.configuration)
+
+        self.expected_user_db_home = configuration.user_db_home[0:-1]
+        self.expected_user_db_file = os.path.join(
+            self.expected_user_db_home, 'MiG-users.db')
+
+    def _provide_configuration(self):
+        return 'testconfig'
+
+    def test_user_db_is_created(self):
+        user_dict = {}
+        user_dict['full_name'] = "Test User"
+        user_dict['organization'] = "Test Org"
+        user_dict['state'] = "NA"
+        user_dict['country'] = "DK"
+        user_dict['email'] = "[email protected]"
+        user_dict['comment'] = "This is the create comment"
+        user_dict['password'] = "password"
+        create_user(user_dict, self.configuration,
+                    keyword_auto, default_renew=True)
+
+        # presence of user home
+        path_kind = MigTestCase._absolute_path_kind(self.expected_user_db_home)
+        self.assertEqual(path_kind, 'dir')
+
+        # presence of user db
+        path_kind = MigTestCase._absolute_path_kind(self.expected_user_db_file)
+        self.assertEqual(path_kind, 'file')
+
+    def test_user_entry_is_recorded(self):
+        def _adjust_user_dict_for_compare(user_obj):
+            obj = dict(user_obj)
+            obj['created'] = 9999999999.9999999
+            obj['expire'] = 9999999999.9999999
+            obj['unique_id'] = '__UNIQUE_ID__'
+            return obj
+
+        expected_user_id = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]'
+        expected_user_password_hash = "PBKDF2$sha256$10000$XMZGaar/pU4PvWDr$w0dYjezF6JGtSiYPexyZMt3lM2134uix"
+
+        user_dict = {}
+        user_dict['full_name'] = "Test User"
+        user_dict['organization'] = "Test Org"
+        user_dict['state'] = "NA"
+        user_dict['country'] = "DK"
+        user_dict['email'] = "[email protected]"
+        user_dict['comment'] = "This is the create comment"
+        user_dict['locality'] = ""
+        user_dict['organizational_unit'] = ""
+        user_dict['password'] = ""
+        user_dict['password_hash'] = expected_user_password_hash
+
+        create_user(user_dict, self.configuration,
+                    keyword_auto, default_renew=True)
+
+        pickled = self.assertPickledFile(self.expected_user_db_file,
+                                         apply_hints=['convert_dict_bytes_to_strings_kv'])
+        self.assertIn(expected_user_id, pickled)
+
+        prepared = self.prepareFixtureAssert('MiG-users.db--example',
+                                             fixture_format='json')
+
+        # TODO: remove resetting the handful of keys here
+        #       this is done to allow the comparision to succeed
+        actual_user_object = _adjust_user_dict_for_compare(pickled[expected_user_id])
+        expected_user_object = _adjust_user_dict_for_compare(prepared.fixture_data[expected_user_id])
+
+        self.maxDiff = None
+        self.assertEqual(actual_user_object, expected_user_object)
+
+
 class MigSharedUseradm__assure_current_htaccess(MigTestCase):
-    """Unit test helper for the migrid code pointed to in class name"""
+    """Coverage of useradm behaviours around htaccess."""
 
     def before_each(self):
         """The create_user call requires quite a few helper dirs"""