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

tests/fixture/peer_user_dict.json

@@ -0,0 +1,14 @@
+{
+  "full_name": "Test Peer User",
+  "organization": "Test Org",
+  "state": "NA",
+  "country": "DK",
+  "email": "[email protected]",
+  "comment": "[email protected]",
+  "locality": "",
+  "organizational_unit": "",
+  "expire": 1758970812,
+  "created": 1727434813.0792377,
+  "openid_names": [],
+  "distinguished_name": "/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/[email protected]"
+}

tests/support/__init__.py

@@ -1,3 +1,3 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 #
@@ -35,6 +35,7 @@
 import json
 import logging
 import os
+import pickle
 import shutil
 import stat
 import sys
@@ -257,6 +258,25 @@
 
     # custom assertions available for common use
 
+    def assertDirEmpty(self, relative_path):
+        """Make sure the supplied path is an empty directory"""
+        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):
+        """Make sure the supplied path is a non-empty directory"""
+        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 +305,11 @@
 
     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 +390,34 @@
         shutil.copyfile(fixture_path, copied_fixture_file)
         return copied_fixture_file
 
+    @staticmethod
+    def _provision_test_user(self, distinguished_name):
+        """Provide a means to fabricate a useable test user on demand.
+        """
+
+        # ensure a user home directory for our test user
+        conf_user_home = os.path.normpath(self.configuration.user_home)
+        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 +469,7 @@
 
     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 +478,9 @@
         raise AssertionError(
             "unsupported fixture format: %s" % (fixture_format,))
 
+    if fixture_format == 'pickle':
+        data = pickle.loads(data)
+
     return data, tmp_path
 
 

tests/test_mig_shared_accountreq.py

@@ -0,0 +1,138 @@
+# -*- 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):
+    """Unit tests for peers related functions within the accountreq module"""
+
+    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_settings_dir(self):
+        return self.configuration.user_settings
+
+    @property
+    def user_pending_dir(self):
+        return self.configuration.user_pending
+
+    def _load_saved_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, _ = fixturefile("peer_user_dict", fixture_format="json")
+        assert fixture_data["distinguished_name"] == self.TEST_PEER_DN
+        return fixture_data
+
+    def _record_peer_acceptance(self, test_client_dir_name, peer_distinguished_name):
+        """Fabricate a peer acceptance record in a particular user settings dir.
+        """
+
+        test_user_accepted_peers_file = os.path.join(self.user_settings_dir, 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 _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_saved_peer(absolute_files[0])
+        self.assertEqual(peer_user_dict, request_dict)
+
+    def test_listing_peers(self):
+        # precondition
+        self.assertDirEmpty(self.user_pending_dir)
+        request_dict = self._peer_dict_from_fixture()
+        accountreq.save_account_request(self.configuration, request_dict)
+
+        success, listing = accountreq.list_account_reqs(self.configuration)
+
+        self.assertTrue(success)
+        self.assertEqual(len(listing), 1)
+        # check the fabricated peer was listed
+        peer_temp_file_name = listing[0]  # sadly listing returns _relative_ dirs
+        peer_pickle_file = os.path.join(self.user_pending_dir, peer_temp_file_name)
+        peer_pickle = self._load_saved_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.user_pending_dir)
+        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_functionality_cat.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 #
 # --- BEGIN_HEADER ---
@@ -69,25 +69,7 @@
         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_datatransfer.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 #
 # --- BEGIN_HEADER ---
@@ -74,31 +74,7 @@
         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",
-        )
-
-        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)
-
-        # ensure the user_settings home directory for our test user
-        conf_user_settings_home = ensure_dirs_exist(self.configuration.user_settings)
-        temppath(conf_user_settings_home, self)
-        test_user_settings_dir = os.path.join(conf_user_settings_home, test_client_dir)
-        ensure_dirs_exist(test_user_settings_dir)
-
+        self.test_user_dir = self._provision_test_user(self, self.TEST_CLIENT_ID)
         self.test_environ = create_http_environ(self.configuration)
 
     def test_default_disabled_site_transfer(self):