Make createuser work for new installations and add basic coverage.

Allow user creation to provision the user_db_home directory as well as
the file itself which is the situation after config generation is run
to completion for a new installation.

Cover the very asic operation of createuser ensuring that a user db
and lock file are correctly create upon a call to create a test user.

Author: albu-diku

mig/server/createuser.py

@@ -91,8 +91,7 @@ def usage(name='createuser.py'):
 """ % {'name': name, 'cert_warn': cert_warn})
 
 
-if '__main__' == __name__:
-    (args, app_dir, db_path) = init_user_adm()
+def main(args, cwd, configuration=None, db_path=keyword_auto):
     conf_path = None
     auth_type = 'custom'
     expire = None
@@ -190,8 +189,11 @@ def usage(name='createuser.py'):
             if verbose:
                 print('using configuration from MIG_CONF (or default)')
 
-    configuration = get_configuration_object(config_file=conf_path)
+    if configuration is None:
+        configuration = get_configuration_object(config_file=conf_path)
+
     logger = configuration.logger
+
     # NOTE: we need explicit db_path lookup here for load_user_dict call
     if db_path == keyword_auto:
         db_path = default_db_path(configuration)
@@ -326,3 +328,7 @@ def usage(name='createuser.py'):
         if verbose:
             print('Cleaning up tmp file: %s' % user_file)
         os.remove(user_file)
+
+if __name__ == '__main__':
+    (args, cwd, db_path) = init_user_adm()
+    main(args, cwd, db_path=db_path)

mig/shared/accountstate.py

@@ -32,6 +32,7 @@
 from __future__ import print_function
 from __future__ import absolute_import
 
+from past.builtins import basestring
 import os
 import time
 

mig/shared/base.py

@@ -36,6 +36,7 @@
 import re
 
 # IMPORTANT: do not import any other MiG modules here - to avoid import loops
+from mig.shared.compat import PY2
 from mig.shared.defaults import default_str_coding, default_fs_coding, \
     keyword_all, keyword_auto, sandbox_names, _user_invisible_files, \
     _user_invisible_dirs, _vgrid_xgi_scripts, cert_field_order, csrf_field, \
@@ -294,7 +295,9 @@ def canonical_user(configuration, user_dict, limit_fields):
         if key == 'full_name':
             # IMPORTANT: we get utf8 coded bytes here and title() treats such
             # chars as word termination. Temporarily force to unicode.
-            val = force_utf8(force_unicode(val).title())
+            val = force_unicode(val).title()
+            if PY2:
+                val = force_utf8(val)
         elif key == 'email':
             val = val.lower()
         elif key == 'country':

mig/shared/useradm.py

@@ -30,7 +30,9 @@
 from __future__ import print_function
 from __future__ import absolute_import
 
+from past.builtins import basestring
 from email.utils import parseaddr
+import codecs
 import datetime
 import fnmatch
 import os
@@ -466,8 +468,24 @@ def create_user_in_db(configuration, db_path, client_id, user, now, authorized,
     flock = None
     user_db = {}
     renew = default_renew
+
+    retry_lock = False
     if do_lock:
+        try:
+            flock = lock_user_db(db_path)
+        except FileNotFoundError as e:
+            user_db_home = os.path.dirname(db_path)
+
+            if not os.path.exists(db_path) and not os.path.exists(user_db_home):
+                os.mkdir(user_db_home)
+                retry_lock = True
+            else:
+                raise Exception("Failed to lock user DB: '%s'" % db_path)
+
+    if retry_lock:
         flock = lock_user_db(db_path)
+        if not flock:
+            raise Exception("Failed to lock user DB: '%s'" % db_path)
 
     if not os.path.exists(db_path):
         # Auto-create missing user DB if either auto_create_db or force is set
@@ -858,7 +876,8 @@ def create_user_in_fs(configuration, client_id, user, now, renew, force, verbose
         # match in htaccess
 
         dn_plain = info['distinguished_name']
-        dn_enc = dn_plain.encode('string_escape')
+        dn_utf8_bytes = codecs.encode(dn_plain, 'utf8')
+        dn_enc = codecs.decode(dn_utf8_bytes, 'unicode_escape')
 
         def upper_repl(match):
             """Translate hex codes to upper case form"""

tests/support/__init__.py

@@ -40,6 +40,7 @@
 import sys
 from unittest import TestCase, main as testmain
 
+from tests.support.configsupp import FakeConfiguration
 from tests.support.suppconst import MIG_BASE, TEST_BASE, TEST_FIXTURE_DIR, \
     TEST_OUTPUT_DIR
 
@@ -48,6 +49,19 @@
 # force defaults to a local environment
 os.environ['MIG_ENV'] = 'local'
 
+# expose the configuraed environment as a constant
+MIG_ENV = os.environ['MIG_ENV']
+
+if MIG_ENV == 'local':
+    # force testconfig as the conig file path
+    is_py2 = PY2
+    _conf_dir_suffix = "-py%s" % ('2' if is_py2 else '3',)
+    _conf_dir = "testconfs%s" % (_conf_dir_suffix,)
+    _local_conf = os.path.join(MIG_BASE, 'envhelp/output', _conf_dir, 'MiGserver.conf')
+    _config_file = os.getenv('MIG_CONF', None)
+    if _config_file is None:
+        os.environ['MIG_CONF'] = _local_conf
+
 # All MiG related code will at some point include bits from the mig module
 # namespace. Rather than have this knowledge spread through every test file,
 # make the sole responsbility of test files to find the support file and
@@ -103,6 +117,7 @@ def __init__(self, *args):
         super(MigTestCase, self).__init__(*args)
         self._cleanup_checks = list()
         self._cleanup_paths = set()
+        self._configuration = None
         self._logger = None
         self._skip_logging = False
 
@@ -153,6 +168,18 @@ def _reset_logging(self, stream):
         root_handler = root_logger.handlers[0]
         root_handler.stream = stream
 
+    # testcase defaults
+
+    def _provide_configuration(self):
+        return FakeConfiguration()
+
+    @property
+    def configuration(self):
+        """Init a fake configuration if not already done"""
+        if self._configuration is None:
+            self._configuration = self._provide_configuration()
+        return self._configuration
+
     @property
     def logger(self):
         """Init a fake logger if not already done"""
@@ -199,6 +226,10 @@ def assertPathExists(self, relative_path):
         assert not os.path.isabs(
             relative_path), "expected relative path within output folder"
         absolute_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
+        return MigTestCase._absolute_path_kind(absolute_path)
+
+    @staticmethod
+    def _absolute_path_kind(absolute_path):
         stat_result = os.lstat(absolute_path)
         if stat.S_ISLNK(stat_result.st_mode):
             return "symlink"

tests/test_mig_server_createuser.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# test_mig_server-createuser - unit tests for the migrid createuser CLI
+# Copyright (C) 2003-2024  The MiG Project by the Science HPC Center at UCPH
+#
+# This file is part of MiG.
+#
+# MiG is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# MiG is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+#
+# --- END_HEADER ---
+#
+
+"""Unit tests for the migrid createuser CLI"""
+
+from __future__ import print_function
+import os
+import sys
+
+from tests.support import MIG_BASE, TEST_OUTPUT_DIR, MigTestCase, testmain
+
+from mig.shared.conf import get_configuration_object
+from mig.server.createuser import main as createuser
+
+class TestBooleans(MigTestCase):
+    def before_each(self):
+        configuration = self._provide_configuration()
+        test_user_db_home = os.path.join(configuration.state_path, 'user_db_home')
+        try:
+            os.rmdir(test_user_db_home)
+        except:
+            pass
+        self.expected_user_db_home = test_user_db_home
+
+    def _provide_configuration(self):
+        return get_configuration_object()
+
+
+    def test_user_db_is_created_and_user_is_added(self):
+        args = [
+            "-r",
+            "Test User",
+            "Test Org",
+            "NA",
+            "DK",
+            "dummy-user",
+            "This is the create comment",
+            "password"
+        ]
+        createuser(args, TEST_OUTPUT_DIR, configuration=self.configuration)
+
+        # presence of user home
+        path_kind = MigTestCase._absolute_path_kind(self.expected_user_db_home)
+        self.assertEqual(path_kind, 'dir')
+
+        # presence of user db
+        expected_user_db_file = os.path.join(self.expected_user_db_home, 'MiG-users.db')
+        path_kind = MigTestCase._absolute_path_kind(expected_user_db_file)
+        self.assertEqual(path_kind, 'file')
+
+
+if __name__ == '__main__':
+    testmain()