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 basic 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/compat.py

@@ -61,6 +61,13 @@ def _unicode_string_to_utf8_bytes(unicode_string):
     return codecs.encode(unicode_string, 'utf8')
 
 
+def _unicode_string_to_escaped_unicode(unicode_string):
+    """Convert utf8 bytes to escaped unicode string."""
+
+    utf8_bytes = dn_utf8_bytes = codecs.encode(unicode_string, 'utf8')
+    return codecs.decode(utf8_bytes, 'unicode_escape')
+
+
 def ensure_native_string(string_or_bytes):
     """Given a supplied input which can be either a string or bytes
     return a representation providing string operations while ensuring that

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
@@ -44,6 +46,7 @@
 from mig.shared.base import client_id_dir, client_dir_id, client_alias, \
     get_client_id, extract_field, fill_user, fill_distinguished_name, \
     is_gdp_user, mask_creds, sandbox_resource
+from mig.shared.compat import _unicode_string_to_escaped_unicode
 from mig.shared.conf import get_configuration_object
 from mig.shared.configuration import Configuration
 from mig.shared.defaults import user_db_filename, keyword_auto, ssh_conf_dir, \
@@ -466,8 +469,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
@@ -862,7 +881,7 @@ 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_enc = _unicode_string_to_escaped_unicode(dn_plain)
 
         def upper_repl(match):
             """Translate hex codes to upper case form"""

tests/test_mig_server_createuser.py

@@ -0,0 +1,75 @@
+# -*- 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.server.createuser import main as createuser
+
+class TestBooleans(MigTestCase):
+    def before_each(self):
+        configuration = self.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 'testconfig'
+
+    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()