Add basic success case unit tests of the shared code dealing with peers.

Author: albu-diku

tests/fixture/MiG-users.db--example.binary renamed to tests/fixture/MiG-users.db--example.pickle

@@ -35,6 +35,7 @@
 import json
 import logging
 import os
+import pickle
 import shutil
 import stat
 import sys
@@ -257,6 +258,23 @@ def temppath(self, relative_path, **kwargs):
 
     # custom assertions available for common use
 
+    def assertDirEmpty(self, relative_path):
+        path_kind = self.assertPathExists(relative_path)
+        assert path_kind == "dir", "expected a directory but found %s" % (
+            path_kind, )
+        absolute_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
+        entries = os.listdir(absolute_path)
+        assert not entries, "directory is not empty"
+
+    def assertDirNotEmpty(self, relative_path):
+        path_kind = self.assertPathExists(relative_path)
+        assert path_kind == "dir", "expected a directory but found %s" % (
+            path_kind, )
+        absolute_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
+        entries = os.listdir(absolute_path)
+        assert entries, "directory is empty"
+        return [os.path.join(absolute_path, entry) for entry in entries]
+
     def assertFileContentIdentical(self, file_actual, file_expected):
         """Make sure file_actual and file_expected are identical"""
         with io.open(file_actual) as f_actual, io.open(file_expected) as f_expected:
@@ -285,9 +303,11 @@ def assertFileExists(self, relative_path):
 
     def assertPathExists(self, relative_path):
         """Make sure file in relative_path exists"""
-        assert not os.path.isabs(
-            relative_path), "expected relative path within output folder"
-        absolute_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
+        if os.path.isabs(relative_path):
+            self.assertPathWithin(relative_path, start=TEST_OUTPUT_DIR)
+            absolute_path = relative_path
+        else:
+            absolute_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
         return MigTestCase._absolute_path_kind(absolute_path)
 
     @staticmethod
@@ -368,6 +388,31 @@ def _fixture_copy_as_temp(testcase, fixture_format, fixture_data, fixture_path,
         shutil.copyfile(fixture_path, copied_fixture_file)
         return copied_fixture_file
 
+    @staticmethod
+    def _provision_test_user(self, distinguished_name):
+        # ensure a user home directory for our test user
+        conf_user_home = self.configuration.user_home[:-1]
+        from mig.shared.base import client_id_dir
+        test_client_dir_name = client_id_dir(distinguished_name)
+        test_user_dir = os.path.join(conf_user_home, test_client_dir_name)
+
+        # ensure a user db that includes our test user
+        conf_user_db_home = ensure_dirs_exist(self.configuration.user_db_home)
+        prepared_fixture = self.prepareFixtureAssert(
+            'MiG-users.db--example',
+            fixture_format='pickle',
+        )
+
+        test_db_file = prepared_fixture.copy_as_temp(prefix=conf_user_db_home)
+
+        # create the test user home directory
+        ensure_dirs_exist(test_user_dir)
+        # create the test user settings directory
+        user_settings_dir = os.path.join(self.configuration.user_settings, test_client_dir_name)
+        ensure_dirs_exist(user_settings_dir)
+
+        return test_user_dir
+
 
 def _to_display_path(value):
     """Convert a relative path to one to be shown as part of test output."""
@@ -419,7 +464,7 @@ def fixturefile(relative_path, fixture_format=None):
 
     data = None
 
-    if fixture_format == 'binary':
+    if fixture_format == 'binary' or fixture_format == 'pickle':
         with open(tmp_path, 'rb') as binfile:
             data = binfile.read()
     elif fixture_format == 'json':
@@ -428,6 +473,9 @@ def fixturefile(relative_path, fixture_format=None):
         raise AssertionError(
             "unsupported fixture format: %s" % (fixture_format,))
 
+    if fixture_format == 'pickle':
+        data = pickle.loads(data)
+
     return data, tmp_path
 
 

tests/support/__init__.py

@@ -0,0 +1,145 @@
+# -*- coding: utf-8 -*-
+#
+# --- BEGIN_HEADER ---
+#
+# test_mig_shared_accountreq - unit test of the corresponding mig lib module
+# 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 ---
+#
+
+"""Unit tests for the migrid module pointed to in the filename"""
+
+import datetime
+import os
+import pickle
+import sys
+import unittest
+
+from tests.support import MigTestCase, testmain, fixturefile, ensure_dirs_exist
+
+import mig.shared.accountreq as accountreq
+from mig.shared.base import canonical_user, fill_distinguished_name
+from mig.shared.defaults import keyword_auto
+
+
+class MigSharedAccountreq__peers(MigTestCase):
+
+    TEST_PEER_DN = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]'
+    TEST_USER_DN = '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]'
+
+    @property
+    def user_pending_dir(self):
+        return self.configuration.user_pending
+
+    def _load_pickled_peer(self, absolute_path):
+        self.assertPathWithin(absolute_path, start=self.user_pending_dir)
+        with open(absolute_path, 'rb') as pickle_file:
+            value = pickle.load(pickle_file)
+
+        def _string_if_bytes(value):
+            if isinstance(value, bytes):
+                return str(value, 'utf8')
+            else:
+                return value
+        return {_string_if_bytes(x): _string_if_bytes(y) for x, y in value.items()}
+
+    def _peer_dict_from_fixture(self):
+        fixture_data, fixture_path = fixturefile(
+            "MiG-users.db--example",
+            fixture_format="pickle"
+        )
+        raw_user = dict(fixture_data[self.TEST_USER_DN])
+        del raw_user["distinguished_name"]
+        del raw_user["unique_id"]
+        del raw_user["password"]
+        del raw_user["password_hash"]
+        raw_user["email"] = "[email protected]"
+
+        # avoid the need for the 'site_peers_explicit_fields' config option..
+        raw_user['comment'] = fixture_data[self.TEST_USER_DN]["email"]
+
+        user_dict = canonical_user(self.configuration, raw_user, raw_user.keys())
+        fill_distinguished_name(user_dict)
+        return user_dict
+
+    def _record_peer_acceptance(self, test_client_dir_name, peer_distinguished_name):
+        test_user_accepted_peers_file = os.path.join(self.configuration.user_settings, test_client_dir_name, "peers")
+        expire_tomorrow =  datetime.date.today() + datetime.timedelta(days=1)
+        with open(test_user_accepted_peers_file, "wb") as test_user_accepted_peers:
+            pickle.dump({peer_distinguished_name:{'expire': str(expire_tomorrow) }}, test_user_accepted_peers)
+
+    def _settings_dir_for_user(self, distinguished_name):
+        return os.path.join(self.configuration.user_settings, distinguished_name)
+
+    def _provide_configuration(self):
+        return 'testconfig'
+
+    def before_each(self):
+        ensure_dirs_exist(self.configuration.user_cache)
+        ensure_dirs_exist(self.configuration.user_pending)
+        ensure_dirs_exist(self.configuration.user_settings)
+        ensure_dirs_exist(self.configuration.mrsl_files_dir)
+        ensure_dirs_exist(self.configuration.resource_pending)
+
+    def test_a_new_peer(self):
+        # precondition
+        self.assertDirEmpty(self.configuration.user_pending)
+        request_dict = self._peer_dict_from_fixture()
+
+        success, _ = accountreq.save_account_request(self.configuration, request_dict)
+
+        # check that we have an output directory now
+        absolute_files = self.assertDirNotEmpty(self.user_pending_dir)
+        self.assertEqual(len(absolute_files), 1)
+        # check the saved peer
+        peer_user_dict = self._load_pickled_peer(absolute_files[0])
+        self.assertEqual(peer_user_dict, request_dict)
+
+    def test_listing_peers(self):
+        # precondition
+        self.assertDirEmpty(self.configuration.user_pending)
+        request_dict = self._peer_dict_from_fixture()
+        success, _ = accountreq.save_account_request(self.configuration, request_dict)
+
+        success, listing = accountreq.list_account_reqs(self.configuration)
+
+        self.assertTrue(success)
+        self.assertEqual(len(listing), 1)
+        peer_pickle_file = os.path.join(self.user_pending_dir, listing[0])
+        peer_pickle = self._load_pickled_peer(peer_pickle_file)
+        self.assertEqual(peer_pickle['distinguished_name'], self.TEST_PEER_DN)
+
+    def test_peer_acceptance(self):
+        test_client_dir = self._provision_test_user(self, self.TEST_USER_DN)
+        test_client_dir_name = os.path.basename(test_client_dir)
+        self._record_peer_acceptance(test_client_dir_name, self.TEST_PEER_DN)
+        self.assertDirEmpty(self.configuration.user_pending)
+        request_dict = self._peer_dict_from_fixture()
+        success, req_path = accountreq.save_account_request(self.configuration, request_dict)
+        arranged_req_id = os.path.basename(req_path)
+
+        success, message = accountreq.accept_account_req(arranged_req_id, self.configuration, keyword_auto)
+
+        self.assertTrue(success)
+
+
+if __name__ == '__main__':
+    testmain()

tests/test_mig_shared_accountreq.py

@@ -69,25 +69,7 @@ def _provide_configuration(self):
         return 'testconfig'
 
     def before_each(self):
-        # ensure a user home directory for our test user
-        conf_user_home = self.configuration.user_home[:-1]
-        test_client_dir = client_id_dir(self.TEST_CLIENT_ID)
-        test_user_dir = os.path.join(conf_user_home, test_client_dir)
-
-        # ensure a user db that includes our test user
-
-        conf_user_db_home = ensure_dirs_exist(self.configuration.user_db_home)
-        temppath(conf_user_db_home, self)
-        prepared_fixture = self.prepareFixtureAssert(
-            'MiG-users.db--example',
-            fixture_format='binary',
-        )
-
-        test_db_file = prepared_fixture.copy_as_temp(prefix=conf_user_db_home)
-
-        # create the test user home directory
-        self.test_user_dir = ensure_dirs_exist(test_user_dir)
-        temppath(self.test_user_dir, 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):

tests/test_mig_shared_functionality_cat.py

@@ -84,7 +84,7 @@ def before_each(self):
         temppath(conf_user_db_home, self)
         prepared_fixture = self.prepareFixtureAssert(
             "MiG-users.db--example",
-            fixture_format="binary",
+            fixture_format="pickle",
         )
 
         prepared_fixture.copy_as_temp(prefix=conf_user_db_home)