From facff1a080fbe79bbf359d696d78647b4bb20439 Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 9 Nov 2025 18:18:39 +0100 Subject: [PATCH 1/2] Basic filemarks function test coverage reusing the overall structure from the other unit tests. --- tests/test_mig_shared_filemarks.py | 135 +++++++++++++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 tests/test_mig_shared_filemarks.py diff --git a/tests/test_mig_shared_filemarks.py b/tests/test_mig_shared_filemarks.py new file mode 100644 index 000000000..6167b580c --- /dev/null +++ b/tests/test_mig_shared_filemarks.py @@ -0,0 +1,135 @@ +# -*- coding: utf-8 -*- +# +# --- BEGIN_HEADER --- +# +# test_mig_shared_filemarks - unit tests for shared filemarks helpers +# 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 mig.shared.filemarks module""" + +import os +import tempfile +import time +import unittest + +from mig.shared.filemarks import update_filemark, get_filemark, reset_filemark +from tests.support import ensure_dirs_exist, MigTestCase, testmain + +TEST_MARKS_DIR = 'TestMarks' +TEST_MARKS_FILE = 'file.mark' + + +class TestMigSharedFilemarks(MigTestCase): + """Test mig.shared.filemarks functions""" + + def _provide_configuration(self): + """Set up isolated test configuration and logger for the tests""" + return 'testconfig' + + def before_each(self): + """Setup fake configuration and temp dir before each test.""" + self.marks_base = os.path.join(self.configuration.mig_system_run, + TEST_MARKS_DIR) + ensure_dirs_exist(os.path.dirname(self.marks_base)) + self.marks_path = os.path.join(self.marks_base, TEST_MARKS_FILE) + + def test_update_filemark_create(self): + """Test update_filemark creates mark files""" + timestamp = time.time() + update_result = update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, timestamp) + self.assertTrue(update_result) + self.assertTrue(os.path.isfile(self.marks_path)) + + def test_update_filemark_timestamp(self): + """Test update_filemark creates files with correct timestamp""" + timestamp = time.time() + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, timestamp) + self.assertTrue(os.path.isfile(self.marks_path)) + self.assertEqual(os.path.getmtime(self.marks_path), timestamp) + + def test_update_filemark_delete(self): + """Test update_filemark deletes mark files with negative timestamp""" + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, time.time()) + delete_result = update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, -1) + self.assertTrue(delete_result) + self.assertFalse(os.path.exists(self.marks_path)) + + def test_get_filemark_existing(self): + """Test get_filemark retrieves timestamp for existing mark""" + timestamp = time.time() + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, timestamp) + retrieved = get_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE) + self.assertEqual(retrieved, timestamp) + + def test_get_filemark_missing(self): + """Test get_filemark returns None for missing mark files""" + retrieved = get_filemark(self.configuration, self.marks_base, + 'missing.mark') + self.assertIsNone(retrieved) + + def test_reset_filemark_single(self): + """Test reset_filemark updates single mark timestamp to 0""" + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, time.time()) + reset_result = reset_filemark(self.configuration, self.marks_base, + [TEST_MARKS_FILE]) + self.assertTrue(reset_result) + retrieved = get_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE) + self.assertEqual(retrieved, 0) + + def test_reset_filemark_delete(self): + """Test reset_filemark deletes marks with delete=True""" + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, time.time()) + reset_result = reset_filemark(self.configuration, self.marks_base, + [TEST_MARKS_FILE], delete=True) + self.assertTrue(reset_result) + retrieved = get_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE) + self.assertIsNone(retrieved) + self.assertFalse(os.path.exists(self.marks_path)) + + def test_reset_filemark_all(self): + """Test reset_filemark resets all marks when mark_list=None""" + marks = ['mark1', 'mark2', 'mark3'] + for mark in marks: + update_filemark(self.configuration, self.marks_base, mark, + time.time()) + + reset_result = reset_filemark(self.configuration, self.marks_base) + self.assertTrue(reset_result) + + for mark in marks: + result = get_filemark(self.configuration, self.marks_base, mark) + self.assertEqual(result, 0) + + +if __name__ == '__main__': + testmain() From cbfb08da869f4debd991700f25da52175ae2251d Mon Sep 17 00:00:00 2001 From: Jonas Bardino Date: Sun, 9 Nov 2025 19:13:58 +0100 Subject: [PATCH 2/2] Increase filemarks coverage with more corner cases. A few are explicitly disabled when run as privileged user like in our doker CI because basic ACLs are not fully enforced in that case. --- tests/test_mig_shared_filemarks.py | 88 +++++++++++++++++++++++++++++- 1 file changed, 87 insertions(+), 1 deletion(-) diff --git a/tests/test_mig_shared_filemarks.py b/tests/test_mig_shared_filemarks.py index 6167b580c..0f7a1ca53 100644 --- a/tests/test_mig_shared_filemarks.py +++ b/tests/test_mig_shared_filemarks.py @@ -28,6 +28,8 @@ """Unit tests for mig.shared.filemarks module""" import os +import shutil +import stat import tempfile import time import unittest @@ -50,7 +52,7 @@ def before_each(self): """Setup fake configuration and temp dir before each test.""" self.marks_base = os.path.join(self.configuration.mig_system_run, TEST_MARKS_DIR) - ensure_dirs_exist(os.path.dirname(self.marks_base)) + ensure_dirs_exist(self.marks_base) self.marks_path = os.path.join(self.marks_base, TEST_MARKS_FILE) def test_update_filemark_create(self): @@ -130,6 +132,90 @@ def test_reset_filemark_all(self): result = get_filemark(self.configuration, self.marks_base, mark) self.assertEqual(result, 0) + def test_update_filemark_fails_when_file_prevents_directory(self): + """Test update_filemark fails when file prevents create directory""" + # Create a file in the way to prevent subdir creation + result = update_filemark(self.configuration, self.marks_base, + 'obstruct', time.time()) + self.assertTrue(result) + result = update_filemark(self.configuration, self.marks_base, + os.path.join('obstruct', 'test.mark'), + time.time()) + self.assertFalse(result) + + @unittest.skipIf(os.getuid() == 0, "enable when not running as priv user") + def test_update_filemark_directory_perms_failure(self): + """Test update_filemark fails on directory creation failure""" + # Create a read-only parent directory to prevent subdir creation + # TODO: permissions are not enforced if running as a privileged user + # like it's the case for our current docker CI + os.chmod(self.marks_base, stat.S_IRUSR) # Remove write permissions + + result = update_filemark(self.configuration, self.marks_base, + os.path.join('noaccess', 'test.mark'), + time.time()) + self.assertFalse(result) + + @unittest.skipIf(os.getuid() == 0, "enable when not running as priv user") + def test_get_filemark_permission_denied(self): + """Test get_filemark returns None when permission denied""" + # Create valid file first + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, time.time()) + + # Remove read permissions through parent dir + os.chmod(self.marks_base, 0) + result = get_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE) + self.assertIsNone(result) + # Restore permissions so cleanup works + os.chmod(self.marks_base, stat.S_IRWXU) + + def test_reset_filemark_string_mark_list(self): + """Test reset_filemark handles single string mark_list""" + update_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE, time.time()) + reset_result = reset_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE) + self.assertTrue(reset_result) + self.assertEqual(get_filemark(self.configuration, self.marks_base, + TEST_MARKS_FILE), 0) + + def test_reset_filemark_invalid_mark_list(self): + """Test reset_filemark fails with invalid mark_list type""" + reset_result = reset_filemark(self.configuration, self.marks_base, + {'invalid': 'type'}) + self.assertFalse(reset_result) + + def test_reset_filemark_all_missing_dir(self): + """Test reset_filemark handles missing directory when mark_list=None""" + shutil.rmtree(self.marks_base) # Ensure directory doesn't exist + reset_result = reset_filemark(self.configuration, self.marks_base) + self.assertFalse(reset_result) + + def test_reset_filemark_partial_failure(self): + """Test reset_filemark with partial success""" + valid_mark = 'valid.mark' + # invalid_mark = os.path.join('restricted_dir', 'invalid.mark') + invalid_mark = 'invalid.mark' + invalid_path = os.path.join(self.marks_base, invalid_mark) + ensure_dirs_exist(invalid_path) + + # Create both marks but remove access to the latter + update_filemark(self.configuration, self.marks_base, valid_mark, + time.time()) + update_filemark(self.configuration, self.marks_base, invalid_mark, + time.time()) + os.chmod(invalid_path, stat.S_IRUSR) # Remove write permissions + + reset_result = reset_filemark(self.configuration, self.marks_base, + [valid_mark, invalid_mark]) + self.assertFalse(reset_result) # Should fail due to partial failure + + # Verify valid mark was still reset + self.assertEqual(get_filemark(self.configuration, + self.marks_base, valid_mark), 0) + if __name__ == '__main__': testmain()