From a651b4c013951aaf8cf71f3e2831bc4cf1e6c3b3 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Wed, 22 Oct 2025 13:32:30 +0200 Subject: [PATCH 01/20] Fundamental test coverage of vgridaccess.py. --- tests/test_mig_shared_vgridaccess.py | 218 +++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 tests/test_mig_shared_vgridaccess.py diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py new file mode 100644 index 000000000..90099f2f9 --- /dev/null +++ b/tests/test_mig_shared_vgridaccess.py @@ -0,0 +1,218 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_vgridaccess - unit tests for vgridaccess helper functions +# 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 vgridaccess helper module""" + +import os +import time +import unittest + +from mig.shared.fileio import read_file, pickle +from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings +from mig.shared.vgridaccess import OWNERS, SETTINGS, VGRIDS, RESOURCES, \ + check_vgrid_access, force_update_resource_map, force_update_user_map, \ + force_update_vgrid_map, get_resource_map, get_vgrid_map, \ + load_resource_map, refresh_vgrid_map, vgrid_inherit_map +from tests.support import MigTestCase, ensure_dirs_exist, testmain + + +class TestMigSharedVgridAccess(MigTestCase): + """Unit tests for vgridaccess related helper functions""" + + TEST_OWNER_DN = \ + '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Owner/'\ + 'emailAddress=owner@example.org' + TEST_MEMBER_DN = \ + '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Member/'\ + 'emailAddress=member@example.org' + TEST_RESOURCE_ID = 'test.example.org.0' + + def _provide_configuration(self): + """Prepare isolated test config""" + return 'testconfig' + + def _create_vgrid(self, vgrid_name, owners=None, members=None, + resources=None, settings=None): + """Helper to create valid skeleton vgrid for testing""" + vgrid_path = os.path.join(self.configuration.vgrid_home, vgrid_name) + ensure_dirs_exist(vgrid_path) + if owners is None: + owners = [] + # Add vgrid owners + status, msg = vgrid_set_entities(self.configuration, vgrid_name, + 'owners', owners, allow_empty=True) + self.assertTrue(status, msg) + if members is not None: + status, msg = vgrid_set_entities(self.configuration, vgrid_name, + 'members', members, + allow_empty=False) + self.assertTrue(status, msg) + if resources is not None: + status, msg = vgrid_set_entities(self.configuration, vgrid_name, + 'resources', resources, + allow_empty=False) + self.assertTrue(status, msg) + if settings is not None: + status, msg = vgrid_set_entities(self.configuration, vgrid_name, + 'settings', settings, + allow_empty=False) + self.assertTrue(status, msg) + + def _create_resource(self, res_name, owners, config=None): + """Helper to create valid skeleton resource for testing""" + res_path = os.path.join(self.configuration.resource_home, res_name) + res_owners_path = os.path.join(res_path, 'owners') + res_config_path = os.path.join(res_path, 'config') + # Add resource skeleton with owners + ensure_dirs_exist(res_path) + if owners is None: + owners = [] + saved = pickle(owners, res_owners_path, self.logger) + self.assertTrue(saved) + if config is None: + config = {} + saved = pickle(config, res_config_path, self.logger) + self.assertTrue(saved) + + def before_each(self): + """Create test environment for vgridaccess tests""" + self._provision_test_user(self, self.TEST_OWNER_DN) + ensure_dirs_exist(self.configuration.mig_system_files) + ensure_dirs_exist(self.configuration.mig_system_run) + ensure_dirs_exist(self.configuration.user_home) + ensure_dirs_exist(self.configuration.user_settings) + ensure_dirs_exist(self.configuration.vgrid_home) + ensure_dirs_exist(self.configuration.resource_home) + # Start with empty maps + force_update_vgrid_map(self.configuration, clean=True) + force_update_user_map(self.configuration, clean=True) + force_update_resource_map(self.configuration, clean=True) + + self.test_vgrid = 'testvgrid' + + def test_vgrid_map_refresh(self): + """Verify vgrid map refresh captures changes""" + # We always init empty maps + initial_map = get_vgrid_map(self.configuration, recursive=False) + self.assertFalse(self.test_vgrid in initial_map.get(VGRIDS, {})) + + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + # Force refresh map + updated_map = force_update_vgrid_map(self.configuration) + vgrids = updated_map.get(VGRIDS, {}) + self.assertTrue(self.test_vgrid in vgrids) + self.assertEqual(vgrids[self.test_vgrid] + [OWNERS], [self.TEST_OWNER_DN]) + + def test_user_map_access(self): + """Test user permissions through cached access maps""" + # Add user as member + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + members=[self.TEST_MEMBER_DN]) + force_update_vgrid_map(self.configuration) + # Verify member access + allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, + self.test_vgrid) + self.assertTrue(allowed) + + def test_resource_map_update(self): + """Verify resource visibility in cache""" + # Check cached resource map does not yet contain entry + cached_map, map_stamp = load_resource_map(self.configuration, + caching=True) + self.assertFalse(cached_map, map_stamp) + self.assertFalse(self.TEST_RESOURCE_ID in cached_map) + + # Add vgrid with assigned resource + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + resources=[self.TEST_RESOURCE_ID]) + updated_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + print("DEBUG: updated vgrid map %s" % updated_vgrid_map) + # Check vgrid map contains resource entry + vgrid_data = updated_vgrid_map.get(VGRIDS, {}) + top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) + + # Check resource map contains resource entry + updated_res_map = force_update_resource_map(self.configuration, + clean=True) + print("DEBUG: updated res map %s" % updated_res_map) + # Check resource map contains entry + self.assertTrue(self.TEST_RESOURCE_ID in updated_res_map) + + def test_settings_inheritance(self): + """Test inherited settings propagation through cached maps""" + # Create top and sub vgrids with 'hidden' setting on top vgrid + top_settings = [('vgrid_name', self.test_vgrid), + ('hidden', True)] + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + settings=top_settings) + sub_vgrid = os.path.join(self.test_vgrid, 'subvgrid') + self._create_vgrid(sub_vgrid) + + # Force refresh of cached map + updated_map = force_update_vgrid_map(self.configuration) + + # Retrieve vgrid data from cached map + vgrid_data = updated_map.get(VGRIDS, {}) + self.assertTrue(vgrid_data) + + # Retrieve top vgrid settings from cached map + top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) + self.assertTrue(top_vgrid_data) + # Convert settings list of tuples to dict + top_settings_dict = dict(top_vgrid_data.get(SETTINGS, [])) + self.assertTrue(top_settings_dict) + + # Verify hidden setting in cache + self.assertEqual(top_settings_dict.get('hidden'), True) + + # Retrieve sub vgrid settings from cached map + sub_vgrid_data = vgrid_data.get(sub_vgrid, {}) + # Convert settings list of tuples to dict + sub_settings_dict = dict(sub_vgrid_data.get(SETTINGS, [])) + + # Verify hidden setting unset without inheritance + self.assertFalse(sub_settings_dict.get('hidden')) + + inherited_map = vgrid_inherit_map(self.configuration, updated_map) + vgrid_data = inherited_map.get(VGRIDS, {}) + self.assertTrue(vgrid_data) + + # Retrieve sub vgrid settings from cached map + sub_vgrid_data = vgrid_data.get(sub_vgrid, {}) + # Convert settings list of tuples to dict + sub_settings_dict = dict(sub_vgrid_data.get(SETTINGS, [])) + + # Verify hidden setting inheritance + self.assertEqual(sub_settings_dict.get('hidden'), True) + + +if __name__ == '__main__': + testmain() From dda151da3f9812105388af2b81f77b55c8fc4cee Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Wed, 22 Oct 2025 13:37:55 +0200 Subject: [PATCH 02/20] Sort imports and remove debug. --- tests/test_mig_shared_vgridaccess.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 90099f2f9..64c9c85aa 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -30,9 +30,9 @@ import time import unittest -from mig.shared.fileio import read_file, pickle +from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings -from mig.shared.vgridaccess import OWNERS, SETTINGS, VGRIDS, RESOURCES, \ +from mig.shared.vgridaccess import OWNERS, RESOURCES, SETTINGS, VGRIDS, \ check_vgrid_access, force_update_resource_map, force_update_user_map, \ force_update_vgrid_map, get_resource_map, get_vgrid_map, \ load_resource_map, refresh_vgrid_map, vgrid_inherit_map @@ -152,7 +152,6 @@ def test_resource_map_update(self): resources=[self.TEST_RESOURCE_ID]) updated_vgrid_map = force_update_vgrid_map(self.configuration, clean=True) - print("DEBUG: updated vgrid map %s" % updated_vgrid_map) # Check vgrid map contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) @@ -162,7 +161,6 @@ def test_resource_map_update(self): # Check resource map contains resource entry updated_res_map = force_update_resource_map(self.configuration, clean=True) - print("DEBUG: updated res map %s" % updated_res_map) # Check resource map contains entry self.assertTrue(self.TEST_RESOURCE_ID in updated_res_map) From f9160302f6c6d8798fcbc7e2a37a925298d73264 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 09:54:00 +0200 Subject: [PATCH 03/20] Extend coverage of vgridaccess.py module. --- tests/test_mig_shared_vgridaccess.py | 166 ++++++++++++++++++++++++++- 1 file changed, 162 insertions(+), 4 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 64c9c85aa..35fda198d 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -32,23 +32,32 @@ from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings -from mig.shared.vgridaccess import OWNERS, RESOURCES, SETTINGS, VGRIDS, \ - check_vgrid_access, force_update_resource_map, force_update_user_map, \ - force_update_vgrid_map, get_resource_map, get_vgrid_map, \ - load_resource_map, refresh_vgrid_map, vgrid_inherit_map +from mig.shared.vgridaccess import CONF, MEMBERS, OWNERS, RESOURCES, SETTINGS, \ + USERID, VGRIDS, check_vgrid_access, force_update_resource_map, \ + force_update_user_map, force_update_vgrid_map, get_resource_map, \ + get_user_map, get_vgrid_map, load_resource_map, refresh_vgrid_map, \ + user_vgrid_access, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain class TestMigSharedVgridAccess(MigTestCase): """Unit tests for vgridaccess related helper functions""" + TEST_USER_DN = \ + '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test User/'\ + 'emailAddress=test@example.com' TEST_OWNER_DN = \ '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Owner/'\ 'emailAddress=owner@example.org' TEST_MEMBER_DN = \ '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Member/'\ 'emailAddress=member@example.org' + TEST_OUTSIDER_DN = \ + '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Outsider/'\ + 'emailAddress=outsider@example.com' TEST_RESOURCE_ID = 'test.example.org.0' + TEST_OWNER_UUID = 'ff326a2b984828d9b32077c9b0b35a05' + TEST_USER_UUID = '707a2213995b4fb385793b5a7cb82a18' def _provide_configuration(self): """Prepare isolated test config""" @@ -211,6 +220,155 @@ def test_settings_inheritance(self): # Verify hidden setting inheritance self.assertEqual(sub_settings_dict.get('hidden'), True) + def test_user_map_fields(self): + """Verify user map includes complete profile/settings data""" + # First add a couple of test users + self._provision_test_user(self, self.TEST_OWNER_DN) + self._provision_test_user(self, self.TEST_USER_DN) + # Force fresh user map + initial_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + user_map = force_update_user_map(self.configuration) + test_owner = user_map.get(self.TEST_OWNER_DN, {}) + self.assertEqual(test_owner.get(USERID), self.TEST_OWNER_UUID) + self.assertTrue(isinstance(test_owner.get(CONF), dict)) + test_user = user_map.get(self.TEST_USER_DN, {}) + self.assertEqual(test_user.get(USERID), self.TEST_USER_UUID) + self.assertTrue(isinstance(test_user.get(CONF), dict)) + + def test_resource_revoked_access(self): + """Verify resource removal propagates through cached maps""" + # First add resource and vgrid + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + resources=[self.TEST_RESOURCE_ID]) + + initial_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + # Check vgrid map contains resource entry + vgrid_data = initial_vgrid_map.get(VGRIDS, {}) + top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) + + # Check resource map contains resource entry + initial_map = force_update_resource_map(self.configuration, + clean=True) + self.assertIn(self.TEST_RESOURCE_ID, initial_map) + + # Remove resource assignment from vgrid + status, msg = vgrid_set_entities(self.configuration, self.test_vgrid, + 'resources', [], allow_empty=True) + self.assertTrue(status, msg) + + updated_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + # Check vgrid map no longer contains resource entry + vgrid_data = updated_vgrid_map.get(VGRIDS, {}) + top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + self.assertFalse(self.TEST_RESOURCE_ID in top_vgrid_res) + + # Verify resource still in resource map + updated_map = force_update_resource_map(self.configuration, + clean=True) + self.assertIn(self.TEST_RESOURCE_ID, updated_map) + + def test_non_recursive_inheritance(self): + """Verify non-recursive map excludes nested vgrids""" + # Create parent+child vgrids + parent_vgrid = 'parent' + self._create_vgrid(parent_vgrid, [self.TEST_OWNER_DN]) + child_vgrid = os.path.join(parent_vgrid, 'child') + self._create_vgrid(child_vgrid, None, [self.TEST_MEMBER_DN]) + + # Force update to avoid auto caching and get non-recursive map + vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + vgrid_map = get_vgrid_map(self.configuration, recursive=False) + # Parent should appear + self.assertIn(parent_vgrid, vgrid_map.get(VGRIDS, {})) + # Child should still appear when non-recursive but just not inherit + self.assertIn(child_vgrid, vgrid_map.get(VGRIDS, {})) + # Check owners and members to verify they aren't inherited + self.assertEqual(vgrid_map[VGRIDS][parent_vgrid][OWNERS], + [self.TEST_OWNER_DN]) + self.assertEqual(len(vgrid_map[VGRIDS][parent_vgrid][MEMBERS]), 0) + self.assertEqual(len(vgrid_map[VGRIDS][child_vgrid][OWNERS]), 0) + self.assertEqual(vgrid_map[VGRIDS][child_vgrid][MEMBERS], + [self.TEST_MEMBER_DN]) + + def test_hidden_setting_propagation(self): + """Verify hidden=True propagates to not infect parent settings""" + parent_vgrid = 'parent' + self._create_vgrid(parent_vgrid, [self.TEST_OWNER_DN]) + child_vgrid = os.path.join(parent_vgrid, 'child') + self._create_vgrid(child_vgrid, [self.TEST_OWNER_DN], + settings=[('vgrid_name', child_vgrid), + ('hidden', True)]) + + # Verify parent remains visible in cache + updated_map = force_update_vgrid_map(self.configuration) + parent_data = updated_map.get(VGRIDS, {}).get(parent_vgrid, {}) + parent_settings = dict(parent_data.get(SETTINGS, [])) + self.assertNotEqual(parent_settings.get('hidden'), True) + + def test_default_vgrid_access(self): + """Verify special access rules for default vgrid""" + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + members=[self.TEST_MEMBER_DN]) + + initial_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + + # Even non-member should have access to default vgrid + participant = check_vgrid_access(self.configuration, + self.TEST_OUTSIDER_DN, + 'Generic') + self.assertFalse(participant) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_OUTSIDER_DN) + self.assertTrue('Generic' in allowed_vgrids) + + # Invalid vgrid should not allow any participation or access + participant = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, + 'invalid-vgrid-name') + self.assertFalse(participant) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_MEMBER_DN) + self.assertFalse('invalid-vgrid-name' in allowed_vgrids) + + def test_general_vgrid_access(self): + """Verify general access rules for vgrids""" + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + members=[self.TEST_MEMBER_DN]) + + initial_vgrid_map = force_update_vgrid_map(self.configuration, + clean=True) + + # Test vgrid must allow owner and members access + allowed = check_vgrid_access(self.configuration, self.TEST_OWNER_DN, + self.test_vgrid) + self.assertTrue(allowed) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_OWNER_DN) + self.assertTrue(self.test_vgrid in allowed_vgrids) + + allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, + self.test_vgrid) + self.assertTrue(allowed) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_MEMBER_DN) + self.assertTrue(self.test_vgrid in allowed_vgrids) + + # Test vgrid must reject allow outsider access + allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, + self.test_vgrid) + self.assertFalse(allowed) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_OUTSIDER_DN) + self.assertFalse(self.test_vgrid in allowed_vgrids) + if __name__ == '__main__': testmain() From ab69337d335051425fa106e4f8d8d45537e70af0 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 19:45:39 +0200 Subject: [PATCH 04/20] Increase coverage. --- tests/test_mig_shared_vgridaccess.py | 70 ++++++++++++++++++++++++++-- 1 file changed, 67 insertions(+), 3 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 35fda198d..06031bdc1 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -33,9 +33,10 @@ from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings from mig.shared.vgridaccess import CONF, MEMBERS, OWNERS, RESOURCES, SETTINGS, \ - USERID, VGRIDS, check_vgrid_access, force_update_resource_map, \ - force_update_user_map, force_update_vgrid_map, get_resource_map, \ - get_user_map, get_vgrid_map, load_resource_map, refresh_vgrid_map, \ + USERID, VGRIDS, check_vgrid_access, check_vgrids_modified, \ + force_update_resource_map, force_update_user_map, force_update_vgrid_map, \ + get_re_provider_map, get_resource_map, get_user_map, get_vgrid_map, \ + load_resource_map, refresh_vgrid_map, resources_using_re, unmap_vgrid, \ user_vgrid_access, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -56,8 +57,10 @@ class TestMigSharedVgridAccess(MigTestCase): '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Outsider/'\ 'emailAddress=outsider@example.com' TEST_RESOURCE_ID = 'test.example.org.0' + TEST_OWNER_UUID = 'ff326a2b984828d9b32077c9b0b35a05' TEST_USER_UUID = '707a2213995b4fb385793b5a7cb82a18' + TEST_RESOURCE_ALIAS = '0835f310d6422c36e33eeb7d0d3e9cf5' def _provide_configuration(self): """Prepare isolated test config""" @@ -369,6 +372,67 @@ def test_general_vgrid_access(self): self.TEST_OUTSIDER_DN) self.assertFalse(self.test_vgrid in allowed_vgrids) + def test_get_re_provider_map(self): + """Test RE provider map includes test resource""" + test_re = 'Python' + res_config = {'RUNTIMEENVIRONMENT': [(test_re, '/python/path')]} + self._create_resource(self.TEST_RESOURCE_ID, [ + self.TEST_OWNER_DN], res_config) + + # Update maps to include new resource + force_update_resource_map(self.configuration) + + # Verify RE appears in provider mapping + re_map = get_re_provider_map(self.configuration) + self.assertIn(test_re, re_map) + self.assertIn(self.TEST_RESOURCE_ALIAS, re_map[test_re]) + + def test_resources_using_re(self): + """Test finding resources with specific runtime environment""" + test_re = 'Bash' + res_config = {'RUNTIMEENVIRONMENT': [(test_re, '/bash/path')]} + self._create_resource(self.TEST_RESOURCE_ID, [ + self.TEST_OWNER_DN], res_config) + + # Refresh resource map + force_update_resource_map(self.configuration) + + # Verify resource appears in RE-specific results + res_list = resources_using_re(self.configuration, test_re) + self.assertIn(self.TEST_RESOURCE_ALIAS, res_list) + + def test_unmap_vgrid(self): + """Verify unmapping marks vgrid modified for update in cached data""" + mod_list, mod_stamp = check_vgrids_modified(self.configuration) + self.assertNotIn(self.test_vgrid, mod_list) + + # Unmap and verify mark modified + unmap_vgrid(self.configuration, self.test_vgrid) + + mod_list, mod_stamp = check_vgrids_modified(self.configuration) + self.assertIn(self.test_vgrid, mod_list) + + def test_access_nonexistent_vgrid(self): + """Ensure checks fail cleanly for non-existent vgrid""" + allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, + 'no-such-vgrid') + self.assertFalse(allowed) + + # Should not appear in allowed vgrids + allowed_vgrids = user_vgrid_access( + self.configuration, self.TEST_MEMBER_DN) + self.assertFalse('no-such-vgrid' in allowed_vgrids) + + def test_empty_member_access(self): + """Verify members-only vgrid rejects outsiders""" + self._create_vgrid(self.test_vgrid, [], [self.TEST_MEMBER_DN]) + force_update_vgrid_map(self.configuration) + + # Outsider should be blocked despite no owners + allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, + self.test_vgrid) + self.assertFalse(allowed) + if __name__ == '__main__': testmain() From 977af0a77fe42d82a914a8253a5a5c7c0d01f233 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 22:32:23 +0200 Subject: [PATCH 05/20] Add better coverage of fundamental function use and clean up some unnecessary clean function arguments. --- tests/test_mig_shared_vgridaccess.py | 159 +++++++++++++++++++++++---- 1 file changed, 136 insertions(+), 23 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 06031bdc1..fb0eaca73 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -36,8 +36,10 @@ USERID, VGRIDS, check_vgrid_access, check_vgrids_modified, \ force_update_resource_map, force_update_user_map, force_update_vgrid_map, \ get_re_provider_map, get_resource_map, get_user_map, get_vgrid_map, \ - load_resource_map, refresh_vgrid_map, resources_using_re, unmap_vgrid, \ - user_vgrid_access, vgrid_inherit_map + load_resource_map, mark_vgrid_modified, refresh_resource_map, \ + refresh_user_map, refresh_vgrid_map, res_vgrid_access, \ + reset_vgrids_modified, resources_using_re, unmap_vgrid, user_vgrid_access, \ + vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -111,7 +113,6 @@ def _create_resource(self, res_name, owners, config=None): def before_each(self): """Create test environment for vgridaccess tests""" - self._provision_test_user(self, self.TEST_OWNER_DN) ensure_dirs_exist(self.configuration.mig_system_files) ensure_dirs_exist(self.configuration.mig_system_run) ensure_dirs_exist(self.configuration.user_home) @@ -125,6 +126,128 @@ def before_each(self): self.test_vgrid = 'testvgrid' + def test_force_update_user_map(self): + """Simple test that user map refresh completes""" + user_map_before = get_user_map(self.configuration) + self.assertFalse(user_map_before) + self._provision_test_user(self, self.TEST_USER_DN) + updated_users = force_update_user_map(self.configuration) + self.assertTrue(updated_users) + self.assertNotEqual(user_map_before, updated_users) + + def test_force_update_resource_map(self): + """Simple test that resource map refresh completes""" + res_map_before = get_resource_map(self.configuration) + self.assertFalse(res_map_before) + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + updated_res = force_update_resource_map(self.configuration) + self.assertTrue(updated_res) + self.assertNotEqual(len(res_map_before), len(updated_res)) + + def test_force_update_vgrid_map(self): + """Simple test that vgrid map refresh completes""" + # Only (implicit) default vgrid in vgrid map before init + vgrid_map_before = get_vgrid_map(self.configuration) + self.assertEqual(len(vgrid_map_before.get(VGRIDS, [])), 1) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + updated_vgrid = force_update_vgrid_map(self.configuration) + self.assertTrue(updated_vgrid) + self.assertNotEqual(len(vgrid_map_before.get(VGRIDS, [])), + len(updated_vgrid.get(VGRIDS, []))) + + def test_refresh_user_map(self): + """Minimal test for user map refresh functionality""" + self._provision_test_user(self, self.TEST_USER_DN) + user_map = refresh_user_map(self.configuration) + self.assertTrue(user_map) + self.assertIn(self.TEST_USER_DN, user_map) + + def test_refresh_resource_map(self): + """Minimal test for resource map refresh functionality""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + res_map = refresh_resource_map(self.configuration) + self.assertTrue(res_map) + self.assertIn(self.TEST_RESOURCE_ID, res_map) + + def test_refresh_vgrid_map(self): + """Minimal test for vgrid map refresh functionality""" + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + vgrid_map = refresh_vgrid_map(self.configuration) + self.assertTrue(vgrid_map) + self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) + + def test_get_user_map(self): + """Minimal test for user map refresh functionality""" + self._provision_test_user(self, self.TEST_USER_DN) + force_update_user_map(self.configuration) + user_map = get_user_map(self.configuration) + self.assertTrue(user_map) + self.assertIn(self.TEST_USER_DN, user_map) + + def test_get_resource_map(self): + """Minimal test for user map refresh functionality""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + force_update_resource_map(self.configuration) + resource_map = get_resource_map(self.configuration) + self.assertTrue(resource_map) + self.assertIn(self.TEST_RESOURCE_ID, resource_map) + + def test_get_vgrid_map(self): + """Minimal test for user map refresh functionality""" + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + vgrid_map = get_vgrid_map(self.configuration) + self.assertTrue(vgrid_map) + self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) + + def test_check_vgrids_modified(self): + """Minimal test for vgrid modified tracking""" + # Initially ALL marked modified until cache init + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, ['ALL']) + # Reset and check ALL gone + reset_vgrids_modified(self.configuration) + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, []) + # Mark modified + mark_vgrid_modified(self.configuration, self.test_vgrid) + modified, stamp = check_vgrids_modified(self.configuration) + self.assertIn(self.test_vgrid, modified) + # Reset and check gone again + reset_vgrids_modified(self.configuration) + modified, stamp = check_vgrids_modified(self.configuration) + self.assertNotIn(self.test_vgrid, modified) + + def test_user_vgrid_access(self): + """Minimal test for user vgrid participation""" + # Start with global access to default vgrid + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_USER_DN) + self.assertTrue('Generic' in allowed_vgrids) + self.assertTrue(len(allowed_vgrids), 1) + # Create private vgrid + self._provision_test_user(self, self.TEST_OWNER_DN) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + # Refresh maps to reflect new content + force_update_vgrid_map(self.configuration) + allowed_vgrids = user_vgrid_access(self.configuration, + self.TEST_OWNER_DN) + self.assertIn(self.test_vgrid, allowed_vgrids) + + def test_res_vgrid_access(self): + """Minimal test for resource vgrid participation""" + # Only Generic access initially + allowed_vgrids = res_vgrid_access( + self.configuration, self.TEST_RESOURCE_ID) + self.assertEqual(allowed_vgrids, ['Generic']) + # Add to vgrid + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, resources=[self.TEST_RESOURCE_ID]) + # Refresh maps to reflect new content + force_update_vgrid_map(self.configuration) + allowed = res_vgrid_access(self.configuration, self.TEST_RESOURCE_ID) + self.assertIn(self.test_vgrid, allowed) + def test_vgrid_map_refresh(self): """Verify vgrid map refresh captures changes""" # We always init empty maps @@ -162,8 +285,7 @@ def test_resource_map_update(self): self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) - updated_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + updated_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) @@ -171,8 +293,7 @@ def test_resource_map_update(self): self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) # Check resource map contains resource entry - updated_res_map = force_update_resource_map(self.configuration, - clean=True) + updated_res_map = force_update_resource_map(self.configuration) # Check resource map contains entry self.assertTrue(self.TEST_RESOURCE_ID in updated_res_map) @@ -229,8 +350,7 @@ def test_user_map_fields(self): self._provision_test_user(self, self.TEST_OWNER_DN) self._provision_test_user(self, self.TEST_USER_DN) # Force fresh user map - initial_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + initial_vgrid_map = force_update_vgrid_map(self.configuration) user_map = force_update_user_map(self.configuration) test_owner = user_map.get(self.TEST_OWNER_DN, {}) self.assertEqual(test_owner.get(USERID), self.TEST_OWNER_UUID) @@ -246,8 +366,7 @@ def test_resource_revoked_access(self): self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) - initial_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + initial_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map contains resource entry vgrid_data = initial_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) @@ -255,8 +374,7 @@ def test_resource_revoked_access(self): self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) # Check resource map contains resource entry - initial_map = force_update_resource_map(self.configuration, - clean=True) + initial_map = force_update_resource_map(self.configuration) self.assertIn(self.TEST_RESOURCE_ID, initial_map) # Remove resource assignment from vgrid @@ -264,8 +382,7 @@ def test_resource_revoked_access(self): 'resources', [], allow_empty=True) self.assertTrue(status, msg) - updated_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + updated_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map no longer contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) @@ -273,8 +390,7 @@ def test_resource_revoked_access(self): self.assertFalse(self.TEST_RESOURCE_ID in top_vgrid_res) # Verify resource still in resource map - updated_map = force_update_resource_map(self.configuration, - clean=True) + updated_map = force_update_resource_map(self.configuration) self.assertIn(self.TEST_RESOURCE_ID, updated_map) def test_non_recursive_inheritance(self): @@ -286,8 +402,7 @@ def test_non_recursive_inheritance(self): self._create_vgrid(child_vgrid, None, [self.TEST_MEMBER_DN]) # Force update to avoid auto caching and get non-recursive map - vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + vgrid_map = force_update_vgrid_map(self.configuration) vgrid_map = get_vgrid_map(self.configuration, recursive=False) # Parent should appear self.assertIn(parent_vgrid, vgrid_map.get(VGRIDS, {})) @@ -321,8 +436,7 @@ def test_default_vgrid_access(self): self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) - initial_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + initial_vgrid_map = force_update_vgrid_map(self.configuration) # Even non-member should have access to default vgrid participant = check_vgrid_access(self.configuration, @@ -346,8 +460,7 @@ def test_general_vgrid_access(self): self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) - initial_vgrid_map = force_update_vgrid_map(self.configuration, - clean=True) + initial_vgrid_map = force_update_vgrid_map(self.configuration) # Test vgrid must allow owner and members access allowed = check_vgrid_access(self.configuration, self.TEST_OWNER_DN, From ab002450cb104a34808ac7be8ac1da4f4c0a29a4 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 22:58:50 +0200 Subject: [PATCH 06/20] Make sure resources are provisioned with a valid minimal conf to get picked up by the traversal in map updates. Add basic load_X_map tests. --- tests/test_mig_shared_vgridaccess.py | 46 +++++++++++++++++++++++++--- 1 file changed, 41 insertions(+), 5 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index fb0eaca73..b15c0ba73 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -36,10 +36,10 @@ USERID, VGRIDS, check_vgrid_access, check_vgrids_modified, \ force_update_resource_map, force_update_user_map, force_update_vgrid_map, \ get_re_provider_map, get_resource_map, get_user_map, get_vgrid_map, \ - load_resource_map, mark_vgrid_modified, refresh_resource_map, \ - refresh_user_map, refresh_vgrid_map, res_vgrid_access, \ - reset_vgrids_modified, resources_using_re, unmap_vgrid, user_vgrid_access, \ - vgrid_inherit_map + load_resource_map, load_user_map, load_vgrid_map, mark_vgrid_modified, \ + refresh_resource_map, refresh_user_map, refresh_vgrid_map, \ + res_vgrid_access, reset_vgrids_modified, resources_using_re, unmap_vgrid, \ + user_vgrid_access, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -107,7 +107,8 @@ def _create_resource(self, res_name, owners, config=None): saved = pickle(owners, res_owners_path, self.logger) self.assertTrue(saved) if config is None: - config = {} + # Make sure conf has one valid field + config = {'HOSTURL': res_name} saved = pickle(config, res_config_path, self.logger) self.assertTrue(saved) @@ -200,6 +201,41 @@ def test_get_vgrid_map(self): self.assertTrue(vgrid_map) self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) + def test_load_user_map(self): + """Basic test for direct user map loading""" + # Get empty map initially + user_map, map_stamp = load_user_map(self.configuration) + self.assertEqual(user_map, {}) + # Add test user + self._provision_test_user(self, self.TEST_USER_DN) + force_update_user_map(self.configuration) + # Verify updated map contains user + updated_map, updated_stamp = load_user_map(self.configuration) + self.assertIn(self.TEST_USER_DN, updated_map) + + def test_load_resource_map(self): + """Basic test for direct resource map loading""" + # Get empty map initially + res_map, map_stamp = load_resource_map(self.configuration) + self.assertEqual(res_map, {}) + # Add test resource + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + force_update_resource_map(self.configuration) + # Verify updated map contains resource after refresh + res_map, map_stamp = load_resource_map(self.configuration) + self.assertIn(self.TEST_RESOURCE_ID, res_map) + + def test_load_vgrid_map(self): + """Basic test for direct vgrid map loading""" + # Get map with at least 'Generic' vgrid + vgrid_map, map_stamp = load_vgrid_map(self.configuration) + self.assertIn('Generic', vgrid_map.get(VGRIDS, {})) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + # Verify updated map contains vgrid after refresh + vgrid_map, map_stamp = load_vgrid_map(self.configuration) + self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, {})) + def test_check_vgrids_modified(self): """Minimal test for vgrid modified tracking""" # Initially ALL marked modified until cache init From ee2837fa3095dfe6a299d61c7a57260838b01bea Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 23:16:13 +0200 Subject: [PATCH 07/20] Add a few more very basic tests. --- tests/test_mig_shared_vgridaccess.py | 62 ++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index b15c0ba73..0b3096f6a 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -236,6 +236,39 @@ def test_load_vgrid_map(self): vgrid_map, map_stamp = load_vgrid_map(self.configuration) self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, {})) + def test_check_vgrids_modified_initial(self): + """Verify initial state of modified vgrids list is empty""" + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, ['ALL']) + reset_vgrids_modified(self.configuration) + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, []) + + def test_resources_using_re_notfound(self): + """Test RE with no assigned resources returns empty list""" + # Nonexistent RE should have no resources + res_list = resources_using_re(self.configuration, 'NoSuchRE') + self.assertEqual(res_list, []) + + def test_vgrid_inherit_map_single(self): + """Test inheritance mapping with single vgrid""" + test_settings = [('vgrid_name', self.test_vgrid), + ('hidden', True)] + test_map = { + VGRIDS: { + self.test_vgrid: { + SETTINGS: test_settings, + OWNERS: [self.TEST_OWNER_DN] + } + } + } + inherited_map = vgrid_inherit_map(self.configuration, test_map) + vgrid_data = inherited_map.get(VGRIDS, {}) + self.assertTrue(self.test_vgrid in vgrid_data) + settings_dict = dict(vgrid_data[self.test_vgrid][SETTINGS]) + self.assertIs(type(settings_dict), dict) + self.assertEqual(settings_dict.get('hidden'), True) + def test_check_vgrids_modified(self): """Minimal test for vgrid modified tracking""" # Initially ALL marked modified until cache init @@ -561,6 +594,35 @@ def test_unmap_vgrid(self): mod_list, mod_stamp = check_vgrids_modified(self.configuration) self.assertIn(self.test_vgrid, mod_list) + def test_check_vgrids_modified_initial(self): + """Verify initial state of modified vgrids list is empty""" + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, ['ALL']) + reset_vgrids_modified(self.configuration) + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, []) + + def test_resources_using_re_notfound(self): + """Test RE with no assigned resources returns empty list""" + # Nonexistent RE should have no resources + res_list = resources_using_re(self.configuration, 'NoSuchRE') + self.assertEqual(res_list, []) + + def test_vgrid_inherit_map_single(self): + """Test inheritance mapping with single vgrid""" + test_settings = [('vgrid_name', self.test_vgrid), + ('description', 'Test description')] + test_map = { + VGRIDS: { + self.test_vgrid: { + SETTINGS: test_settings + } + } + } + inherited_map = vgrid_inherit_map(self.configuration, test_map) + settings_dict = dict(inherited_map[VGRIDS][self.test_vgrid][SETTINGS]) + self.assertEqual(settings_dict['description'], 'Test description') + def test_access_nonexistent_vgrid(self): """Ensure checks fail cleanly for non-existent vgrid""" allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, From ae30422e432add9658e39bb57dc4ccd42fcac10b Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Thu, 23 Oct 2025 23:51:24 +0200 Subject: [PATCH 08/20] Tweak dummy user provisioning a bit more to pick up more samples in tested functions and increase basic coverage. --- tests/test_mig_shared_vgridaccess.py | 40 ++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 0b3096f6a..626a413bd 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -39,7 +39,8 @@ load_resource_map, load_user_map, load_vgrid_map, mark_vgrid_modified, \ refresh_resource_map, refresh_user_map, refresh_vgrid_map, \ res_vgrid_access, reset_vgrids_modified, resources_using_re, unmap_vgrid, \ - user_vgrid_access, vgrid_inherit_map + user_allowed_res_confs, user_vgrid_access, user_visible_res_confs, \ + user_visible_user_confs, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -108,7 +109,10 @@ def _create_resource(self, res_name, owners, config=None): self.assertTrue(saved) if config is None: # Make sure conf has one valid field - config = {'HOSTURL': res_name} + config = {'HOSTURL': res_name, + 'EXECONFIG': [{'name': 'exe', 'vgrid': ['Generic']}], + 'STORECONFIG': [{'name': 'exe', 'vgrid': ['Generic']}] + } saved = pickle(config, res_config_path, self.logger) self.assertTrue(saved) @@ -554,6 +558,38 @@ def test_general_vgrid_access(self): self.TEST_OUTSIDER_DN) self.assertFalse(self.test_vgrid in allowed_vgrids) + def test_user_allowed_res_confs(self): + """Minimal test for user_allowed_res_confs""" + # Create test user and add test resource to vgrid + self._provision_test_user(self, self.TEST_OWNER_DN) + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + resources=[self.TEST_RESOURCE_ID]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + # Owner should be allowed access + allowed = user_allowed_res_confs(self.configuration, + self.TEST_OWNER_DN) + self.assertIn(self.TEST_RESOURCE_ALIAS, allowed) + + def test_user_visible_res_confs(self): + """Minimal test for user_visible_res_confs""" + # Owner should see owned resources even without vgrid access + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + force_update_resource_map(self.configuration) + visible = user_visible_res_confs( + self.configuration, self.TEST_OWNER_DN) + self.assertIn(self.TEST_RESOURCE_ALIAS, visible) + + def test_user_visible_user_confs(self): + """Minimal test for user_visible_user_confs""" + # Owner should see themselves in auto map + self._provision_test_user(self, self.TEST_OWNER_DN) + force_update_user_map(self.configuration) + visible = user_visible_user_confs( + self.configuration, self.TEST_OWNER_DN) + self.assertIn(self.TEST_OWNER_UUID, visible) + def test_get_re_provider_map(self): """Test RE provider map includes test resource""" test_re = 'Python' From 1030730597d6913504e5347c14b8e516dd4b0209 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Fri, 24 Oct 2025 00:51:38 +0200 Subject: [PATCH 09/20] More or less completed basic function use coverage. Clean up some redundant assertions. --- tests/test_mig_shared_vgridaccess.py | 159 ++++++++++++++++++++++++--- 1 file changed, 145 insertions(+), 14 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 626a413bd..94eaf0c7b 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -33,14 +33,19 @@ from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings from mig.shared.vgridaccess import CONF, MEMBERS, OWNERS, RESOURCES, SETTINGS, \ - USERID, VGRIDS, check_vgrid_access, check_vgrids_modified, \ - force_update_resource_map, force_update_user_map, force_update_vgrid_map, \ - get_re_provider_map, get_resource_map, get_user_map, get_vgrid_map, \ - load_resource_map, load_user_map, load_vgrid_map, mark_vgrid_modified, \ - refresh_resource_map, refresh_user_map, refresh_vgrid_map, \ - res_vgrid_access, reset_vgrids_modified, resources_using_re, unmap_vgrid, \ - user_allowed_res_confs, user_vgrid_access, user_visible_res_confs, \ - user_visible_user_confs, vgrid_inherit_map + USERID, VGRIDS, check_resources_modified, check_vgrid_access, \ + check_vgrids_modified, fill_placeholder_cache, force_update_resource_map, \ + force_update_user_map, force_update_vgrid_map, get_re_provider_map, \ + get_resource_map, get_user_map, get_vgrid_map, get_vgrid_map_vgrids, \ + is_vgrid_parent_placeholder, load_resource_map, load_user_map, \ + load_vgrid_map, mark_vgrid_modified, refresh_resource_map, \ + refresh_user_map, refresh_vgrid_map, res_vgrid_access, \ + reset_resources_modified, reset_vgrids_modified, resources_using_re, \ + unmap_inheritance, unmap_resource, unmap_vgrid, user_allowed_res_confs, \ + user_allowed_res_exes, user_allowed_res_stores, user_allowed_res_units, \ + user_allowed_user_confs, user_owned_res_exes, user_owned_res_stores, \ + user_vgrid_access, user_visible_res_confs, user_visible_res_exes, \ + user_visible_res_stores, user_visible_user_confs, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -164,21 +169,18 @@ def test_refresh_user_map(self): """Minimal test for user map refresh functionality""" self._provision_test_user(self, self.TEST_USER_DN) user_map = refresh_user_map(self.configuration) - self.assertTrue(user_map) self.assertIn(self.TEST_USER_DN, user_map) def test_refresh_resource_map(self): """Minimal test for resource map refresh functionality""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) res_map = refresh_resource_map(self.configuration) - self.assertTrue(res_map) self.assertIn(self.TEST_RESOURCE_ID, res_map) def test_refresh_vgrid_map(self): """Minimal test for vgrid map refresh functionality""" self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) vgrid_map = refresh_vgrid_map(self.configuration) - self.assertTrue(vgrid_map) self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) def test_get_user_map(self): @@ -186,7 +188,6 @@ def test_get_user_map(self): self._provision_test_user(self, self.TEST_USER_DN) force_update_user_map(self.configuration) user_map = get_user_map(self.configuration) - self.assertTrue(user_map) self.assertIn(self.TEST_USER_DN, user_map) def test_get_resource_map(self): @@ -194,7 +195,6 @@ def test_get_resource_map(self): self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) force_update_resource_map(self.configuration) resource_map = get_resource_map(self.configuration) - self.assertTrue(resource_map) self.assertIn(self.TEST_RESOURCE_ID, resource_map) def test_get_vgrid_map(self): @@ -202,7 +202,6 @@ def test_get_vgrid_map(self): self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) vgrid_map = get_vgrid_map(self.configuration) - self.assertTrue(vgrid_map) self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) def test_load_user_map(self): @@ -240,6 +239,108 @@ def test_load_vgrid_map(self): vgrid_map, map_stamp = load_vgrid_map(self.configuration) self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, {})) + def test_get_vgrid_map_vgrids(self): + """Test get_vgrid_map_vgrids returns vgrid list""" + vgrid_list = get_vgrid_map_vgrids(self.configuration) + self.assertTrue(isinstance(vgrid_list, list)) + self.assertEqual(['Generic'], vgrid_list) + + def test_user_owned_res_exes(self): + """Test user_owned_res_exes returns owned execution nodes""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + force_update_resource_map(self.configuration) + owned = user_owned_res_exes(self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(owned, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, owned) + + def test_user_owned_res_stores(self): + """Test user_owned_res_stores returns owned storage nodes""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + force_update_resource_map(self.configuration) + owned = user_owned_res_stores(self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(owned, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, owned) + + def test_user_allowed_res_units(self): + """Test user_allowed_res_units returns allowed units""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + allowed = user_allowed_res_units( + self.configuration, self.TEST_OWNER_DN, "exe") + self.assertTrue(isinstance(allowed, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, allowed) + + def test_user_allowed_res_exes(self): + """Test user_allowed_res_exes returns allowed exes""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + allowed = user_allowed_res_exes(self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(allowed, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, allowed) + + def test_user_allowed_res_stores(self): + """Test user_allowed_res_stores returns allowed stores""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + allowed = user_allowed_res_stores( + self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(allowed, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, allowed) + + def test_user_visible_res_exes(self): + """Test user_visible_res_exes returns visible exes""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + visible = user_visible_res_exes(self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(visible, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, visible) + + def test_user_visible_res_stores(self): + """Test user_visible_res_stores returns visible stores""" + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + force_update_vgrid_map(self.configuration) + force_update_resource_map(self.configuration) + visible = user_visible_res_stores( + self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(visible, dict)) + self.assertIn(self.TEST_RESOURCE_ALIAS, visible) + + def test_user_allowed_user_confs(self): + """Test user_allowed_user_confs returns allowed user confs""" + self._provision_test_user(self, self.TEST_OWNER_DN) + self._provision_test_user(self, self.TEST_USER_DN) + self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN], + [self.TEST_USER_DN]) + force_update_vgrid_map(self.configuration) + force_update_user_map(self.configuration) + allowed = user_allowed_user_confs( + self.configuration, self.TEST_OWNER_DN) + self.assertTrue(isinstance(allowed, dict)) + self.assertIn(self.TEST_USER_UUID, allowed) + self.assertIn(self.TEST_OWNER_UUID, allowed) + + def test_fill_placeholder_cache(self): + """Test fill_placeholder_cache populates cache""" + cache = {} + fill_placeholder_cache(self.configuration, cache, [self.test_vgrid]) + self.assertIn(self.test_vgrid, cache) + + def test_is_vgrid_parent_placeholder(self): + """Test is_vgrid_parent_placeholder detection""" + test_path = os.path.join(self.configuration.user_home, 'testvgrid') + result = is_vgrid_parent_placeholder(self.configuration, test_path, + test_path) + self.assertIsNone(result) + def test_check_vgrids_modified_initial(self): """Verify initial state of modified vgrids list is empty""" modified, stamp = check_vgrids_modified(self.configuration) @@ -417,6 +518,22 @@ def test_settings_inheritance(self): # Verify hidden setting inheritance self.assertEqual(sub_settings_dict.get('hidden'), True) + def test_unmap_inheritance(self): + """Test unmap_inheritance clears inherited mappings""" + self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN]) + sub_vgrid = os.path.join(self.test_vgrid, 'subvgrid') + self._create_vgrid(sub_vgrid) + + # Force refresh of cached map + updated_map = force_update_vgrid_map(self.configuration) + + # Unmap and verify mark modified + unmap_inheritance(self.configuration, self.test_vgrid, + self.TEST_OWNER_DN) + + modified, stamp = check_vgrids_modified(self.configuration) + self.assertEqual(modified, [self.test_vgrid, sub_vgrid]) + def test_user_map_fields(self): """Verify user map includes complete profile/settings data""" # First add a couple of test users @@ -630,6 +747,20 @@ def test_unmap_vgrid(self): mod_list, mod_stamp = check_vgrids_modified(self.configuration) self.assertIn(self.test_vgrid, mod_list) + def test_unmap_resource(self): + """Test unmap_resource marks resource modified""" + mod_list, mod_stamp = check_resources_modified(self.configuration) + self.assertNotIn(self.TEST_RESOURCE_ID, mod_list) + + # Unmap and verify mark modified + unmap_resource(self.configuration, self.TEST_RESOURCE_ID) + + mod_list, mod_stamp = check_vgrids_modified(self.configuration) + self.assertIn(self.TEST_RESOURCE_ID, mod_list) + # TODO: fix and enable next + # mod_list, mod_stamp = check_resources_modified(self.configuration) + # self.assertIn(self.TEST_RESOURCE_ID, mod_list) + def test_check_vgrids_modified_initial(self): """Verify initial state of modified vgrids list is empty""" modified, stamp = check_vgrids_modified(self.configuration) From 7bbd02b7795e8026eaf6323386adcb2e5c7dcb29 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Fri, 24 Oct 2025 01:00:46 +0200 Subject: [PATCH 10/20] Removed inadvertently duplicated test functions. --- tests/test_mig_shared_vgridaccess.py | 29 ---------------------------- 1 file changed, 29 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 94eaf0c7b..19c8aa1d2 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -761,35 +761,6 @@ def test_unmap_resource(self): # mod_list, mod_stamp = check_resources_modified(self.configuration) # self.assertIn(self.TEST_RESOURCE_ID, mod_list) - def test_check_vgrids_modified_initial(self): - """Verify initial state of modified vgrids list is empty""" - modified, stamp = check_vgrids_modified(self.configuration) - self.assertEqual(modified, ['ALL']) - reset_vgrids_modified(self.configuration) - modified, stamp = check_vgrids_modified(self.configuration) - self.assertEqual(modified, []) - - def test_resources_using_re_notfound(self): - """Test RE with no assigned resources returns empty list""" - # Nonexistent RE should have no resources - res_list = resources_using_re(self.configuration, 'NoSuchRE') - self.assertEqual(res_list, []) - - def test_vgrid_inherit_map_single(self): - """Test inheritance mapping with single vgrid""" - test_settings = [('vgrid_name', self.test_vgrid), - ('description', 'Test description')] - test_map = { - VGRIDS: { - self.test_vgrid: { - SETTINGS: test_settings - } - } - } - inherited_map = vgrid_inherit_map(self.configuration, test_map) - settings_dict = dict(inherited_map[VGRIDS][self.test_vgrid][SETTINGS]) - self.assertEqual(settings_dict['description'], 'Test description') - def test_access_nonexistent_vgrid(self): """Ensure checks fail cleanly for non-existent vgrid""" allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, From d060e809fb20577541381a547286671e30c6cb05 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Fri, 24 Oct 2025 09:27:42 +0200 Subject: [PATCH 11/20] Consistency fixes across the board. --- tests/test_mig_shared_vgridaccess.py | 238 +++++++++++++++------------ 1 file changed, 136 insertions(+), 102 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 19c8aa1d2..a0f1bf934 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -65,11 +65,16 @@ class TestMigSharedVgridAccess(MigTestCase): '/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Outsider/'\ 'emailAddress=outsider@example.com' TEST_RESOURCE_ID = 'test.example.org.0' + TEST_VGRID_NAME = 'testvgrid' TEST_OWNER_UUID = 'ff326a2b984828d9b32077c9b0b35a05' TEST_USER_UUID = '707a2213995b4fb385793b5a7cb82a18' TEST_RESOURCE_ALIAS = '0835f310d6422c36e33eeb7d0d3e9cf5' + # Default vgrid is initially set up without settings when force loaded + MINIMAL_VGRIDS = {'Generic': {OWNERS: [], MEMBERS: [], RESOURCES: [], + SETTINGS: []}} + def _provide_configuration(self): """Prepare isolated test config""" return 'testconfig' @@ -123,47 +128,62 @@ def _create_resource(self, res_name, owners, config=None): def before_each(self): """Create test environment for vgridaccess tests""" - ensure_dirs_exist(self.configuration.mig_system_files) - ensure_dirs_exist(self.configuration.mig_system_run) - ensure_dirs_exist(self.configuration.user_home) - ensure_dirs_exist(self.configuration.user_settings) - ensure_dirs_exist(self.configuration.vgrid_home) - ensure_dirs_exist(self.configuration.resource_home) - # Start with empty maps - force_update_vgrid_map(self.configuration, clean=True) - force_update_user_map(self.configuration, clean=True) - force_update_resource_map(self.configuration, clean=True) - - self.test_vgrid = 'testvgrid' + used_state_dirs = [ + self.configuration.mig_system_files, + self.configuration.mig_system_run, + self.configuration.user_home, + self.configuration.user_settings, + self.configuration.vgrid_home, + self.configuration.resource_home, + ] + for state_dir in used_state_dirs: + ensure_dirs_exist(state_dir) + + # IMPORTANT: don't use refresh_X here as it does not reset last_X cache + # Start with minimal maps and corresponding cache - only default vgrid + vgrid_map = force_update_vgrid_map(self.configuration) + self.assertEqual(vgrid_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) + res_map = force_update_resource_map(self.configuration) + self.assertEqual(res_map, {}) + user_map = force_update_user_map(self.configuration) + self.assertEqual(user_map, {}) def test_force_update_user_map(self): """Simple test that user map refresh completes""" - user_map_before = get_user_map(self.configuration) - self.assertFalse(user_map_before) + # Verify empty map on init + user_map_before, _ = load_user_map(self.configuration) + self.assertEqual(user_map_before, {}) + self._provision_test_user(self, self.TEST_USER_DN) updated_users = force_update_user_map(self.configuration) self.assertTrue(updated_users) self.assertNotEqual(user_map_before, updated_users) + self.assertIn(self.TEST_USER_DN, updated_users) def test_force_update_resource_map(self): """Simple test that resource map refresh completes""" - res_map_before = get_resource_map(self.configuration) - self.assertFalse(res_map_before) + # Verify empty map on init + res_map_before, _ = load_resource_map(self.configuration) + self.assertEqual(res_map_before, {}) + self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) updated_res = force_update_resource_map(self.configuration) self.assertTrue(updated_res) self.assertNotEqual(len(res_map_before), len(updated_res)) + self.assertIn(self.TEST_RESOURCE_ID, updated_res) def test_force_update_vgrid_map(self): """Simple test that vgrid map refresh completes""" # Only (implicit) default vgrid in vgrid map before init - vgrid_map_before = get_vgrid_map(self.configuration) - self.assertEqual(len(vgrid_map_before.get(VGRIDS, [])), 1) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + vgrid_map_before, _ = load_vgrid_map(self.configuration) + self.assertEqual(vgrid_map_before.get(VGRIDS, {}), self.MINIMAL_VGRIDS) + + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) updated_vgrid = force_update_vgrid_map(self.configuration) self.assertTrue(updated_vgrid) - self.assertNotEqual(len(vgrid_map_before.get(VGRIDS, [])), - len(updated_vgrid.get(VGRIDS, []))) + self.assertNotEqual(len(vgrid_map_before.get(VGRIDS, {})), + len(updated_vgrid.get(VGRIDS, {}))) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid.get(VGRIDS, {})) def test_refresh_user_map(self): """Minimal test for user map refresh functionality""" @@ -179,9 +199,9 @@ def test_refresh_resource_map(self): def test_refresh_vgrid_map(self): """Minimal test for vgrid map refresh functionality""" - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) vgrid_map = refresh_vgrid_map(self.configuration) - self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) + self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) def test_get_user_map(self): """Minimal test for user map refresh functionality""" @@ -199,48 +219,65 @@ def test_get_resource_map(self): def test_get_vgrid_map(self): """Minimal test for user map refresh functionality""" - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) vgrid_map = get_vgrid_map(self.configuration) - self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, [])) + self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) def test_load_user_map(self): """Basic test for direct user map loading""" # Get empty map initially user_map, map_stamp = load_user_map(self.configuration) self.assertEqual(user_map, {}) + self.assertFalse(user_map, map_stamp) + # Add test user + now = time.time() self._provision_test_user(self, self.TEST_USER_DN) force_update_user_map(self.configuration) # Verify updated map contains user updated_map, updated_stamp = load_user_map(self.configuration) self.assertIn(self.TEST_USER_DN, updated_map) + self.assertTrue(updated_stamp > now) def test_load_resource_map(self): """Basic test for direct resource map loading""" # Get empty map initially res_map, map_stamp = load_resource_map(self.configuration) self.assertEqual(res_map, {}) + self.assertFalse(res_map, map_stamp) + # Add test resource + now = time.time() self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) force_update_resource_map(self.configuration) # Verify updated map contains resource after refresh res_map, map_stamp = load_resource_map(self.configuration) self.assertIn(self.TEST_RESOURCE_ID, res_map) + self.assertTrue(map_stamp > now) def test_load_vgrid_map(self): """Basic test for direct vgrid map loading""" - # Get map with at least 'Generic' vgrid + # Check map only contains 'Generic' vgrid initially vgrid_map, map_stamp = load_vgrid_map(self.configuration) - self.assertIn('Generic', vgrid_map.get(VGRIDS, {})) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self.assertEqual(vgrid_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) + self.assertTrue(map_stamp) + + # Add test vgrid + now = time.time() + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) # Verify updated map contains vgrid after refresh vgrid_map, map_stamp = load_vgrid_map(self.configuration) - self.assertIn(self.test_vgrid, vgrid_map.get(VGRIDS, {})) + self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) + self.assertTrue(map_stamp > now) def test_get_vgrid_map_vgrids(self): """Test get_vgrid_map_vgrids returns vgrid list""" + # Check map only contains 'Generic' vgrid initially + vgrid_map, _ = load_vgrid_map(self.configuration) + self.assertEqual(vgrid_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) + vgrid_list = get_vgrid_map_vgrids(self.configuration) self.assertTrue(isinstance(vgrid_list, list)) self.assertEqual(['Generic'], vgrid_list) @@ -264,7 +301,7 @@ def test_user_owned_res_stores(self): def test_user_allowed_res_units(self): """Test user_allowed_res_units returns allowed units""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) allowed = user_allowed_res_units( @@ -275,7 +312,7 @@ def test_user_allowed_res_units(self): def test_user_allowed_res_exes(self): """Test user_allowed_res_exes returns allowed exes""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) allowed = user_allowed_res_exes(self.configuration, self.TEST_OWNER_DN) @@ -285,7 +322,7 @@ def test_user_allowed_res_exes(self): def test_user_allowed_res_stores(self): """Test user_allowed_res_stores returns allowed stores""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) allowed = user_allowed_res_stores( @@ -296,7 +333,7 @@ def test_user_allowed_res_stores(self): def test_user_visible_res_exes(self): """Test user_visible_res_exes returns visible exes""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) visible = user_visible_res_exes(self.configuration, self.TEST_OWNER_DN) @@ -306,7 +343,7 @@ def test_user_visible_res_exes(self): def test_user_visible_res_stores(self): """Test user_visible_res_stores returns visible stores""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) visible = user_visible_res_stores( @@ -318,7 +355,7 @@ def test_user_allowed_user_confs(self): """Test user_allowed_user_confs returns allowed user confs""" self._provision_test_user(self, self.TEST_OWNER_DN) self._provision_test_user(self, self.TEST_USER_DN) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN], [self.TEST_USER_DN]) force_update_vgrid_map(self.configuration) force_update_user_map(self.configuration) @@ -331,8 +368,9 @@ def test_user_allowed_user_confs(self): def test_fill_placeholder_cache(self): """Test fill_placeholder_cache populates cache""" cache = {} - fill_placeholder_cache(self.configuration, cache, [self.test_vgrid]) - self.assertIn(self.test_vgrid, cache) + fill_placeholder_cache(self.configuration, cache, [ + self.TEST_VGRID_NAME]) + self.assertIn(self.TEST_VGRID_NAME, cache) def test_is_vgrid_parent_placeholder(self): """Test is_vgrid_parent_placeholder detection""" @@ -341,14 +379,6 @@ def test_is_vgrid_parent_placeholder(self): test_path) self.assertIsNone(result) - def test_check_vgrids_modified_initial(self): - """Verify initial state of modified vgrids list is empty""" - modified, stamp = check_vgrids_modified(self.configuration) - self.assertEqual(modified, ['ALL']) - reset_vgrids_modified(self.configuration) - modified, stamp = check_vgrids_modified(self.configuration) - self.assertEqual(modified, []) - def test_resources_using_re_notfound(self): """Test RE with no assigned resources returns empty list""" # Nonexistent RE should have no resources @@ -357,11 +387,11 @@ def test_resources_using_re_notfound(self): def test_vgrid_inherit_map_single(self): """Test inheritance mapping with single vgrid""" - test_settings = [('vgrid_name', self.test_vgrid), + test_settings = [('vgrid_name', self.TEST_VGRID_NAME), ('hidden', True)] test_map = { VGRIDS: { - self.test_vgrid: { + self.TEST_VGRID_NAME: { SETTINGS: test_settings, OWNERS: [self.TEST_OWNER_DN] } @@ -369,28 +399,32 @@ def test_vgrid_inherit_map_single(self): } inherited_map = vgrid_inherit_map(self.configuration, test_map) vgrid_data = inherited_map.get(VGRIDS, {}) - self.assertTrue(self.test_vgrid in vgrid_data) - settings_dict = dict(vgrid_data[self.test_vgrid][SETTINGS]) + self.assertTrue(self.TEST_VGRID_NAME in vgrid_data) + settings_dict = dict(vgrid_data[self.TEST_VGRID_NAME][SETTINGS]) self.assertIs(type(settings_dict), dict) self.assertEqual(settings_dict.get('hidden'), True) - def test_check_vgrids_modified(self): - """Minimal test for vgrid modified tracking""" - # Initially ALL marked modified until cache init + # TODO: move these two modified tests to a test_mig_shared_modified.py + def test_check_vgrids_modified_initial(self): + """Verify initial modified vgrids list marks ALL and empty on reset""" modified, stamp = check_vgrids_modified(self.configuration) self.assertEqual(modified, ['ALL']) - # Reset and check ALL gone reset_vgrids_modified(self.configuration) modified, stamp = check_vgrids_modified(self.configuration) self.assertEqual(modified, []) + + def test_check_vgrids_modified(self): + """Minimal test for vgrid modified tracking""" + # Reset markers + reset_vgrids_modified(self.configuration) # Mark modified - mark_vgrid_modified(self.configuration, self.test_vgrid) + mark_vgrid_modified(self.configuration, self.TEST_VGRID_NAME) modified, stamp = check_vgrids_modified(self.configuration) - self.assertIn(self.test_vgrid, modified) + self.assertIn(self.TEST_VGRID_NAME, modified) # Reset and check gone again reset_vgrids_modified(self.configuration) modified, stamp = check_vgrids_modified(self.configuration) - self.assertNotIn(self.test_vgrid, modified) + self.assertNotIn(self.TEST_VGRID_NAME, modified) def test_user_vgrid_access(self): """Minimal test for user vgrid participation""" @@ -401,12 +435,12 @@ def test_user_vgrid_access(self): self.assertTrue(len(allowed_vgrids), 1) # Create private vgrid self._provision_test_user(self, self.TEST_OWNER_DN) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) # Refresh maps to reflect new content force_update_vgrid_map(self.configuration) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OWNER_DN) - self.assertIn(self.test_vgrid, allowed_vgrids) + self.assertIn(self.TEST_VGRID_NAME, allowed_vgrids) def test_res_vgrid_access(self): """Minimal test for resource vgrid participation""" @@ -416,54 +450,54 @@ def test_res_vgrid_access(self): self.assertEqual(allowed_vgrids, ['Generic']) # Add to vgrid self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, resources=[self.TEST_RESOURCE_ID]) + self._create_vgrid(self.TEST_VGRID_NAME, resources=[ + self.TEST_RESOURCE_ID]) # Refresh maps to reflect new content force_update_vgrid_map(self.configuration) allowed = res_vgrid_access(self.configuration, self.TEST_RESOURCE_ID) - self.assertIn(self.test_vgrid, allowed) + self.assertIn(self.TEST_VGRID_NAME, allowed) def test_vgrid_map_refresh(self): """Verify vgrid map refresh captures changes""" # We always init empty maps - initial_map = get_vgrid_map(self.configuration, recursive=False) - self.assertFalse(self.test_vgrid in initial_map.get(VGRIDS, {})) + initial_map, _ = load_vgrid_map(self.configuration) + self.assertEqual(initial_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) - self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) # Force refresh map updated_map = force_update_vgrid_map(self.configuration) vgrids = updated_map.get(VGRIDS, {}) - self.assertTrue(self.test_vgrid in vgrids) - self.assertEqual(vgrids[self.test_vgrid] + self.assertTrue(self.TEST_VGRID_NAME in vgrids) + self.assertEqual(vgrids[self.TEST_VGRID_NAME] [OWNERS], [self.TEST_OWNER_DN]) def test_user_map_access(self): """Test user permissions through cached access maps""" # Add user as member - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) force_update_vgrid_map(self.configuration) # Verify member access allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, - self.test_vgrid) + self.TEST_VGRID_NAME) self.assertTrue(allowed) def test_resource_map_update(self): """Verify resource visibility in cache""" # Check cached resource map does not yet contain entry - cached_map, map_stamp = load_resource_map(self.configuration, - caching=True) - self.assertFalse(cached_map, map_stamp) - self.assertFalse(self.TEST_RESOURCE_ID in cached_map) + res_map_before, _ = load_resource_map(self.configuration, + caching=True) + self.assertEqual(res_map_before, {}) # Add vgrid with assigned resource self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) updated_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) - top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) - top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) # Check resource map contains resource entry @@ -474,11 +508,11 @@ def test_resource_map_update(self): def test_settings_inheritance(self): """Test inherited settings propagation through cached maps""" # Create top and sub vgrids with 'hidden' setting on top vgrid - top_settings = [('vgrid_name', self.test_vgrid), + top_settings = [('vgrid_name', self.TEST_VGRID_NAME), ('hidden', True)] - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], settings=top_settings) - sub_vgrid = os.path.join(self.test_vgrid, 'subvgrid') + sub_vgrid = os.path.join(self.TEST_VGRID_NAME, 'subvgrid') self._create_vgrid(sub_vgrid) # Force refresh of cached map @@ -489,7 +523,7 @@ def test_settings_inheritance(self): self.assertTrue(vgrid_data) # Retrieve top vgrid settings from cached map - top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) + top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) self.assertTrue(top_vgrid_data) # Convert settings list of tuples to dict top_settings_dict = dict(top_vgrid_data.get(SETTINGS, [])) @@ -520,19 +554,19 @@ def test_settings_inheritance(self): def test_unmap_inheritance(self): """Test unmap_inheritance clears inherited mappings""" - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN]) - sub_vgrid = os.path.join(self.test_vgrid, 'subvgrid') + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN]) + sub_vgrid = os.path.join(self.TEST_VGRID_NAME, 'subvgrid') self._create_vgrid(sub_vgrid) # Force refresh of cached map updated_map = force_update_vgrid_map(self.configuration) # Unmap and verify mark modified - unmap_inheritance(self.configuration, self.test_vgrid, + unmap_inheritance(self.configuration, self.TEST_VGRID_NAME, self.TEST_OWNER_DN) modified, stamp = check_vgrids_modified(self.configuration) - self.assertEqual(modified, [self.test_vgrid, sub_vgrid]) + self.assertEqual(modified, [self.TEST_VGRID_NAME, sub_vgrid]) def test_user_map_fields(self): """Verify user map includes complete profile/settings data""" @@ -553,14 +587,14 @@ def test_resource_revoked_access(self): """Verify resource removal propagates through cached maps""" # First add resource and vgrid self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) initial_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map contains resource entry vgrid_data = initial_vgrid_map.get(VGRIDS, {}) - top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) - top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) # Check resource map contains resource entry @@ -568,15 +602,15 @@ def test_resource_revoked_access(self): self.assertIn(self.TEST_RESOURCE_ID, initial_map) # Remove resource assignment from vgrid - status, msg = vgrid_set_entities(self.configuration, self.test_vgrid, + status, msg = vgrid_set_entities(self.configuration, self.TEST_VGRID_NAME, 'resources', [], allow_empty=True) self.assertTrue(status, msg) updated_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map no longer contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) - top_vgrid_data = vgrid_data.get(self.test_vgrid, {}) - top_vgrid_res = top_vgrid_data.get(RESOURCES, []) + top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) + top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) self.assertFalse(self.TEST_RESOURCE_ID in top_vgrid_res) # Verify resource still in resource map @@ -623,7 +657,7 @@ def test_hidden_setting_propagation(self): def test_default_vgrid_access(self): """Verify special access rules for default vgrid""" - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) @@ -647,40 +681,40 @@ def test_default_vgrid_access(self): def test_general_vgrid_access(self): """Verify general access rules for vgrids""" - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) # Test vgrid must allow owner and members access allowed = check_vgrid_access(self.configuration, self.TEST_OWNER_DN, - self.test_vgrid) + self.TEST_VGRID_NAME) self.assertTrue(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OWNER_DN) - self.assertTrue(self.test_vgrid in allowed_vgrids) + self.assertTrue(self.TEST_VGRID_NAME in allowed_vgrids) allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, - self.test_vgrid) + self.TEST_VGRID_NAME) self.assertTrue(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_MEMBER_DN) - self.assertTrue(self.test_vgrid in allowed_vgrids) + self.assertTrue(self.TEST_VGRID_NAME in allowed_vgrids) # Test vgrid must reject allow outsider access allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, - self.test_vgrid) + self.TEST_VGRID_NAME) self.assertFalse(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN) - self.assertFalse(self.test_vgrid in allowed_vgrids) + self.assertFalse(self.TEST_VGRID_NAME in allowed_vgrids) def test_user_allowed_res_confs(self): """Minimal test for user_allowed_res_confs""" # Create test user and add test resource to vgrid self._provision_test_user(self, self.TEST_OWNER_DN) self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) - self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN], + self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) force_update_vgrid_map(self.configuration) force_update_resource_map(self.configuration) @@ -739,13 +773,13 @@ def test_resources_using_re(self): def test_unmap_vgrid(self): """Verify unmapping marks vgrid modified for update in cached data""" mod_list, mod_stamp = check_vgrids_modified(self.configuration) - self.assertNotIn(self.test_vgrid, mod_list) + self.assertNotIn(self.TEST_VGRID_NAME, mod_list) # Unmap and verify mark modified - unmap_vgrid(self.configuration, self.test_vgrid) + unmap_vgrid(self.configuration, self.TEST_VGRID_NAME) mod_list, mod_stamp = check_vgrids_modified(self.configuration) - self.assertIn(self.test_vgrid, mod_list) + self.assertIn(self.TEST_VGRID_NAME, mod_list) def test_unmap_resource(self): """Test unmap_resource marks resource modified""" @@ -774,12 +808,12 @@ def test_access_nonexistent_vgrid(self): def test_empty_member_access(self): """Verify members-only vgrid rejects outsiders""" - self._create_vgrid(self.test_vgrid, [], [self.TEST_MEMBER_DN]) + self._create_vgrid(self.TEST_VGRID_NAME, [], [self.TEST_MEMBER_DN]) force_update_vgrid_map(self.configuration) # Outsider should be blocked despite no owners allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, - self.test_vgrid) + self.TEST_VGRID_NAME) self.assertFalse(allowed) From 188e4b8859afdefd2690516a6f525339bb24f866 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 11:29:45 +0100 Subject: [PATCH 12/20] Introduce thorough cleaning of cache map and stamps in between tests and verify that each test really has a clean environment to rule out CI test failures due to stale data. --- tests/test_mig_shared_vgridaccess.py | 57 ++++++++++++++++++++++------ 1 file changed, 46 insertions(+), 11 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index a0f1bf934..acde40178 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -33,19 +33,20 @@ from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings from mig.shared.vgridaccess import CONF, MEMBERS, OWNERS, RESOURCES, SETTINGS, \ - USERID, VGRIDS, check_resources_modified, check_vgrid_access, \ + USERID, USERS, VGRIDS, check_resources_modified, check_vgrid_access, \ check_vgrids_modified, fill_placeholder_cache, force_update_resource_map, \ force_update_user_map, force_update_vgrid_map, get_re_provider_map, \ get_resource_map, get_user_map, get_vgrid_map, get_vgrid_map_vgrids, \ - is_vgrid_parent_placeholder, load_resource_map, load_user_map, \ - load_vgrid_map, mark_vgrid_modified, refresh_resource_map, \ - refresh_user_map, refresh_vgrid_map, res_vgrid_access, \ - reset_resources_modified, reset_vgrids_modified, resources_using_re, \ - unmap_inheritance, unmap_resource, unmap_vgrid, user_allowed_res_confs, \ - user_allowed_res_exes, user_allowed_res_stores, user_allowed_res_units, \ - user_allowed_user_confs, user_owned_res_exes, user_owned_res_stores, \ - user_vgrid_access, user_visible_res_confs, user_visible_res_exes, \ - user_visible_res_stores, user_visible_user_confs, vgrid_inherit_map + is_vgrid_parent_placeholder, last_load, last_map, last_refresh, \ + load_resource_map, load_user_map, load_vgrid_map, mark_vgrid_modified, \ + refresh_resource_map, refresh_user_map, refresh_vgrid_map, \ + res_vgrid_access, reset_resources_modified, reset_vgrids_modified, \ + resources_using_re, unmap_inheritance, unmap_resource, unmap_vgrid, \ + user_allowed_res_confs, user_allowed_res_exes, user_allowed_res_stores, \ + user_allowed_res_units, user_allowed_user_confs, user_owned_res_exes, \ + user_owned_res_stores, user_vgrid_access, user_visible_res_confs, \ + user_visible_res_exes, user_visible_res_stores, user_visible_user_confs, \ + vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -126,8 +127,15 @@ def _create_resource(self, res_name, owners, config=None): saved = pickle(config, res_config_path, self.logger) self.assertTrue(saved) + def _reset_caches(self): + """Assure all vgrid maps and stamps are wiped for next test""" + for field in (USERS, RESOURCES, VGRIDS): + last_refresh[field] = 0 + last_load[field] = 0 + last_map[field].clear() + def before_each(self): - """Create test environment for vgridaccess tests""" + """Create clean test environment for vgridaccess tests""" used_state_dirs = [ self.configuration.mig_system_files, self.configuration.mig_system_run, @@ -139,15 +147,42 @@ def before_each(self): for state_dir in used_state_dirs: ensure_dirs_exist(state_dir) + # Drop all caches and state between tests + self._reset_caches() + + # Make sure we start with a clean slate for each test + for field in (USERS, RESOURCES, VGRIDS): + self.assertTrue(last_refresh[field] == 0) + self.assertTrue(last_load[field] == 0) + self.assertEqual(last_map[field], {}) + # IMPORTANT: don't use refresh_X here as it does not reset last_X cache # Start with minimal maps and corresponding cache - only default vgrid vgrid_map = force_update_vgrid_map(self.configuration) self.assertEqual(vgrid_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) + + # Make sure all vgrid map entries are complete + for vgrid_name in vgrid_map.get(VGRIDS, {}): + for field in (OWNERS, MEMBERS, SETTINGS, RESOURCES): + vgrid_entry = vgrid_map[VGRIDS][vgrid_name] + self.assertIn(field, vgrid_entry) + res_map = force_update_resource_map(self.configuration) + # Make sure no resource map entries exist initially self.assertEqual(res_map, {}) + user_map = force_update_user_map(self.configuration) + # Make sure no user map entries exist initially self.assertEqual(user_map, {}) + # Make sure accounting and caching entries are consistent and complete + for field in (USERS, RESOURCES, VGRIDS): + self.assertTrue(last_refresh[field] > 0) + self.assertTrue(last_load[field] > 0) + self.assertEqual(vgrid_map, last_map[VGRIDS]) + self.assertEqual(res_map, last_map[RESOURCES]) + self.assertEqual(user_map, last_map[USERS]) + def test_force_update_user_map(self): """Simple test that user map refresh completes""" # Verify empty map on init From 9443b77229cf444eb98de1c4405f3168109a3dc5 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 11:41:10 +0100 Subject: [PATCH 13/20] Go one step further to assure vgridaccess cache maps and stamps really get reset in between tests. --- tests/test_mig_shared_vgridaccess.py | 43 ++++++++++++++-------------- 1 file changed, 22 insertions(+), 21 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index acde40178..dfdb892c2 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -30,6 +30,8 @@ import time import unittest +# IMPORTANT: reset of cross-module globals requires module-level import +import mig.shared.vgridaccess as vgridaccess from mig.shared.fileio import pickle, read_file from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings from mig.shared.vgridaccess import CONF, MEMBERS, OWNERS, RESOURCES, SETTINGS, \ @@ -37,16 +39,15 @@ check_vgrids_modified, fill_placeholder_cache, force_update_resource_map, \ force_update_user_map, force_update_vgrid_map, get_re_provider_map, \ get_resource_map, get_user_map, get_vgrid_map, get_vgrid_map_vgrids, \ - is_vgrid_parent_placeholder, last_load, last_map, last_refresh, \ - load_resource_map, load_user_map, load_vgrid_map, mark_vgrid_modified, \ - refresh_resource_map, refresh_user_map, refresh_vgrid_map, \ - res_vgrid_access, reset_resources_modified, reset_vgrids_modified, \ - resources_using_re, unmap_inheritance, unmap_resource, unmap_vgrid, \ - user_allowed_res_confs, user_allowed_res_exes, user_allowed_res_stores, \ - user_allowed_res_units, user_allowed_user_confs, user_owned_res_exes, \ - user_owned_res_stores, user_vgrid_access, user_visible_res_confs, \ - user_visible_res_exes, user_visible_res_stores, user_visible_user_confs, \ - vgrid_inherit_map + is_vgrid_parent_placeholder, load_resource_map, load_user_map, \ + load_vgrid_map, mark_vgrid_modified, refresh_resource_map, \ + refresh_user_map, refresh_vgrid_map, res_vgrid_access, \ + reset_resources_modified, reset_vgrids_modified, resources_using_re, \ + unmap_inheritance, unmap_resource, unmap_vgrid, user_allowed_res_confs, \ + user_allowed_res_exes, user_allowed_res_stores, user_allowed_res_units, \ + user_allowed_user_confs, user_owned_res_exes, user_owned_res_stores, \ + user_vgrid_access, user_visible_res_confs, user_visible_res_exes, \ + user_visible_res_stores, user_visible_user_confs, vgrid_inherit_map from tests.support import MigTestCase, ensure_dirs_exist, testmain @@ -130,9 +131,9 @@ def _create_resource(self, res_name, owners, config=None): def _reset_caches(self): """Assure all vgrid maps and stamps are wiped for next test""" for field in (USERS, RESOURCES, VGRIDS): - last_refresh[field] = 0 - last_load[field] = 0 - last_map[field].clear() + vgridaccess.last_refresh[field] = 0 + vgridaccess.last_load[field] = 0 + vgridaccess.last_map[field].clear() def before_each(self): """Create clean test environment for vgridaccess tests""" @@ -152,9 +153,9 @@ def before_each(self): # Make sure we start with a clean slate for each test for field in (USERS, RESOURCES, VGRIDS): - self.assertTrue(last_refresh[field] == 0) - self.assertTrue(last_load[field] == 0) - self.assertEqual(last_map[field], {}) + self.assertTrue(vgridaccess.last_refresh[field] == 0) + self.assertTrue(vgridaccess.last_load[field] == 0) + self.assertEqual(vgridaccess.last_map[field], {}) # IMPORTANT: don't use refresh_X here as it does not reset last_X cache # Start with minimal maps and corresponding cache - only default vgrid @@ -177,11 +178,11 @@ def before_each(self): # Make sure accounting and caching entries are consistent and complete for field in (USERS, RESOURCES, VGRIDS): - self.assertTrue(last_refresh[field] > 0) - self.assertTrue(last_load[field] > 0) - self.assertEqual(vgrid_map, last_map[VGRIDS]) - self.assertEqual(res_map, last_map[RESOURCES]) - self.assertEqual(user_map, last_map[USERS]) + self.assertTrue(vgridaccess.last_refresh[field] > 0) + self.assertTrue(vgridaccess.last_load[field] > 0) + self.assertEqual(vgrid_map, vgridaccess.last_map[VGRIDS]) + self.assertEqual(res_map, vgridaccess.last_map[RESOURCES]) + self.assertEqual(user_map, vgridaccess.last_map[USERS]) def test_force_update_user_map(self): """Simple test that user map refresh completes""" From a664e9d243adad6b644e3eb406250edd030439df Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 12:28:11 +0100 Subject: [PATCH 14/20] Another level ofassurance that tests don't have stale data and that all vgrids are fully populated with the vgrid_home/X pickles like after createvgrid call. --- tests/test_mig_shared_vgridaccess.py | 51 +++++++++++++++++----------- 1 file changed, 31 insertions(+), 20 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index dfdb892c2..91a39bd69 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -82,31 +82,40 @@ def _provide_configuration(self): return 'testconfig' def _create_vgrid(self, vgrid_name, owners=None, members=None, - resources=None, settings=None): + resources=None, settings=None, triggers=None): """Helper to create valid skeleton vgrid for testing""" vgrid_path = os.path.join(self.configuration.vgrid_home, vgrid_name) ensure_dirs_exist(vgrid_path) + # Save vgrid owners, members, resources, settings and triggers if owners is None: owners = [] - # Add vgrid owners - status, msg = vgrid_set_entities(self.configuration, vgrid_name, - 'owners', owners, allow_empty=True) - self.assertTrue(status, msg) - if members is not None: - status, msg = vgrid_set_entities(self.configuration, vgrid_name, + success_and_msg = vgrid_set_entities(self.configuration, vgrid_name, + 'owners', owners, allow_empty=True) + self.assertEqual(success_and_msg, (True, "")) + if members is None: + members = [] + success_and_msg = vgrid_set_entities(self.configuration, vgrid_name, 'members', members, - allow_empty=False) - self.assertTrue(status, msg) - if resources is not None: - status, msg = vgrid_set_entities(self.configuration, vgrid_name, + allow_empty=True) + self.assertEqual(success_and_msg, (True, "")) + if resources is None: + resources = [] + success_and_msg = vgrid_set_entities(self.configuration, vgrid_name, 'resources', resources, - allow_empty=False) - self.assertTrue(status, msg) - if settings is not None: - status, msg = vgrid_set_entities(self.configuration, vgrid_name, + allow_empty=True) + self.assertEqual(success_and_msg, (True, "")) + if settings is None: + settings = [('vgrid_name', vgrid_name)] + success_and_msg = vgrid_set_entities(self.configuration, vgrid_name, 'settings', settings, - allow_empty=False) - self.assertTrue(status, msg) + allow_empty=True) + self.assertEqual(success_and_msg, (True, "")) + if triggers is None: + triggers = [] + success_and_msg = vgrid_set_entities(self.configuration, vgrid_name, + 'triggers', triggers, + allow_empty=True) + self.assertEqual(success_and_msg, (True, "")) def _create_resource(self, res_name, owners, config=None): """Helper to create valid skeleton resource for testing""" @@ -147,6 +156,8 @@ def before_each(self): ] for state_dir in used_state_dirs: ensure_dirs_exist(state_dir) + # Make sure no stale data is left + self.assertEqual(os.listdir(state_dir), []) # Drop all caches and state between tests self._reset_caches() @@ -638,9 +649,9 @@ def test_resource_revoked_access(self): self.assertIn(self.TEST_RESOURCE_ID, initial_map) # Remove resource assignment from vgrid - status, msg = vgrid_set_entities(self.configuration, self.TEST_VGRID_NAME, - 'resources', [], allow_empty=True) - self.assertTrue(status, msg) + success_and_msg = vgrid_set_entities(self.configuration, self.TEST_VGRID_NAME, + 'resources', [], allow_empty=True) + self.assertEqual(success_and_msg, (True, '')) updated_vgrid_map = force_update_vgrid_map(self.configuration) # Check vgrid map no longer contains resource entry From 85adb30f5ee7cc5c7dba403c600bf1f5e363b4b1 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 12:51:59 +0100 Subject: [PATCH 15/20] Use assertIn helper to test for key or entry in list or dicts. More strict checking that created vgrids are really always in vgrid map after initial force_update_vgrid_map calls to rule out issues there in relation to remaining inconsistent CI errors. --- tests/test_mig_shared_vgridaccess.py | 45 +++++++++++++++++----------- 1 file changed, 27 insertions(+), 18 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 91a39bd69..213164b6f 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -446,7 +446,7 @@ def test_vgrid_inherit_map_single(self): } inherited_map = vgrid_inherit_map(self.configuration, test_map) vgrid_data = inherited_map.get(VGRIDS, {}) - self.assertTrue(self.TEST_VGRID_NAME in vgrid_data) + self.assertIn(self.TEST_VGRID_NAME, vgrid_data) settings_dict = dict(vgrid_data[self.TEST_VGRID_NAME][SETTINGS]) self.assertIs(type(settings_dict), dict) self.assertEqual(settings_dict.get('hidden'), True) @@ -478,7 +478,7 @@ def test_user_vgrid_access(self): # Start with global access to default vgrid allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_USER_DN) - self.assertTrue('Generic' in allowed_vgrids) + self.assertIn('Generic', allowed_vgrids) self.assertTrue(len(allowed_vgrids), 1) # Create private vgrid self._provision_test_user(self, self.TEST_OWNER_DN) @@ -514,7 +514,7 @@ def test_vgrid_map_refresh(self): # Force refresh map updated_map = force_update_vgrid_map(self.configuration) vgrids = updated_map.get(VGRIDS, {}) - self.assertTrue(self.TEST_VGRID_NAME in vgrids) + self.assertIn(self.TEST_VGRID_NAME, vgrids) self.assertEqual(vgrids[self.TEST_VGRID_NAME] [OWNERS], [self.TEST_OWNER_DN]) @@ -545,12 +545,12 @@ def test_resource_map_update(self): vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) - self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) + self.assertIn(self.TEST_RESOURCE_ID, top_vgrid_res) # Check resource map contains resource entry updated_res_map = force_update_resource_map(self.configuration) # Check resource map contains entry - self.assertTrue(self.TEST_RESOURCE_ID in updated_res_map) + self.assertIn(self.TEST_RESOURCE_ID, updated_res_map) def test_settings_inheritance(self): """Test inherited settings propagation through cached maps""" @@ -638,11 +638,12 @@ def test_resource_revoked_access(self): resources=[self.TEST_RESOURCE_ID]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Check vgrid map contains resource entry vgrid_data = initial_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) - self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res) + self.assertIn(self.TEST_RESOURCE_ID, top_vgrid_res) # Check resource map contains resource entry initial_map = force_update_resource_map(self.configuration) @@ -658,7 +659,7 @@ def test_resource_revoked_access(self): vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) top_vgrid_res = top_vgrid_data.get(RESOURCES, {}) - self.assertFalse(self.TEST_RESOURCE_ID in top_vgrid_res) + self.assertNotIn(self.TEST_RESOURCE_ID, top_vgrid_res) # Verify resource still in resource map updated_map = force_update_resource_map(self.configuration) @@ -668,12 +669,14 @@ def test_non_recursive_inheritance(self): """Verify non-recursive map excludes nested vgrids""" # Create parent+child vgrids parent_vgrid = 'parent' - self._create_vgrid(parent_vgrid, [self.TEST_OWNER_DN]) + self._create_vgrid(parent_vgrid, owners=[self.TEST_OWNER_DN]) child_vgrid = os.path.join(parent_vgrid, 'child') - self._create_vgrid(child_vgrid, None, [self.TEST_MEMBER_DN]) + self._create_vgrid(child_vgrid, members=[self.TEST_MEMBER_DN]) # Force update to avoid auto caching and get non-recursive map - vgrid_map = force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(parent_vgrid, initial_vgrid_map.get(VGRIDS, {})) + self.assertIn(child_vgrid, initial_vgrid_map.get(VGRIDS, {})) vgrid_map = get_vgrid_map(self.configuration, recursive=False) # Parent should appear self.assertIn(parent_vgrid, vgrid_map.get(VGRIDS, {})) @@ -698,6 +701,8 @@ def test_hidden_setting_propagation(self): # Verify parent remains visible in cache updated_map = force_update_vgrid_map(self.configuration) + self.assertIn(parent_vgrid, updated_map.get(VGRIDS, {})) + self.assertIn(child_vgrid, updated_map.get(VGRIDS, {})) parent_data = updated_map.get(VGRIDS, {}).get(parent_vgrid, {}) parent_settings = dict(parent_data.get(SETTINGS, [])) self.assertNotEqual(parent_settings.get('hidden'), True) @@ -708,6 +713,7 @@ def test_default_vgrid_access(self): members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Even non-member should have access to default vgrid participant = check_vgrid_access(self.configuration, @@ -716,7 +722,7 @@ def test_default_vgrid_access(self): self.assertFalse(participant) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN) - self.assertTrue('Generic' in allowed_vgrids) + self.assertIn('Generic', allowed_vgrids) # Invalid vgrid should not allow any participation or access participant = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, @@ -724,7 +730,7 @@ def test_default_vgrid_access(self): self.assertFalse(participant) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_MEMBER_DN) - self.assertFalse('invalid-vgrid-name' in allowed_vgrids) + self.assertNotIn('invalid-vgrid-name', allowed_vgrids) def test_general_vgrid_access(self): """Verify general access rules for vgrids""" @@ -732,6 +738,7 @@ def test_general_vgrid_access(self): members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Test vgrid must allow owner and members access allowed = check_vgrid_access(self.configuration, self.TEST_OWNER_DN, @@ -739,14 +746,14 @@ def test_general_vgrid_access(self): self.assertTrue(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OWNER_DN) - self.assertTrue(self.TEST_VGRID_NAME in allowed_vgrids) + self.assertIn(self.TEST_VGRID_NAME, allowed_vgrids) allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, self.TEST_VGRID_NAME) self.assertTrue(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_MEMBER_DN) - self.assertTrue(self.TEST_VGRID_NAME in allowed_vgrids) + self.assertIn(self.TEST_VGRID_NAME, allowed_vgrids) # Test vgrid must reject allow outsider access allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, @@ -754,7 +761,7 @@ def test_general_vgrid_access(self): self.assertFalse(allowed) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN) - self.assertFalse(self.TEST_VGRID_NAME in allowed_vgrids) + self.assertNotIn(self.TEST_VGRID_NAME, allowed_vgrids) def test_user_allowed_res_confs(self): """Minimal test for user_allowed_res_confs""" @@ -763,7 +770,8 @@ def test_user_allowed_res_confs(self): self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) # Owner should be allowed access allowed = user_allowed_res_confs(self.configuration, @@ -851,12 +859,13 @@ def test_access_nonexistent_vgrid(self): # Should not appear in allowed vgrids allowed_vgrids = user_vgrid_access( self.configuration, self.TEST_MEMBER_DN) - self.assertFalse('no-such-vgrid' in allowed_vgrids) + self.assertNotIn('no-such-vgrid', allowed_vgrids) def test_empty_member_access(self): """Verify members-only vgrid rejects outsiders""" self._create_vgrid(self.TEST_VGRID_NAME, [], [self.TEST_MEMBER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Outsider should be blocked despite no owners allowed = check_vgrid_access(self.configuration, self.TEST_OUTSIDER_DN, From b5a0ae2d420f6953753bc3ef9a6b898c9ce7e9ac Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 13:24:06 +0100 Subject: [PATCH 16/20] More integrity self-checks. --- tests/test_mig_shared_vgridaccess.py | 108 +++++++++++++++++++-------- 1 file changed, 77 insertions(+), 31 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 213164b6f..5b5afe7be 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -144,6 +144,13 @@ def _reset_caches(self): vgridaccess.last_load[field] = 0 vgridaccess.last_map[field].clear() + def _verify_vgrid_map_integrity(self, vgrid_map): + """Assure integrity of all vgrid map entries""" + for vgrid_name in vgrid_map.get(VGRIDS, {}): + for field in (OWNERS, MEMBERS, SETTINGS, RESOURCES): + vgrid_entry = vgrid_map[VGRIDS][vgrid_name] + self.assertIn(field, vgrid_entry) + def before_each(self): """Create clean test environment for vgridaccess tests""" used_state_dirs = [ @@ -174,10 +181,7 @@ def before_each(self): self.assertEqual(vgrid_map.get(VGRIDS, {}), self.MINIMAL_VGRIDS) # Make sure all vgrid map entries are complete - for vgrid_name in vgrid_map.get(VGRIDS, {}): - for field in (OWNERS, MEMBERS, SETTINGS, RESOURCES): - vgrid_entry = vgrid_map[VGRIDS][vgrid_name] - self.assertIn(field, vgrid_entry) + self._verify_vgrid_map_integrity(vgrid_map) res_map = force_update_resource_map(self.configuration) # Make sure no resource map entries exist initially @@ -226,11 +230,12 @@ def test_force_update_vgrid_map(self): self.assertEqual(vgrid_map_before.get(VGRIDS, {}), self.MINIMAL_VGRIDS) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - updated_vgrid = force_update_vgrid_map(self.configuration) - self.assertTrue(updated_vgrid) + updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertTrue(updated_vgrid_map) self.assertNotEqual(len(vgrid_map_before.get(VGRIDS, {})), - len(updated_vgrid.get(VGRIDS, {}))) - self.assertIn(self.TEST_VGRID_NAME, updated_vgrid.get(VGRIDS, {})) + len(updated_vgrid_map.get(VGRIDS, {}))) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid_map.get(VGRIDS, {})) def test_refresh_user_map(self): """Minimal test for user map refresh functionality""" @@ -267,7 +272,9 @@ def test_get_resource_map(self): def test_get_vgrid_map(self): """Minimal test for user map refresh functionality""" self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) vgrid_map = get_vgrid_map(self.configuration) self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) @@ -313,7 +320,10 @@ def test_load_vgrid_map(self): # Add test vgrid now = time.time() self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) + # Verify updated map contains vgrid after refresh vgrid_map, map_stamp = load_vgrid_map(self.configuration) self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) @@ -349,7 +359,9 @@ def test_user_allowed_res_units(self): """Test user_allowed_res_units returns allowed units""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) allowed = user_allowed_res_units( self.configuration, self.TEST_OWNER_DN, "exe") @@ -360,7 +372,9 @@ def test_user_allowed_res_exes(self): """Test user_allowed_res_exes returns allowed exes""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) allowed = user_allowed_res_exes(self.configuration, self.TEST_OWNER_DN) self.assertTrue(isinstance(allowed, dict)) @@ -370,7 +384,9 @@ def test_user_allowed_res_stores(self): """Test user_allowed_res_stores returns allowed stores""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) allowed = user_allowed_res_stores( self.configuration, self.TEST_OWNER_DN) @@ -381,7 +397,9 @@ def test_user_visible_res_exes(self): """Test user_visible_res_exes returns visible exes""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) visible = user_visible_res_exes(self.configuration, self.TEST_OWNER_DN) self.assertTrue(isinstance(visible, dict)) @@ -391,7 +409,9 @@ def test_user_visible_res_stores(self): """Test user_visible_res_stores returns visible stores""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) visible = user_visible_res_stores( self.configuration, self.TEST_OWNER_DN) @@ -404,7 +424,9 @@ def test_user_allowed_user_confs(self): self._provision_test_user(self, self.TEST_USER_DN) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN], [self.TEST_USER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_user_map(self.configuration) allowed = user_allowed_user_confs( self.configuration, self.TEST_OWNER_DN) @@ -484,7 +506,9 @@ def test_user_vgrid_access(self): self._provision_test_user(self, self.TEST_OWNER_DN) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) # Refresh maps to reflect new content - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) allowed_vgrids = user_vgrid_access(self.configuration, self.TEST_OWNER_DN) self.assertIn(self.TEST_VGRID_NAME, allowed_vgrids) @@ -500,7 +524,9 @@ def test_res_vgrid_access(self): self._create_vgrid(self.TEST_VGRID_NAME, resources=[ self.TEST_RESOURCE_ID]) # Refresh maps to reflect new content - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) allowed = res_vgrid_access(self.configuration, self.TEST_RESOURCE_ID) self.assertIn(self.TEST_VGRID_NAME, allowed) @@ -512,8 +538,9 @@ def test_vgrid_map_refresh(self): self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) # Force refresh map - updated_map = force_update_vgrid_map(self.configuration) - vgrids = updated_map.get(VGRIDS, {}) + updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + vgrids = updated_vgrid_map.get(VGRIDS, {}) self.assertIn(self.TEST_VGRID_NAME, vgrids) self.assertEqual(vgrids[self.TEST_VGRID_NAME] [OWNERS], [self.TEST_OWNER_DN]) @@ -523,7 +550,9 @@ def test_user_map_access(self): # Add user as member self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], members=[self.TEST_MEMBER_DN]) - force_update_vgrid_map(self.configuration) + initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Verify member access allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN, self.TEST_VGRID_NAME) @@ -541,6 +570,8 @@ def test_resource_map_update(self): self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid_map.get(VGRIDS, {})) # Check vgrid map contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) @@ -563,10 +594,12 @@ def test_settings_inheritance(self): self._create_vgrid(sub_vgrid) # Force refresh of cached map - updated_map = force_update_vgrid_map(self.configuration) + updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid_map.get(VGRIDS, {})) # Retrieve vgrid data from cached map - vgrid_data = updated_map.get(VGRIDS, {}) + vgrid_data = updated_vgrid_map.get(VGRIDS, {}) self.assertTrue(vgrid_data) # Retrieve top vgrid settings from cached map @@ -587,7 +620,8 @@ def test_settings_inheritance(self): # Verify hidden setting unset without inheritance self.assertFalse(sub_settings_dict.get('hidden')) - inherited_map = vgrid_inherit_map(self.configuration, updated_map) + inherited_map = vgrid_inherit_map( + self.configuration, updated_vgrid_map) vgrid_data = inherited_map.get(VGRIDS, {}) self.assertTrue(vgrid_data) @@ -606,7 +640,9 @@ def test_unmap_inheritance(self): self._create_vgrid(sub_vgrid) # Force refresh of cached map - updated_map = force_update_vgrid_map(self.configuration) + updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid_map.get(VGRIDS, {})) # Unmap and verify mark modified unmap_inheritance(self.configuration, self.TEST_VGRID_NAME, @@ -622,6 +658,7 @@ def test_user_map_fields(self): self._provision_test_user(self, self.TEST_USER_DN) # Force fresh user map initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) user_map = force_update_user_map(self.configuration) test_owner = user_map.get(self.TEST_OWNER_DN, {}) self.assertEqual(test_owner.get(USERID), self.TEST_OWNER_UUID) @@ -638,6 +675,7 @@ def test_resource_revoked_access(self): resources=[self.TEST_RESOURCE_ID]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Check vgrid map contains resource entry vgrid_data = initial_vgrid_map.get(VGRIDS, {}) @@ -655,6 +693,8 @@ def test_resource_revoked_access(self): self.assertEqual(success_and_msg, (True, '')) updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertIn(self.TEST_VGRID_NAME, updated_vgrid_map.get(VGRIDS, {})) # Check vgrid map no longer contains resource entry vgrid_data = updated_vgrid_map.get(VGRIDS, {}) top_vgrid_data = vgrid_data.get(self.TEST_VGRID_NAME, {}) @@ -662,8 +702,8 @@ def test_resource_revoked_access(self): self.assertNotIn(self.TEST_RESOURCE_ID, top_vgrid_res) # Verify resource still in resource map - updated_map = force_update_resource_map(self.configuration) - self.assertIn(self.TEST_RESOURCE_ID, updated_map) + updated_res_map = force_update_resource_map(self.configuration) + self.assertIn(self.TEST_RESOURCE_ID, updated_res_map) def test_non_recursive_inheritance(self): """Verify non-recursive map excludes nested vgrids""" @@ -675,6 +715,7 @@ def test_non_recursive_inheritance(self): # Force update to avoid auto caching and get non-recursive map initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(parent_vgrid, initial_vgrid_map.get(VGRIDS, {})) self.assertIn(child_vgrid, initial_vgrid_map.get(VGRIDS, {})) vgrid_map = get_vgrid_map(self.configuration, recursive=False) @@ -700,10 +741,11 @@ def test_hidden_setting_propagation(self): ('hidden', True)]) # Verify parent remains visible in cache - updated_map = force_update_vgrid_map(self.configuration) - self.assertIn(parent_vgrid, updated_map.get(VGRIDS, {})) - self.assertIn(child_vgrid, updated_map.get(VGRIDS, {})) - parent_data = updated_map.get(VGRIDS, {}).get(parent_vgrid, {}) + updated_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(updated_vgrid_map) + self.assertIn(parent_vgrid, updated_vgrid_map.get(VGRIDS, {})) + self.assertIn(child_vgrid, updated_vgrid_map.get(VGRIDS, {})) + parent_data = updated_vgrid_map.get(VGRIDS, {}).get(parent_vgrid, {}) parent_settings = dict(parent_data.get(SETTINGS, [])) self.assertNotEqual(parent_settings.get('hidden'), True) @@ -713,6 +755,7 @@ def test_default_vgrid_access(self): members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Even non-member should have access to default vgrid @@ -738,6 +781,7 @@ def test_general_vgrid_access(self): members=[self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Test vgrid must allow owner and members access @@ -771,6 +815,7 @@ def test_user_allowed_res_confs(self): self._create_vgrid(self.TEST_VGRID_NAME, owners=[self.TEST_OWNER_DN], resources=[self.TEST_RESOURCE_ID]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) force_update_resource_map(self.configuration) # Owner should be allowed access @@ -865,6 +910,7 @@ def test_empty_member_access(self): """Verify members-only vgrid rejects outsiders""" self._create_vgrid(self.TEST_VGRID_NAME, [], [self.TEST_MEMBER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) + self._verify_vgrid_map_integrity(initial_vgrid_map) self.assertIn(self.TEST_VGRID_NAME, initial_vgrid_map.get(VGRIDS, {})) # Outsider should be blocked despite no owners From 96c4f887827381fda80db884170cb14acf96b98e Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 14:22:39 +0100 Subject: [PATCH 17/20] Pull in PR #380 to see if that fixes the spurious CI test failures as speculated in that PR description. We would prefer to avoid mixing them up in this test-only PR but unfortunately I'm completely unable to trigger similar failures locally. So if it solves the problems I'll merge PR #380 and rebase this one on it before merge to keep them apart. --- mig/shared/vgridaccess.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/mig/shared/vgridaccess.py b/mig/shared/vgridaccess.py index 2c9f46e07..054ae27db 100644 --- a/mig/shared/vgridaccess.py +++ b/mig/shared/vgridaccess.py @@ -401,18 +401,18 @@ def refresh_vgrid_map(configuration, clean=False): optional_conf = [SETTINGS, ] for vgrid in all_vgrids: + # Make sure vgrid dict exists before filling it + vgrid_map[VGRIDS][vgrid] = vgrid_map[VGRIDS].get(vgrid, {}) for (field, name, list_call) in conf_read: conf_path = os.path.join(configuration.vgrid_home, vgrid, name) if not os.path.isfile(conf_path): - # Make sure vgrid dict exists before filling it - vgrid_map[VGRIDS][vgrid] = vgrid_map[VGRIDS].get(vgrid, {}) vgrid_map[VGRIDS][vgrid][field] = [] if vgrid != default_vgrid and field not in optional_conf: _logger.warning('missing file: %s' % conf_path) dirty[VGRIDS] = dirty.get(VGRIDS, []) + [vgrid] - elif vgrid not in vgrid_map[VGRIDS] or \ + elif field not in vgrid_map[VGRIDS][vgrid] or \ os.path.getmtime(conf_path) >= map_stamp: (status, entries) = list_call(vgrid, configuration, recursive=False) @@ -425,6 +425,8 @@ def refresh_vgrid_map(configuration, clean=False): vgrid_map[VGRIDS][vgrid] = map_entry vgrid_map[VGRIDS][vgrid][field] = entries dirty[VGRIDS] = dirty.get(VGRIDS, []) + [vgrid] + else: + _logger.debug("caching %s for %s in vgrid map" % (name, vgrid)) # Remove any missing vgrids from map missing_vgrids = [vgrid for vgrid in vgrid_map[VGRIDS] if not vgrid in all_vgrids] @@ -853,7 +855,6 @@ def check_vgrid_access(configuration, client_id, vgrid_name, recursive=True, the vgrid module and should replace that one everywhere that only vgrid map (cached) lookups are needed. """ - vgrid_access = [default_vgrid] vgrid_map = get_vgrid_map(configuration, recursive, caching) vgrid_entry = vgrid_map.get(VGRIDS, {}).get( vgrid_name, {OWNERS: [], MEMBERS: []}) From 5d825fd0a1ca4b1d9f1b25cab46a8b02ea7f0aa9 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 21:38:12 +0100 Subject: [PATCH 18/20] Minor docstring adjustments to fit tested function. --- tests/test_mig_shared_vgridaccess.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 5b5afe7be..9f418bbc6 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -200,7 +200,7 @@ def before_each(self): self.assertEqual(user_map, vgridaccess.last_map[USERS]) def test_force_update_user_map(self): - """Simple test that user map refresh completes""" + """Simple test that forced user map update completes""" # Verify empty map on init user_map_before, _ = load_user_map(self.configuration) self.assertEqual(user_map_before, {}) @@ -212,7 +212,7 @@ def test_force_update_user_map(self): self.assertIn(self.TEST_USER_DN, updated_users) def test_force_update_resource_map(self): - """Simple test that resource map refresh completes""" + """Simple test that forced resource map update completes""" # Verify empty map on init res_map_before, _ = load_resource_map(self.configuration) self.assertEqual(res_map_before, {}) @@ -224,7 +224,7 @@ def test_force_update_resource_map(self): self.assertIn(self.TEST_RESOURCE_ID, updated_res) def test_force_update_vgrid_map(self): - """Simple test that vgrid map refresh completes""" + """Simple test that forced vgrid map update completes""" # Only (implicit) default vgrid in vgrid map before init vgrid_map_before, _ = load_vgrid_map(self.configuration) self.assertEqual(vgrid_map_before.get(VGRIDS, {}), self.MINIMAL_VGRIDS) @@ -256,14 +256,14 @@ def test_refresh_vgrid_map(self): self.assertIn(self.TEST_VGRID_NAME, vgrid_map.get(VGRIDS, {})) def test_get_user_map(self): - """Minimal test for user map refresh functionality""" + """Minimal test for get user map functionality""" self._provision_test_user(self, self.TEST_USER_DN) force_update_user_map(self.configuration) user_map = get_user_map(self.configuration) self.assertIn(self.TEST_USER_DN, user_map) def test_get_resource_map(self): - """Minimal test for user map refresh functionality""" + """Minimal test for get user map functionality""" self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) force_update_resource_map(self.configuration) resource_map = get_resource_map(self.configuration) From 3b2355074a81d14da12ceac8e98973a6291d595c Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 22:14:15 +0100 Subject: [PATCH 19/20] It looks like the initial PR #380 adjustment to `VGRIDS` update loop may in fact have fixed the spurious `OWNERS` errors in CI unit tests. Extend the vgridaccess fix from PR #380 to apply similar fixes in the the `RESOURCES` and `USER` update loop of refresh_vgrid_map. --- mig/shared/vgridaccess.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/mig/shared/vgridaccess.py b/mig/shared/vgridaccess.py index 054ae27db..2988d54a0 100644 --- a/mig/shared/vgridaccess.py +++ b/mig/shared/vgridaccess.py @@ -451,7 +451,8 @@ def refresh_vgrid_map(configuration, clean=False): conf_path = os.path.join(configuration.resource_home, res, "config") if not os.path.isfile(conf_path): continue - if os.path.getmtime(conf_path) >= map_stamp: + if res not in vgrid_map[RESOURCES] or \ + os.path.getmtime(conf_path) >= map_stamp: # Read maps of exe name to vgrid list and of store name to vgrid # list. Save them separately to be able to distinguish them in # exe / store access and visibility @@ -562,7 +563,7 @@ def refresh_vgrid_map(configuration, clean=False): else: conf_mtime = -1 user_conf = {} - if conf_mtime >= map_stamp: + if user not in vgrid_map[USERS] or conf_mtime >= map_stamp: vgrid_map[USERS][user] = user_conf vgrid_map[USERS][user][ASSIGN] = vgrid_map[USERS][user].get(ASSIGN, []) From 8d8de651c5e795e2ff531a5f7dca96cd1aec17d9 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 2 Nov 2025 22:34:38 +0100 Subject: [PATCH 20/20] Comment-only change to explain the implicit resource assignment to Generic. --- tests/test_mig_shared_vgridaccess.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_mig_shared_vgridaccess.py b/tests/test_mig_shared_vgridaccess.py index 9f418bbc6..b69c8d220 100644 --- a/tests/test_mig_shared_vgridaccess.py +++ b/tests/test_mig_shared_vgridaccess.py @@ -357,6 +357,7 @@ def test_user_owned_res_stores(self): def test_user_allowed_res_units(self): """Test user_allowed_res_units returns allowed units""" + # NOTE: create assigns TEST_RESOURCE_ID to default vgrid self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) @@ -370,6 +371,7 @@ def test_user_allowed_res_units(self): def test_user_allowed_res_exes(self): """Test user_allowed_res_exes returns allowed exes""" + # NOTE: create assigns TEST_RESOURCE_ID to default vgrid self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration) @@ -382,6 +384,7 @@ def test_user_allowed_res_exes(self): def test_user_allowed_res_stores(self): """Test user_allowed_res_stores returns allowed stores""" + # NOTE: create assigns TEST_RESOURCE_ID to default vgrid self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN]) self._create_vgrid(self.TEST_VGRID_NAME, [self.TEST_OWNER_DN]) initial_vgrid_map = force_update_vgrid_map(self.configuration)