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/__init__.py

@@ -43,7 +43,8 @@
 
 from tests.support.configsupp import FakeConfiguration
 from tests.support.fixturesupp import _PreparedFixture
-from tests.support.suppconst import MIG_BASE, TEST_BASE, \
+from tests.support.pathsupp import is_path_within
+from tests.support.suppconst import MIG_BASE, TEST_BASE, TEST_FIXTURE_DIR, \
     TEST_DATA_DIR, TEST_OUTPUT_DIR, ENVHELP_OUTPUT_DIR
 
 from tests.support._env import MIG_ENV, PY2
@@ -383,16 +384,6 @@ def _provision_test_user(testcase, distinguished_name):
         return test_user_dir
 
 
-def is_path_within(path, start=None, _msg=None):
-    """Check if path is within start directory"""
-    try:
-        assert os.path.isabs(path), _msg
-        relative = os.path.relpath(path, start=start)
-    except:
-        return False
-    return not relative.startswith('..')
-
-
 def ensure_dirs_exist(absolute_dir):
     """A simple helper to create absolute_dir and any parents if missing"""
     try:

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/pathsupp.py

@@ -0,0 +1,40 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# __init__ - package marker and core package functions
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+# -- END_HEADER ---
+#
+
+"""Path related details within the test support library."""
+
+import os
+
+
+def is_path_within(path, start=None, _msg=None):
+    """Check if path is within start directory"""
+    try:
+        assert os.path.isabs(path), _msg
+        relative = os.path.relpath(path, start=start)
+    except AssertionError:
+        return False
+    return not relative.startswith('..')

tests/support/picklesupp.py

@@ -0,0 +1,24 @@
+import pickle
+
+from tests.support.suppconst import TEST_OUTPUT_DIR
+from tests.support.fixturesupp import _FIXTUREFILE_APPLIERS_ARGLESS
+from tests.support.pathsupp import is_path_within
+
+
+class PickleAssertMixin:
+    def assertPickledFile(self, pickle_file_path, apply_hints=None):
+        #assert is_path_within(pickle_file_path, TEST_OUTPUT_DIR)
+
+        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,6 +83,87 @@
                                )
 
 
+class TestMigSharedUsedadm_create_user(MigTestCase,
+                                       FixtureAssertMixin,
+                                       PickleAssertMixin):
+    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'] = "user@example.com"
+        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
+
+        def _generate_salt():
+            return b'CCCC12344321CCCC'
+
+        expected_user_id = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/emailAddress=test@example.com'
+        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'] = "test@example.com"
+        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 done because changes
+        #       to make them assertion frienfly values will increase the size
+        #       of the diff which, at time of commit, are best minimised.
+        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"""