Skip to content

Commit a651b4c

Browse files
jonasbardinojonasbardino
committed
Fundamental test coverage of vgridaccess.py.
1 parent 6e3df14 commit a651b4c

File tree

1 file changed

+218
-0
lines changed

1 file changed

+218
-0
lines changed

‎tests/test_mig_shared_vgridaccess.py‎

Lines changed: 218 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,218 @@
1+
# -*- coding: utf-8 -*-
2+
#
3+
# --- BEGIN_HEADER ---
4+
#
5+
# test_mig_shared_vgridaccess - unit tests for vgridaccess helper functions
6+
# Copyright (C) 2003-2025 The MiG Project by the Science HPC Center at UCPH
7+
#
8+
# This file is part of MiG.
9+
#
10+
# MiG is free software: you can redistribute it and/or modify
11+
# it under the terms of the GNU General Public License as published by
12+
# the Free Software Foundation; either version 2 of the License, or
13+
# (at your option) any later version.
14+
#
15+
# MiG is distributed in the hope that it will be useful,
16+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
17+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18+
# GNU General Public License for more details.
19+
#
20+
# You should have received a copy of the GNU General Public License
21+
# along with this program; if not, write to the Free Software
22+
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
23+
#
24+
# -- END_HEADER ---
25+
#
26+
27+
"""Unit tests for vgridaccess helper module"""
28+
29+
import os
30+
import time
31+
import unittest
32+
33+
from mig.shared.fileio import read_file, pickle
34+
from mig.shared.vgrid import vgrid_list, vgrid_set_entities, vgrid_settings
35+
from mig.shared.vgridaccess import OWNERS, SETTINGS, VGRIDS, RESOURCES, \
36+
check_vgrid_access, force_update_resource_map, force_update_user_map, \
37+
force_update_vgrid_map, get_resource_map, get_vgrid_map, \
38+
load_resource_map, refresh_vgrid_map, vgrid_inherit_map
39+
from tests.support import MigTestCase, ensure_dirs_exist, testmain
40+
41+
42+
class TestMigSharedVgridAccess(MigTestCase):
43+
"""Unit tests for vgridaccess related helper functions"""
44+
45+
TEST_OWNER_DN = \
46+
'/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Owner/'\
47+
'emailAddress=owner@example.org'
48+
TEST_MEMBER_DN = \
49+
'/C=DK/ST=NA/L=NA/O=Test Org/OU=NA/CN=Test Member/'\
50+
'emailAddress=member@example.org'
51+
TEST_RESOURCE_ID = 'test.example.org.0'
52+
53+
def _provide_configuration(self):
54+
"""Prepare isolated test config"""
55+
return 'testconfig'
56+
57+
def _create_vgrid(self, vgrid_name, owners=None, members=None,
58+
resources=None, settings=None):
59+
"""Helper to create valid skeleton vgrid for testing"""
60+
vgrid_path = os.path.join(self.configuration.vgrid_home, vgrid_name)
61+
ensure_dirs_exist(vgrid_path)
62+
if owners is None:
63+
owners = []
64+
# Add vgrid owners
65+
status, msg = vgrid_set_entities(self.configuration, vgrid_name,
66+
'owners', owners, allow_empty=True)
67+
self.assertTrue(status, msg)
68+
if members is not None:
69+
status, msg = vgrid_set_entities(self.configuration, vgrid_name,
70+
'members', members,
71+
allow_empty=False)
72+
self.assertTrue(status, msg)
73+
if resources is not None:
74+
status, msg = vgrid_set_entities(self.configuration, vgrid_name,
75+
'resources', resources,
76+
allow_empty=False)
77+
self.assertTrue(status, msg)
78+
if settings is not None:
79+
status, msg = vgrid_set_entities(self.configuration, vgrid_name,
80+
'settings', settings,
81+
allow_empty=False)
82+
self.assertTrue(status, msg)
83+
84+
def _create_resource(self, res_name, owners, config=None):
85+
"""Helper to create valid skeleton resource for testing"""
86+
res_path = os.path.join(self.configuration.resource_home, res_name)
87+
res_owners_path = os.path.join(res_path, 'owners')
88+
res_config_path = os.path.join(res_path, 'config')
89+
# Add resource skeleton with owners
90+
ensure_dirs_exist(res_path)
91+
if owners is None:
92+
owners = []
93+
saved = pickle(owners, res_owners_path, self.logger)
94+
self.assertTrue(saved)
95+
if config is None:
96+
config = {}
97+
saved = pickle(config, res_config_path, self.logger)
98+
self.assertTrue(saved)
99+
100+
def before_each(self):
101+
"""Create test environment for vgridaccess tests"""
102+
self._provision_test_user(self, self.TEST_OWNER_DN)
103+
ensure_dirs_exist(self.configuration.mig_system_files)
104+
ensure_dirs_exist(self.configuration.mig_system_run)
105+
ensure_dirs_exist(self.configuration.user_home)
106+
ensure_dirs_exist(self.configuration.user_settings)
107+
ensure_dirs_exist(self.configuration.vgrid_home)
108+
ensure_dirs_exist(self.configuration.resource_home)
109+
# Start with empty maps
110+
force_update_vgrid_map(self.configuration, clean=True)
111+
force_update_user_map(self.configuration, clean=True)
112+
force_update_resource_map(self.configuration, clean=True)
113+
114+
self.test_vgrid = 'testvgrid'
115+
116+
def test_vgrid_map_refresh(self):
117+
"""Verify vgrid map refresh captures changes"""
118+
# We always init empty maps
119+
initial_map = get_vgrid_map(self.configuration, recursive=False)
120+
self.assertFalse(self.test_vgrid in initial_map.get(VGRIDS, {}))
121+
122+
self._create_vgrid(self.test_vgrid, [self.TEST_OWNER_DN])
123+
# Force refresh map
124+
updated_map = force_update_vgrid_map(self.configuration)
125+
vgrids = updated_map.get(VGRIDS, {})
126+
self.assertTrue(self.test_vgrid in vgrids)
127+
self.assertEqual(vgrids[self.test_vgrid]
128+
[OWNERS], [self.TEST_OWNER_DN])
129+
130+
def test_user_map_access(self):
131+
"""Test user permissions through cached access maps"""
132+
# Add user as member
133+
self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN],
134+
members=[self.TEST_MEMBER_DN])
135+
force_update_vgrid_map(self.configuration)
136+
# Verify member access
137+
allowed = check_vgrid_access(self.configuration, self.TEST_MEMBER_DN,
138+
self.test_vgrid)
139+
self.assertTrue(allowed)
140+
141+
def test_resource_map_update(self):
142+
"""Verify resource visibility in cache"""
143+
# Check cached resource map does not yet contain entry
144+
cached_map, map_stamp = load_resource_map(self.configuration,
145+
caching=True)
146+
self.assertFalse(cached_map, map_stamp)
147+
self.assertFalse(self.TEST_RESOURCE_ID in cached_map)
148+
149+
# Add vgrid with assigned resource
150+
self._create_resource(self.TEST_RESOURCE_ID, [self.TEST_OWNER_DN])
151+
self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN],
152+
resources=[self.TEST_RESOURCE_ID])
153+
updated_vgrid_map = force_update_vgrid_map(self.configuration,
154+
clean=True)
155+
print("DEBUG: updated vgrid map %s" % updated_vgrid_map)
156+
# Check vgrid map contains resource entry
157+
vgrid_data = updated_vgrid_map.get(VGRIDS, {})
158+
top_vgrid_data = vgrid_data.get(self.test_vgrid, {})
159+
top_vgrid_res = top_vgrid_data.get(RESOURCES, [])
160+
self.assertTrue(self.TEST_RESOURCE_ID in top_vgrid_res)
161+
162+
# Check resource map contains resource entry
163+
updated_res_map = force_update_resource_map(self.configuration,
164+
clean=True)
165+
print("DEBUG: updated res map %s" % updated_res_map)
166+
# Check resource map contains entry
167+
self.assertTrue(self.TEST_RESOURCE_ID in updated_res_map)
168+
169+
def test_settings_inheritance(self):
170+
"""Test inherited settings propagation through cached maps"""
171+
# Create top and sub vgrids with 'hidden' setting on top vgrid
172+
top_settings = [('vgrid_name', self.test_vgrid),
173+
('hidden', True)]
174+
self._create_vgrid(self.test_vgrid, owners=[self.TEST_OWNER_DN],
175+
settings=top_settings)
176+
sub_vgrid = os.path.join(self.test_vgrid, 'subvgrid')
177+
self._create_vgrid(sub_vgrid)
178+
179+
# Force refresh of cached map
180+
updated_map = force_update_vgrid_map(self.configuration)
181+
182+
# Retrieve vgrid data from cached map
183+
vgrid_data = updated_map.get(VGRIDS, {})
184+
self.assertTrue(vgrid_data)
185+
186+
# Retrieve top vgrid settings from cached map
187+
top_vgrid_data = vgrid_data.get(self.test_vgrid, {})
188+
self.assertTrue(top_vgrid_data)
189+
# Convert settings list of tuples to dict
190+
top_settings_dict = dict(top_vgrid_data.get(SETTINGS, []))
191+
self.assertTrue(top_settings_dict)
192+
193+
# Verify hidden setting in cache
194+
self.assertEqual(top_settings_dict.get('hidden'), True)
195+
196+
# Retrieve sub vgrid settings from cached map
197+
sub_vgrid_data = vgrid_data.get(sub_vgrid, {})
198+
# Convert settings list of tuples to dict
199+
sub_settings_dict = dict(sub_vgrid_data.get(SETTINGS, []))
200+
201+
# Verify hidden setting unset without inheritance
202+
self.assertFalse(sub_settings_dict.get('hidden'))
203+
204+
inherited_map = vgrid_inherit_map(self.configuration, updated_map)
205+
vgrid_data = inherited_map.get(VGRIDS, {})
206+
self.assertTrue(vgrid_data)
207+
208+
# Retrieve sub vgrid settings from cached map
209+
sub_vgrid_data = vgrid_data.get(sub_vgrid, {})
210+
# Convert settings list of tuples to dict
211+
sub_settings_dict = dict(sub_vgrid_data.get(SETTINGS, []))
212+
213+
# Verify hidden setting inheritance
214+
self.assertEqual(sub_settings_dict.get('hidden'), True)
215+
216+
217+
if __name__ == '__main__':
218+
testmain()

0 commit comments

Comments
 (0)