relink.py: Require that source_dir(s) be in inputdata_root and target_dir not be.

Author: samsrabin

relink.py

@@ -180,12 +180,12 @@ def replace_files_with_symlinks(
 
     # Use efficient scandir-based search
     for file_path in find_owned_files_scandir(source_dir, user_uid, inputdata_root):
-        replace_one_file_with_symlink(source_dir, target_dir, file_path, dry_run=dry_run)
+        replace_one_file_with_symlink(
+            source_dir, target_dir, file_path, dry_run=dry_run
+        )
 
 
-def replace_one_file_with_symlink(
-    source_dir, target_dir, file_path, dry_run=False
-):
+def replace_one_file_with_symlink(source_dir, target_dir, file_path, dry_run=False):
     """
     Given a file, replaces it with a symbolic link to the same relative path in a target directory
     tree.
@@ -356,6 +356,26 @@ def process_args(args):
     else:
         args.log_level = logging.INFO
 
+    # Ensure that source_root is a list
+    if hasattr(args, "source_root") and not isinstance(args.source_root, list):
+        args.source_root = [args.source_root]
+
+    # Check that every item in source_root is a child of inputdata_root
+    if hasattr(args, "source_root"):  # Sometimes doesn't if we're testing
+        for item in args.source_root:
+            if not Path(item).is_relative_to(args.inputdata_root):
+                raise argparse.ArgumentTypeError(
+                    f"Item '{item}' not under inputdata root '{args.inputdata_root}'"
+                )
+
+    # Check that target_root is NOT a child of inputdata_root
+    if hasattr(args, "target_root"):  # Sometimes doesn't if we're testing
+        if Path(args.target_root).is_relative_to(args.inputdata_root):
+            raise argparse.ArgumentTypeError(
+                f"Target root ('{args.target_root}') must not be under inputdata root "
+                f"'{args.inputdata_root}'"
+            )
+
 
 def main():
 

tests/relink/conftest.py

@@ -17,7 +17,8 @@ def fixture_temp_dirs():
     target_dir = tempfile.mkdtemp(prefix="test_target_")
 
     with patch("relink.DEFAULT_SOURCE_ROOT", source_dir):
-        yield source_dir, target_dir
+        with patch("relink.DEFAULT_TARGET_ROOT", target_dir):
+            yield source_dir, target_dir
 
     # Cleanup
     shutil.rmtree(source_dir, ignore_errors=True)

tests/relink/test_args.py

@@ -4,8 +4,7 @@
 
 import os
 import sys
-import tempfile
-import shutil
+from pathlib import Path
 import logging
 import argparse
 from unittest.mock import patch
@@ -20,112 +19,91 @@
 import relink  # noqa: E402
 
 
-@pytest.fixture(scope="function", name="mock_default_dirs")
-def fixture_mock_default_dirs(temp_dirs):
-    """Mock the default directories to use temporary directories."""
-    source_dir = tempfile.mkdtemp(prefix="test_default_source_")
-    target_dir = tempfile.mkdtemp(prefix="test_default_target_")
-    source_dir, target_dir = temp_dirs
-
-    with patch.object(relink, "DEFAULT_SOURCE_ROOT", source_dir):
-        with patch.object(relink, "DEFAULT_TARGET_ROOT", target_dir):
-            yield source_dir, target_dir
-
-    # Cleanup
-    shutil.rmtree(source_dir, ignore_errors=True)
-    shutil.rmtree(target_dir, ignore_errors=True)
-
-
 class TestParseArguments:
     """Test suite for parse_arguments function."""
 
-    def test_default_arguments(self, mock_default_dirs):
+    def test_default_arguments(self, temp_dirs):
         """Test that default arguments are used when none provided."""
-        source_dir, target_dir = mock_default_dirs
-        with patch("sys.argv", ["relink.py"]):
+        source_dir, target_dir = temp_dirs
+        with patch("sys.argv", ["relink.py", source_dir]):
             args = relink.parse_arguments()
-            assert args.source_root == source_dir
+            assert args.source_root == [source_dir]
             assert args.target_root == target_dir
 
-    def test_custom_source_root(self, mock_default_dirs, tmp_path):
+    def test_custom_source_root(self, temp_dirs):
         """Test custom source root argument."""
-        _, target_dir = mock_default_dirs
-        custom_source = tmp_path / "custom_source"
+        source_dir, target_dir = temp_dirs
+        custom_source = Path(os.path.join(source_dir, "custom_source"))
         custom_source.mkdir()
         with patch("sys.argv", ["relink.py", str(custom_source)]):
             args = relink.parse_arguments()
             assert args.source_root == [str(custom_source.resolve())]
             assert args.target_root == target_dir
 
-    def test_custom_target_root(self, mock_default_dirs, tmp_path):
+    def test_custom_target_root(self, temp_dirs):
         """Test custom target root argument."""
-        source_dir, _ = mock_default_dirs
-        custom_target = tmp_path / "custom_target"
+        source_dir, target_dir = temp_dirs
+        custom_target = Path(os.path.join(target_dir, "custom_target"))
         custom_target.mkdir()
         with patch("sys.argv", ["relink.py", "--target-root", str(custom_target)]):
             args = relink.parse_arguments()
-            assert args.source_root == source_dir
+            assert args.source_root == [source_dir]
             assert args.target_root == str(custom_target.resolve())
 
-    def test_both_custom_paths(self, tmp_path):
+    def test_both_custom_paths(self, temp_dirs):
         """Test both custom source and target roots."""
-        source_path = tmp_path / "custom_source"
-        target_path = tmp_path / "custom_target"
-        source_path.mkdir()
-        target_path.mkdir()
+        source_root, target_root = temp_dirs
+        source_dir = Path(os.path.join(source_root, "custom_source"))
+        target_dir = Path(os.path.join(target_root, "custom_target"))
+        source_dir.mkdir()
+        target_dir.mkdir()
         with patch(
             "sys.argv",
             [
                 "relink.py",
-                str(source_path),
+                str(source_dir),
                 "--target-root",
-                str(target_path),
+                str(target_dir),
             ],
         ):
             args = relink.parse_arguments()
-            assert args.source_root == [str(source_path.resolve())]
-            assert args.target_root == str(target_path.resolve())
+            assert args.source_root == [str(source_dir.resolve())]
+            assert args.target_root == str(target_dir.resolve())
 
-    def test_verbose_flag(self, mock_default_dirs):  # pylint: disable=unused-argument
+    def test_verbose_flag(self, temp_dirs):  # pylint: disable=unused-argument
         """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, mock_default_dirs):  # pylint: disable=unused-argument
+    def test_quiet_flag(self, temp_dirs):  # pylint: disable=unused-argument
         """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, mock_default_dirs
-    ):  # pylint: disable=unused-argument
+    def test_verbose_short_flag(self, temp_dirs):  # pylint: disable=unused-argument
         """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, mock_default_dirs
-    ):  # pylint: disable=unused-argument
+    def test_quiet_short_flag(self, temp_dirs):  # pylint: disable=unused-argument
         """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, mock_default_dirs
-    ):  # pylint: disable=unused-argument
+    def test_default_verbosity(self, temp_dirs):  # pylint: disable=unused-argument
         """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, mock_default_dirs):
+    def test_verbose_and_quiet_mutually_exclusive(self, temp_dirs):
         """Test that --verbose and --quiet cannot be used together."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py", "--verbose", "--quiet"]):
@@ -134,7 +112,7 @@ def test_verbose_and_quiet_mutually_exclusive(self, mock_default_dirs):
             # Mutually exclusive arguments cause SystemExit with code 2
             assert exc_info.value.code == 2
 
-    def test_verbose_and_quiet_short_flags_mutually_exclusive(self, mock_default_dirs):
+    def test_verbose_and_quiet_short_flags_mutually_exclusive(self, temp_dirs):
         """Test that -v and -q cannot be used together."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py", "-v", "-q"]):
@@ -143,40 +121,40 @@ def test_verbose_and_quiet_short_flags_mutually_exclusive(self, mock_default_dir
             # Mutually exclusive arguments cause SystemExit with code 2
             assert exc_info.value.code == 2
 
-    def test_dry_run_flag(self, mock_default_dirs):
+    def test_dry_run_flag(self, temp_dirs):
         """Test that --dry-run flag is parsed correctly."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py", "--dry-run"]):
             args = relink.parse_arguments()
             assert args.dry_run is True
 
-    def test_dry_run_default(self, mock_default_dirs):
+    def test_dry_run_default(self, temp_dirs):
         """Test that dry_run defaults to False."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py"]):
             args = relink.parse_arguments()
             assert args.dry_run is False
 
-    def test_timing_flag(self, mock_default_dirs):
+    def test_timing_flag(self, temp_dirs):
         """Test that --timing flag is parsed correctly."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py", "--timing"]):
             args = relink.parse_arguments()
             assert args.timing is True
 
-    def test_timing_default(self, mock_default_dirs):
+    def test_timing_default(self, temp_dirs):
         """Test that timing defaults to False."""
         # pylint: disable=unused-argument
         with patch("sys.argv", ["relink.py"]):
             args = relink.parse_arguments()
             assert args.timing is False
 
-    def test_multiple_source_roots(self, mock_default_dirs, tmp_path):
+    def test_multiple_source_roots(self, temp_dirs):
         """Test that multiple source root arguments are parsed correctly."""
-        _, target_dir = mock_default_dirs
-        source1 = tmp_path / "source1"
-        source2 = tmp_path / "source2"
-        source3 = tmp_path / "source3"
+        inputdata_root, target_dir = temp_dirs
+        source1 = Path(os.path.join(inputdata_root, "source1"))
+        source2 = Path(os.path.join(inputdata_root, "source2"))
+        source3 = Path(os.path.join(inputdata_root, "source3"))
         source1.mkdir()
         source2.mkdir()
         source3.mkdir()
@@ -189,11 +167,12 @@ def test_multiple_source_roots(self, mock_default_dirs, tmp_path):
             assert str(source3.resolve()) in args.source_root
             assert args.target_root == target_dir
 
-    def test_multiple_source_roots_with_target(self, tmp_path):
+    def test_multiple_source_roots_with_target(self, temp_dirs):
         """Test multiple source roots with custom target root."""
-        source1 = tmp_path / "source1"
-        source2 = tmp_path / "source2"
-        target = tmp_path / "target"
+        inputdata_root, target_dir = temp_dirs
+        source1 = Path(os.path.join(inputdata_root, "source1"))
+        source2 = Path(os.path.join(inputdata_root, "source2"))
+        target = Path(os.path.join(target_dir, "target"))
         source1.mkdir()
         source2.mkdir()
         target.mkdir()
@@ -330,3 +309,26 @@ def test_process_args_modifies_args_in_place(self):
         # Should be the same object, modified in place
         assert args is original_args
         assert hasattr(args, "log_level")
+
+    def test_error_if_source_not_in_inputdata(self):
+        """Test that process_args errors if source isn't in inputdata_root."""
+        args = argparse.Namespace(
+            quiet=False, verbose=False, source_root="abc123", inputdata_root="def456"
+        )
+        with pytest.raises(argparse.ArgumentTypeError) as exc_info:
+            relink.process_args(args)
+        assert "not under inputdata root" in str(exc_info.value)
+
+    def test_error_if_target_in_inputdata(self):
+        """Test that process_args errors if target is in inputdata_root."""
+        inputdata_root = "inputdata"
+        target_root = os.path.join(inputdata_root, "abc123")
+        args = argparse.Namespace(
+            quiet=False,
+            verbose=False,
+            target_root=target_root,
+            inputdata_root=inputdata_root,
+        )
+        with pytest.raises(argparse.ArgumentTypeError) as exc_info:
+            relink.process_args(args)
+        assert "must not be under inputdata root" in str(exc_info.value)

tests/relink/test_cmdline.py

@@ -5,6 +5,7 @@
 import os
 import sys
 import subprocess
+from pathlib import Path
 
 import pytest
 
@@ -98,15 +99,15 @@ def test_command_line_execution_actual_run(mock_dirs):
     assert "Created symbolic link:" in result.stdout
 
 
-def test_command_line_multiple_source_dirs(tmp_path):
+def test_command_line_multiple_source_dirs(temp_dirs):
     """Test executing relink.py with multiple source directories."""
+    inputdata_dir, target_dir = temp_dirs
     # Create multiple source directories
-    source1 = tmp_path / "source1"
-    source2 = tmp_path / "source2"
-    target_dir = tmp_path / "target"
+    source1 = Path(os.path.join(inputdata_dir, "source1"))
+    source2 = Path(os.path.join(inputdata_dir, "source2"))
     source1.mkdir()
     source2.mkdir()
-    target_dir.mkdir()
+    target_dir = Path(target_dir)
 
     # Create files in each source directory
     source1_file = source1 / "file1.txt"
@@ -134,7 +135,7 @@ def test_command_line_multiple_source_dirs(tmp_path):
         "--target-root",
         str(target_dir),
         "--inputdata-root",
-        str(tmp_path),
+        str(inputdata_dir),
     ]
 
     # Execute the command