Adjust vgrid self-tests to run as legacy_main for reuse in unit tests. Extended

test coverage to more functions.

Author: jonasbardino

mig/shared/vgrid.py

@@ -2309,9 +2309,9 @@ def vgrid_rm_entry(configuration, vgrid):
     return (success, msg)
 
 
-if __name__ == "__main__":
+def legacy_main(_exit=sys.exit, _print=print):
     from mig.shared.conf import get_configuration_object
-    conf = get_configuration_object()
+    conf = get_configuration_object(skip_log=True, disable_auth_log=True)
     client_id = '/C=DK/CN=John Doe/emailAddress=john@doe.org'
     if sys.argv[1:]:
         client_id = sys.argv[1]
@@ -2432,3 +2432,7 @@ def vgrid_rm_entry(configuration, vgrid):
     if load_status:
         print(vgrid_set_members(conf, dummy_vgrid, orig_members))
     print(vgrid_members(dummy_vgrid, conf))
+
+
+if __name__ == "__main__":
+    legacy_main()

tests/test_mig_shared_vgrid.py

@@ -29,15 +29,19 @@
 
 import os
 import time
+import unittest
 
-from tests.support import MigTestCase, ensure_dirs_exist, testmain
 from mig.shared.base import client_id_dir
 from mig.shared.serial import dump
-from mig.shared.vgrid import vgrid_is_default, vgrid_exists, vgrid_list_vgrids, \
-    vgrid_add_entities, vgrid_remove_entities, vgrid_settings, \
-    vgrid_is_owner_or_member, vgrid_is_owner, vgrid_is_member, \
-    vgrid_list, vgrid_set_entities, get_vgrid_workflow_jobs, vgrid_add_owners, \
-    vgrid_add_members, vgrid_add_workflow_jobs
+from mig.shared.vgrid import get_vgrid_workflow_jobs, legacy_main, \
+    vgrid_add_entities, vgrid_add_members, vgrid_add_owners, \
+    vgrid_add_workflow_jobs, vgrid_allow_restrict_write, vgrid_exists, \
+    vgrid_is_default, vgrid_is_member, vgrid_is_owner, \
+    vgrid_is_owner_or_member, vgrid_list, vgrid_list_parents, \
+    vgrid_list_subvgrids, vgrid_list_vgrids, vgrid_match_resources, \
+    vgrid_nest_sep, vgrid_remove_entities, vgrid_restrict_write, \
+    vgrid_set_entities, vgrid_settings
+from tests.support import MigTestCase, ensure_dirs_exist, testmain
 
 
 class TestMigSharedVgrid(MigTestCase):
@@ -197,11 +201,13 @@ def test_vgrid_settings_inheritance(self):
     def test_vgrid_permission_checks(self):
         """Test owner/member permission verification"""
         # Setup owners and members
-        vgrid_add_entities(self.configuration, self.test_vgrid,
-                           'owners', [self.TEST_OWNER_DN])
+        added, msg = vgrid_add_entities(self.configuration, self.test_vgrid,
+                                        'owners', [self.TEST_OWNER_DN])
+        self.assertTrue(added, msg)
         time.sleep(0.1)
-        vgrid_add_entities(self.configuration, self.test_vgrid,
-                           'members', [self.TEST_MEMBER_DN])
+        added, msg = vgrid_add_entities(self.configuration, self.test_vgrid,
+                                        'members', [self.TEST_MEMBER_DN])
+        self.assertTrue(added, msg)
         time.sleep(0.1)
 
         # Verify owner permissions
@@ -230,14 +236,8 @@ def test_workflow_job_management(self):
                                self.TEST_OWNER_DIR)
         ensure_dirs_exist(job_dir)
         job_path = os.path.join(job_dir, '%s.mRSL' % self.TEST_JOB_ID)
-        # with open(job_path, 'w') as job_fd:
-        #    job_fd.write('\n')
         dump({'job_id': self.TEST_JOB_ID, 'EXECUTE': 'uptime'}, job_path)
 
-        print("DEBUG: wrote %r" % job_path)
-        # vgrid_add_workflow_jobs(self.configuration, self.test_vgrid,
-        #                        [self.TEST_JOB_ID])
-
         # Add job
         status, msg = vgrid_add_entities(self.configuration, self.test_vgrid,
                                          'jobqueue', [job_entry])
@@ -282,6 +282,155 @@ def test_workflow_job_management(self):
         self.assertTrue(status)
         self.assertEqual(len(jobs), 0)
 
+    def test_vgrid_list_subvgrids(self):
+        """Test retrieving subvgrids of given vgrid"""
+        status, subvgrids = vgrid_list_subvgrids(self.test_vgrid,
+                                                 self.configuration)
+        self.assertTrue(status)
+        self.assertEqual(subvgrids, [self.test_subvgrid])
+
+    def test_vgrid_list_parents(self):
+        """Test listing parent vgrids (root first)"""
+        parents = vgrid_list_parents(self.test_subvgrid, self.configuration)
+        expected = [self.test_vgrid]
+        self.assertEqual(parents, expected)
+
+    def test_vgrid_match_resources(self):
+        """Test resource filtering for vgrid"""
+        test_resources = ['res1', 'res2', 'invalid_res', self.TEST_RESOURCE_DN]
+        added, msg = vgrid_add_entities(self.configuration, self.test_vgrid,
+                                        'resources', [self.TEST_RESOURCE_DN])
+        self.assertTrue(added, msg)
+
+        matched = vgrid_match_resources(self.test_vgrid, test_resources,
+                                        self.configuration)
+        self.assertEqual(matched, [self.TEST_RESOURCE_DN])
+
+    # TODO: adjust API to allow enabling the next test
+    @unittest.skipIf(True, "requires read-only mount")
+    def test_vgrid_allow_restrict_write(self):
+        """Test write restriction validation logic"""
+        # Create parent-child structure
+        parent_write_setting = [('write_shared_files', 'none')]
+        added, msg = vgrid_add_entities(self.configuration, self.test_vgrid,
+                                        'settings', parent_write_setting)
+        self.assertTrue(added, msg)
+
+        # Child tries to set write_shared_files to "members"
+        result = vgrid_allow_restrict_write(self.test_subvgrid, 'members',
+                                            self.configuration,
+                                            auto_migrate=True)
+        self.assertFalse(result)
+
+        # Valid case: parent allows writes, child sets to "none"
+        parent_write_setting = [('write_shared_files', 'members')]
+        vgrid_set_entities(self.configuration, self.test_vgrid,
+                           'settings', parent_write_setting, True)
+
+        result = vgrid_allow_restrict_write(self.test_subvgrid, 'none',
+                                            self.configuration,
+                                            auto_migrate=True)
+        self.assertTrue(result)
+
+    # TODO: adjust API to allow enabling the next test
+    @unittest.skipIf(True, "requires read-only mount")
+    def test_vgrid_restrict_write(self):
+        """Test write restriction enforcement"""
+        # Setup test share
+        test_share = os.path.join(self.configuration.vgrid_files_home,
+                                  self.test_vgrid)
+        ensure_dirs_exist(test_share)
+
+        # Migrate to restricted mode
+        result = vgrid_restrict_write(self.test_vgrid, 'none',
+                                      self.configuration, auto_migrate=True)
+        self.assertTrue(result)
+
+        # Verify symlink points to readonly
+        flat_vgrid = self.test_vgrid.replace('/', vgrid_nest_sep)
+        read_path = os.path.join(self.configuration.vgrid_files_readonly,
+                                 flat_vgrid)
+        self.assertEqual(os.path.realpath(test_share),
+                         os.path.realpath(read_path))
+
+    def test_vgrid_settings_scopes(self):
+        """Test different vgrid settings lookup scopes"""
+        # Local settings only
+        local_settings = [('vgrid_name', self.test_subvgrid),
+                          ('description', 'test subvgrid'),
+                          ('write_shared_files', 'members'),
+                          ]
+        added, msg = vgrid_add_entities(self.configuration, self.test_subvgrid,
+                                        'settings', local_settings)
+        self.assertTrue(added, msg)
+
+        # Check recursive vs direct
+        status, recursive_settings = vgrid_settings(
+            self.test_subvgrid, self.configuration, recursive=True, as_dict=True)
+        self.assertTrue(status)
+        status, direct_settings = vgrid_settings(
+            self.test_subvgrid, self.configuration, recursive=False, as_dict=True)
+        self.assertTrue(status)
+        self.assertEqual(direct_settings['description'], 'test subvgrid')
+        self.assertIn('write_shared_files', recursive_settings)  # Inherited
+
+    def test_vgrid_add_owner_rank(self):
+        """Test owner ranking/ordering functionality"""
+        new_owner = '/C=DK/CN=New Owner'
+        added, msg = vgrid_add_owners(
+            self.configuration,
+            self.test_vgrid,
+            [new_owner],
+            rank=0  # Add as first owner
+        )
+        self.assertTrue(added, msg)
+
+        owners = vgrid_list(self.test_vgrid, 'owners', self.configuration)[1]
+        self.assertEqual(owners[0], new_owner)
+
+    # TODO: adjust API to allow enabling the next test
+    @unittest.skipIf(True, "requires tweaking of funcion")
+    def test_workflow_job_priority(self):
+        """Test workflow job queue ordering and limits"""
+        # Create max jobs + 1
+        job_entries = [
+            {'client_id': self.TEST_OWNER_DN, 'job_id': str(i)}
+            for i in range(101)
+        ]
+        added, msg = vgrid_add_workflow_jobs(
+            self.configuration,
+            self.test_vgrid,
+            job_entries
+        )
+        self.assertTrue(added, msg)
+
+        status, jobs = vgrid_list(
+            self.test_vgrid, 'jobqueue', self.configuration)
+        self.assertTrue(status)
+        # Should stay at max 100 by removing oldest
+        self.assertEqual(len(jobs), 100)
+        self.assertEqual(jobs[-1]['job_id'], '100')  # Newest at end
+
+
+class TestMigSharedVgrid__main(MigTestCase):
+    """Unit tests for vgrid self-checks"""
+
+    def test_existing_main(self):
+        def raise_on_error_exit(exit_code):
+            if exit_code != 0:
+                if raise_on_error_exit.last_print is not None:
+                    identifying_message = raise_on_error_exit.last_print
+                else:
+                    identifying_message = 'unknown'
+                raise AssertionError(
+                    'failure in unittest/testcore: %s' % (identifying_message,))
+        raise_on_error_exit.last_print = None
+
+        def record_last_print(value):
+            raise_on_error_exit.last_print = value
+
+        legacy_main(_exit=raise_on_error_exit, _print=record_last_print)
+
 
 if __name__ == '__main__':
     testmain()