From cb1b51d4e9be44ef8f4c749b5fbbf224f1552c8e Mon Sep 17 00:00:00 2001 From: Sam Rabin Date: Tue, 13 Jan 2026 16:29:00 -0700 Subject: [PATCH 1/5] Add tests for printed messages. --- tests/test_relink.py | 119 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 118 insertions(+), 1 deletion(-) diff --git a/tests/test_relink.py b/tests/test_relink.py index c694ff7..03fe3be 100644 --- a/tests/test_relink.py +++ b/tests/test_relink.py @@ -86,7 +86,7 @@ def test_nested_directory_structure(self, temp_dirs, current_user): assert os.path.islink(source_file), "Nested file should be a symlink" assert os.readlink(source_file) == target_file - def test_skip_existing_symlinks(self, temp_dirs, current_user): + def test_skip_existing_symlinks(self, temp_dirs, current_user, capsys): """Test that existing symlinks are skipped.""" source_dir, target_dir = temp_dirs username = current_user @@ -119,6 +119,11 @@ def test_skip_existing_symlinks(self, temp_dirs, current_user): os.readlink(source_link) == dummy_target ), "Symlink target should be unchanged" + # Check that "Skipping symlink" message was printed + captured = capsys.readouterr() + assert "Skipping symlink:" in captured.out + assert source_link in captured.out + def test_missing_target_file(self, temp_dirs, current_user, capsys): """Test behavior when target file doesn't exist.""" source_dir, target_dir = temp_dirs @@ -215,6 +220,64 @@ def test_absolute_paths(self, temp_dirs, current_user): finally: os.chdir(cwd) + def test_print_searching_message(self, temp_dirs, current_user, capsys): + """Test that searching message is printed.""" + source_dir, target_dir = temp_dirs + username = current_user + + # Run the function + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Check that searching message was printed + captured = capsys.readouterr() + assert f"Searching for files owned by '{username}'" in captured.out + assert f"in '{os.path.abspath(source_dir)}'" in captured.out + + def test_print_found_owned_file(self, temp_dirs, current_user, capsys): + """Test that 'Found owned file' message is printed.""" + source_dir, target_dir = temp_dirs + username = current_user + + # Create a file owned by current user + source_file = os.path.join(source_dir, "owned_file.txt") + target_file = os.path.join(target_dir, "owned_file.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("content") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target content") + + # Run the function + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Check that "Found owned file" message was printed + captured = capsys.readouterr() + assert "Found owned file:" in captured.out + assert source_file in captured.out + + def test_print_deleted_and_created_messages(self, temp_dirs, current_user, capsys): + """Test that deleted and created symlink messages are printed.""" + source_dir, target_dir = temp_dirs + username = current_user + + # Create files + source_file = os.path.join(source_dir, "test_file.txt") + target_file = os.path.join(target_dir, "test_file.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("source") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target") + + # Run the function + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Check messages + captured = capsys.readouterr() + assert "Deleted original file:" in captured.out + assert "Created symbolic link:" in captured.out + assert f"{source_file} -> {target_file}" in captured.out + class TestParseArguments: """Test suite for parse_arguments function.""" @@ -323,3 +386,57 @@ def test_file_with_special_characters(self, temp_dirs): # Verify assert os.path.islink(source_file) assert os.readlink(source_file) == target_file + + def test_error_deleting_file(self, temp_dirs, capsys): + """Test error message when file deletion fails.""" + source_dir, target_dir = temp_dirs + username = os.environ["USER"] + + # Create files + source_file = os.path.join(source_dir, "test.txt") + target_file = os.path.join(target_dir, "test.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("source") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target") + + # Mock os.rename to raise an error + def mock_rename(src, dst): + raise OSError("Simulated rename error") + + with patch("os.rename", side_effect=mock_rename): + # Run the function + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Check error message + captured = capsys.readouterr() + assert "Error deleting file" in captured.out + assert source_file in captured.out + + def test_error_creating_symlink(self, temp_dirs, capsys): + """Test error message when symlink creation fails.""" + source_dir, target_dir = temp_dirs + username = os.environ["USER"] + + # Create source file + source_file = os.path.join(source_dir, "test.txt") + target_file = os.path.join(target_dir, "test.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("source") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target") + + # Mock os.symlink to raise an error + def mock_symlink(src, dst): + raise OSError("Simulated symlink error") + + with patch("os.symlink", side_effect=mock_symlink): + # Run the function + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Check error message + captured = capsys.readouterr() + assert "Error creating symlink" in captured.out + assert source_file in captured.out From 96915309720b899613bfc23d5657c2ab6f2e69ba Mon Sep 17 00:00:00 2001 From: Sam Rabin Date: Tue, 13 Jan 2026 16:48:22 -0700 Subject: [PATCH 2/5] relink.py: Use Python logging instead of print() statements. --- relink.py | 40 ++++++++++++---- tests/test_relink.py | 107 +++++++++++++++++++++++++------------------ 2 files changed, 94 insertions(+), 53 deletions(-) diff --git a/relink.py b/relink.py index f1f3ed0..a436838 100644 --- a/relink.py +++ b/relink.py @@ -7,10 +7,14 @@ import os import pwd import argparse +import logging DEFAULT_SOURCE_ROOT = '/glade/campaign/cesm/cesmdata/cseg/inputdata/' DEFAULT_TARGET_ROOT = '/glade/campaign/collections/gdex/data/d651077/cesmdata/inputdata/' +# Set up logger +logger = logging.getLogger(__name__) + def find_and_replace_owned_files(source_dir, target_dir, username): """ Finds files owned by a specific user in a source directory tree, @@ -29,10 +33,15 @@ def find_and_replace_owned_files(source_dir, target_dir, username): try: user_uid = pwd.getpwnam(username).pw_uid except KeyError: - print(f"Error: User '{username}' not found. Exiting.") + logger.error("Error: User '%s' not found. Exiting.", username) return - print(f"Searching for files owned by '{username}' (UID: {user_uid}) in '{source_dir}'...") + logger.info( + "Searching for files owned by '%s' (UID: %s) in '%s'...", + username, + user_uid, + source_dir + ) for dirpath, _, filenames in os.walk(source_dir): for filename in filenames: @@ -41,7 +50,7 @@ def find_and_replace_owned_files(source_dir, target_dir, username): # Use os.stat().st_uid to get the file's owner UID try: if os.path.islink(file_path): - print(f"Skipping symlink: {file_path}") + logger.info("Skipping symlink: %s", file_path) continue file_uid = os.stat(file_path).st_uid @@ -49,7 +58,7 @@ def find_and_replace_owned_files(source_dir, target_dir, username): continue # Skip if file was deleted during traversal if file_uid == user_uid: - print(f"Found owned file: {file_path}") + logger.info("Found owned file: %s", file_path) # Determine the relative path and the new link's destination relative_path = os.path.relpath(file_path, source_dir) @@ -57,7 +66,12 @@ def find_and_replace_owned_files(source_dir, target_dir, username): # Check if the target file actually exists if not os.path.exists(link_target): - print(f"Warning: Corresponding file not found in '{target_dir}' for '{file_path}'. Skipping.") + logger.warning( + "Warning: Corresponding file not found in '%s' " + "for '%s'. Skipping.", + target_dir, + file_path + ) continue # Get the link name @@ -66,9 +80,9 @@ def find_and_replace_owned_files(source_dir, target_dir, username): # Remove the original file try: os.rename(link_name, link_name+".tmp") - print(f"Deleted original file: {link_name}") + logger.info("Deleted original file: %s", link_name) except OSError as e: - print(f"Error deleting file {link_name}: {e}. Skipping.") + logger.error("Error deleting file %s: %s. Skipping.", link_name, e) continue # Create the symbolic link, handling necessary parent directories @@ -77,10 +91,10 @@ def find_and_replace_owned_files(source_dir, target_dir, username): os.makedirs(os.path.dirname(link_name), exist_ok=True) os.symlink(link_target, link_name) os.remove(link_name+".tmp") - print(f"Created symbolic link: {link_name} -> {link_target}") + logger.info("Created symbolic link: %s -> %s", link_name, link_target) except OSError as e: os.rename(link_name+".tmp", link_name) - print(f"Error creating symlink for {link_name}: {e}. Skipping.") + logger.error("Error creating symlink for %s: %s. Skipping.", link_name, e) def parse_arguments(): """ @@ -114,6 +128,14 @@ def parse_arguments(): return parser.parse_args() if __name__ == '__main__': + # Configure logging to display INFO and above to console (stdout) + import sys + logging.basicConfig( + level=logging.INFO, + format='%(message)s', + stream=sys.stdout + ) + # --- Configuration --- args = parse_arguments() my_username = os.environ['USER'] diff --git a/tests/test_relink.py b/tests/test_relink.py index 03fe3be..3a8f7d8 100644 --- a/tests/test_relink.py +++ b/tests/test_relink.py @@ -7,12 +7,29 @@ import tempfile import shutil import pwd +import logging from unittest.mock import patch + import pytest # Add parent directory to path to import relink module sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -import relink +import relink # noqa: E402 + + +@pytest.fixture(scope="function", autouse=True) +def configure_logging(): + """Configure logging to output to stdout for all tests.""" + # Configure logging before each test + logging.basicConfig( + level=logging.INFO, + format="%(message)s", + stream=sys.stdout, + force=True, # Force reconfiguration + ) + yield + # Clean up logging handlers after each test + logging.getLogger().handlers.clear() class TestFindAndReplaceOwnedFiles: @@ -86,7 +103,7 @@ def test_nested_directory_structure(self, temp_dirs, current_user): assert os.path.islink(source_file), "Nested file should be a symlink" assert os.readlink(source_file) == target_file - def test_skip_existing_symlinks(self, temp_dirs, current_user, capsys): + def test_skip_existing_symlinks(self, temp_dirs, current_user, caplog): """Test that existing symlinks are skipped.""" source_dir, target_dir = temp_dirs username = current_user @@ -107,7 +124,8 @@ def test_skip_existing_symlinks(self, temp_dirs, current_user, capsys): mtime_before = stat_before.st_mtime # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) # Verify the symlink is unchanged (same inode means it wasn't deleted/recreated) stat_after = os.lstat(source_link) @@ -119,12 +137,11 @@ def test_skip_existing_symlinks(self, temp_dirs, current_user, capsys): os.readlink(source_link) == dummy_target ), "Symlink target should be unchanged" - # Check that "Skipping symlink" message was printed - captured = capsys.readouterr() - assert "Skipping symlink:" in captured.out - assert source_link in captured.out + # Check that "Skipping symlink" message was logged + assert "Skipping symlink:" in caplog.text + assert source_link in caplog.text - def test_missing_target_file(self, temp_dirs, current_user, capsys): + def test_missing_target_file(self, temp_dirs, current_user, caplog): """Test behavior when target file doesn't exist.""" source_dir, target_dir = temp_dirs username = current_user @@ -135,17 +152,17 @@ def test_missing_target_file(self, temp_dirs, current_user, capsys): f.write("orphan content") # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) # Verify the file is NOT converted to symlink assert not os.path.islink(source_file), "File should not be a symlink" assert os.path.isfile(source_file), "Original file should still exist" # Check warning message - captured = capsys.readouterr() - assert "Warning: Corresponding file not found" in captured.out + assert "Warning: Corresponding file not found" in caplog.text - def test_invalid_username(self, temp_dirs, capsys): + def test_invalid_username(self, temp_dirs, caplog): """Test behavior with invalid username.""" source_dir, target_dir = temp_dirs @@ -159,12 +176,14 @@ def test_invalid_username(self, temp_dirs, capsys): raise RuntimeError(f"{invalid_username=} DOES actually exist") # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, invalid_username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files( + source_dir, target_dir, invalid_username + ) # Check error message - captured = capsys.readouterr() - assert "Error: User" in captured.out - assert "not found" in captured.out + assert "Error: User" in caplog.text + assert "not found" in caplog.text def test_multiple_files(self, temp_dirs, current_user): """Test with multiple files in the directory.""" @@ -220,20 +239,20 @@ def test_absolute_paths(self, temp_dirs, current_user): finally: os.chdir(cwd) - def test_print_searching_message(self, temp_dirs, current_user, capsys): + def test_print_searching_message(self, temp_dirs, current_user, caplog): """Test that searching message is printed.""" source_dir, target_dir = temp_dirs username = current_user # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) - # Check that searching message was printed - captured = capsys.readouterr() - assert f"Searching for files owned by '{username}'" in captured.out - assert f"in '{os.path.abspath(source_dir)}'" in captured.out + # Check that searching message was logged + assert f"Searching for files owned by '{username}'" in caplog.text + assert f"in '{os.path.abspath(source_dir)}'" in caplog.text - def test_print_found_owned_file(self, temp_dirs, current_user, capsys): + def test_print_found_owned_file(self, temp_dirs, current_user, caplog): """Test that 'Found owned file' message is printed.""" source_dir, target_dir = temp_dirs username = current_user @@ -248,14 +267,14 @@ def test_print_found_owned_file(self, temp_dirs, current_user, capsys): f.write("target content") # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) - # Check that "Found owned file" message was printed - captured = capsys.readouterr() - assert "Found owned file:" in captured.out - assert source_file in captured.out + # Check that "Found owned file" message was logged + assert "Found owned file:" in caplog.text + assert source_file in caplog.text - def test_print_deleted_and_created_messages(self, temp_dirs, current_user, capsys): + def test_print_deleted_and_created_messages(self, temp_dirs, current_user, caplog): """Test that deleted and created symlink messages are printed.""" source_dir, target_dir = temp_dirs username = current_user @@ -270,13 +289,13 @@ def test_print_deleted_and_created_messages(self, temp_dirs, current_user, capsy f.write("target") # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) # Check messages - captured = capsys.readouterr() - assert "Deleted original file:" in captured.out - assert "Created symbolic link:" in captured.out - assert f"{source_file} -> {target_file}" in captured.out + assert "Deleted original file:" in caplog.text + assert "Created symbolic link:" in caplog.text + assert f"{source_file} -> {target_file}" in caplog.text class TestParseArguments: @@ -387,7 +406,7 @@ def test_file_with_special_characters(self, temp_dirs): assert os.path.islink(source_file) assert os.readlink(source_file) == target_file - def test_error_deleting_file(self, temp_dirs, capsys): + def test_error_deleting_file(self, temp_dirs, caplog): """Test error message when file deletion fails.""" source_dir, target_dir = temp_dirs username = os.environ["USER"] @@ -407,14 +426,14 @@ def mock_rename(src, dst): with patch("os.rename", side_effect=mock_rename): # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) # Check error message - captured = capsys.readouterr() - assert "Error deleting file" in captured.out - assert source_file in captured.out + assert "Error deleting file" in caplog.text + assert source_file in caplog.text - def test_error_creating_symlink(self, temp_dirs, capsys): + def test_error_creating_symlink(self, temp_dirs, caplog): """Test error message when symlink creation fails.""" source_dir, target_dir = temp_dirs username = os.environ["USER"] @@ -434,9 +453,9 @@ def mock_symlink(src, dst): with patch("os.symlink", side_effect=mock_symlink): # Run the function - relink.find_and_replace_owned_files(source_dir, target_dir, username) + with caplog.at_level(logging.INFO): + relink.find_and_replace_owned_files(source_dir, target_dir, username) # Check error message - captured = capsys.readouterr() - assert "Error creating symlink" in captured.out - assert source_file in captured.out + assert "Error creating symlink" in caplog.text + assert source_file in caplog.text From cbb88dbf82987bb16c0f01194f71f354b4b4c399 Mon Sep 17 00:00:00 2001 From: Sam Rabin Date: Tue, 13 Jan 2026 16:50:01 -0700 Subject: [PATCH 3/5] Reduce number of vars in test_skip_existing_symlinks(). --- tests/test_relink.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_relink.py b/tests/test_relink.py index 3a8f7d8..74cfe9f 100644 --- a/tests/test_relink.py +++ b/tests/test_relink.py @@ -120,8 +120,6 @@ def test_skip_existing_symlinks(self, temp_dirs, current_user, caplog): # Get the inode and mtime before running the function stat_before = os.lstat(source_link) - inode_before = stat_before.st_ino - mtime_before = stat_before.st_mtime # Run the function with caplog.at_level(logging.INFO): @@ -130,9 +128,11 @@ def test_skip_existing_symlinks(self, temp_dirs, current_user, caplog): # Verify the symlink is unchanged (same inode means it wasn't deleted/recreated) stat_after = os.lstat(source_link) assert ( - inode_before == stat_after.st_ino + stat_before.st_ino == stat_after.st_ino ), "Symlink should not have been recreated" - assert mtime_before == stat_after.st_mtime, "Symlink mtime should be unchanged" + assert ( + stat_before.st_mtime == stat_after.st_mtime + ), "Symlink mtime should be unchanged" assert ( os.readlink(source_link) == dummy_target ), "Symlink target should be unchanged" From 9d8624d2ffa254bfba98c1eac730012202e75970 Mon Sep 17 00:00:00 2001 From: Sam Rabin Date: Tue, 13 Jan 2026 17:15:16 -0700 Subject: [PATCH 4/5] relink.py: Add verbosity level flags. --- relink.py | 32 +++++++-- tests/test_relink.py | 166 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 5 deletions(-) diff --git a/relink.py b/relink.py index a436838..95c54e2 100644 --- a/relink.py +++ b/relink.py @@ -101,8 +101,8 @@ def parse_arguments(): Parse command-line arguments. Returns: - argparse.Namespace: Parsed arguments containing source_root - and target_root. + argparse.Namespace: Parsed arguments containing source_root, + target_root, and verbosity settings. """ parser = argparse.ArgumentParser( description=( @@ -125,19 +125,41 @@ def parse_arguments(): ) ) + # Verbosity options (mutually exclusive) + verbosity_group = parser.add_mutually_exclusive_group() + verbosity_group.add_argument( + '-v', '--verbose', + action='store_true', + help='Enable verbose output' + ) + verbosity_group.add_argument( + '-q', '--quiet', + action='store_true', + help='Quiet mode (show only warnings and errors)' + ) + return parser.parse_args() if __name__ == '__main__': # Configure logging to display INFO and above to console (stdout) import sys + # --- Configuration --- + args = parse_arguments() + + # Configure logging based on verbosity flags + if args.quiet: + LOG_LEVEL = logging.WARNING + elif args.verbose: + LOG_LEVEL = logging.DEBUG + else: + LOG_LEVEL = logging.INFO + logging.basicConfig( - level=logging.INFO, + level=LOG_LEVEL, format='%(message)s', stream=sys.stdout ) - # --- Configuration --- - args = parse_arguments() my_username = os.environ['USER'] # --- Execution --- diff --git a/tests/test_relink.py b/tests/test_relink.py index 74cfe9f..6a78253 100644 --- a/tests/test_relink.py +++ b/tests/test_relink.py @@ -336,6 +336,172 @@ def test_both_custom_paths(self): assert args.source_root == source_path assert args.target_root == target_path + def test_verbose_flag(self): + """Test that --verbose flag is parsed correctly.""" + with patch("sys.argv", ["relink.py", "--verbose"]): + args = relink.parse_arguments() + assert args.verbose is True + assert args.quiet is False + + def test_quiet_flag(self): + """Test that --quiet flag is parsed correctly.""" + with patch("sys.argv", ["relink.py", "--quiet"]): + args = relink.parse_arguments() + assert args.quiet is True + assert args.verbose is False + + def test_verbose_short_flag(self): + """Test that -v flag is parsed correctly.""" + with patch("sys.argv", ["relink.py", "-v"]): + args = relink.parse_arguments() + assert args.verbose is True + + def test_quiet_short_flag(self): + """Test that -q flag is parsed correctly.""" + with patch("sys.argv", ["relink.py", "-q"]): + args = relink.parse_arguments() + assert args.quiet is True + + def test_default_verbosity(self): + """Test that default verbosity has both flags as False.""" + with patch("sys.argv", ["relink.py"]): + args = relink.parse_arguments() + assert args.verbose is False + assert args.quiet is False + + def test_verbose_and_quiet_mutually_exclusive(self): + """Test that --verbose and --quiet cannot be used together.""" + with patch("sys.argv", ["relink.py", "--verbose", "--quiet"]): + with pytest.raises(SystemExit) as exc_info: + relink.parse_arguments() + # Mutually exclusive arguments cause SystemExit with code 2 + assert exc_info.value.code == 2 + + def test_verbose_and_quiet_short_flags_mutually_exclusive(self): + """Test that -v and -q cannot be used together.""" + with patch("sys.argv", ["relink.py", "-v", "-q"]): + with pytest.raises(SystemExit) as exc_info: + relink.parse_arguments() + # Mutually exclusive arguments cause SystemExit with code 2 + assert exc_info.value.code == 2 + + +class TestVerbosityLevels: + """Test suite for verbosity level behavior.""" + + @pytest.fixture + def temp_dirs(self): + """Create temporary source and target directories for testing.""" + source_dir = tempfile.mkdtemp(prefix="test_source_") + target_dir = tempfile.mkdtemp(prefix="test_target_") + + yield source_dir, target_dir + + # Cleanup + shutil.rmtree(source_dir, ignore_errors=True) + shutil.rmtree(target_dir, ignore_errors=True) + + def test_quiet_mode_suppresses_info_messages(self, temp_dirs, caplog): + """Test that quiet mode suppresses INFO level messages.""" + source_dir, target_dir = temp_dirs + username = os.environ["USER"] + + # Create files + source_file = os.path.join(source_dir, "test_file.txt") + target_file = os.path.join(target_dir, "test_file.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("source") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target") + + # Create a symlink to test "Skipping symlink" message + source_link = os.path.join(source_dir, "existing_link.txt") + dummy_target = os.path.join(tempfile.gettempdir(), "somewhere") + os.symlink(dummy_target, source_link) + + # Run the function with WARNING level (quiet mode) + with caplog.at_level(logging.WARNING): + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Verify INFO messages are NOT in the log + assert "Searching for files owned by" not in caplog.text + assert "Skipping symlink:" not in caplog.text + assert "Found owned file:" not in caplog.text + assert "Deleted original file:" not in caplog.text + assert "Created symbolic link:" not in caplog.text + + def test_quiet_mode_shows_warnings(self, temp_dirs, caplog): + """Test that quiet mode still shows WARNING level messages.""" + source_dir, target_dir = temp_dirs + username = os.environ["USER"] + + # Create only source file (no corresponding target) to trigger warning + source_file = os.path.join(source_dir, "orphan.txt") + with open(source_file, "w", encoding="utf-8") as f: + f.write("orphan content") + + # Run the function with WARNING level (quiet mode) + with caplog.at_level(logging.WARNING): + relink.find_and_replace_owned_files(source_dir, target_dir, username) + + # Verify WARNING message IS in the log + assert "Warning: Corresponding file not found" in caplog.text + + def test_quiet_mode_shows_errors(self, temp_dirs, caplog): + """Test that quiet mode still shows ERROR level messages.""" + source_dir, target_dir = temp_dirs + username = os.environ["USER"] + + # Test 1: Invalid username error + invalid_username = "nonexistent_user_12345" + with caplog.at_level(logging.WARNING): + relink.find_and_replace_owned_files( + source_dir, target_dir, invalid_username + ) + assert "Error: User" in caplog.text + assert "not found" in caplog.text + + # Clear the log for next test + caplog.clear() + + # Test 2: Error deleting file + source_file = os.path.join(source_dir, "test.txt") + target_file = os.path.join(target_dir, "test.txt") + + with open(source_file, "w", encoding="utf-8") as f: + f.write("source") + with open(target_file, "w", encoding="utf-8") as f: + f.write("target") + + def mock_rename(src, dst): + raise OSError("Simulated rename error") + + with patch("os.rename", side_effect=mock_rename): + with caplog.at_level(logging.WARNING): + relink.find_and_replace_owned_files(source_dir, target_dir, username) + assert "Error deleting file" in caplog.text + + # Clear the log for next test + caplog.clear() + + # Test 3: Error creating symlink + source_file2 = os.path.join(source_dir, "test2.txt") + target_file2 = os.path.join(target_dir, "test2.txt") + + with open(source_file2, "w", encoding="utf-8") as f: + f.write("source2") + with open(target_file2, "w", encoding="utf-8") as f: + f.write("target2") + + def mock_symlink(src, dst): + raise OSError("Simulated symlink error") + + with patch("os.symlink", side_effect=mock_symlink): + with caplog.at_level(logging.WARNING): + relink.find_and_replace_owned_files(source_dir, target_dir, username) + assert "Error creating symlink" in caplog.text + class TestEdgeCases: """Test edge cases and error handling.""" From ddde6366529b80c86dfe47e2dc2195489b44a040 Mon Sep 17 00:00:00 2001 From: Sam Rabin Date: Tue, 13 Jan 2026 17:18:04 -0700 Subject: [PATCH 5/5] relink.py: Move import of sys. --- relink.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/relink.py b/relink.py index 95c54e2..1cc1e96 100644 --- a/relink.py +++ b/relink.py @@ -5,6 +5,7 @@ """ import os +import sys import pwd import argparse import logging @@ -141,9 +142,7 @@ def parse_arguments(): return parser.parse_args() if __name__ == '__main__': - # Configure logging to display INFO and above to console (stdout) - import sys - # --- Configuration --- + args = parse_arguments() # Configure logging based on verbosity flags