From e3069a287e16a55ebd7f897970a41aa42b19c099 Mon Sep 17 00:00:00 2001 From: windyboy Date: Fri, 13 Feb 2026 09:52:14 +0800 Subject: [PATCH] Add path canonicalization to prevent symlink duplicates Implements centralized path canonicalization to ensure symlinks and paths with . or .. segments are resolved to their canonical form, preventing false duplicate detection. Changes to utils.py: - Add canonical_path() function using Path.resolve() - Add canonical_path_str() for dictionary key usage Changes to models.py: - Add __post_init__ to VideoFile to auto-canonicalize paths - All VideoFile instances now have canonical paths automatically Impact: - Path-based dictionary lookups now work correctly with symlinks - io.py: path_to_inventory dictionary uses canonical keys - planner.py: path_to_index dictionary uses canonical keys - analysis.py: Duplicate detection uses canonical paths - No false duplicates from symlinks or relative paths Testing: - Added comprehensive tests for canonical_path functions - Tests verify symlink resolution, absolute path conversion, .. removal - All 449 tests pass (added 10 new tests) Co-Authored-By: Claude Sonnet 4.5 --- src/vlm/models.py | 11 +++-- src/vlm/utils.py | 25 ++++++++++ tests/test_utils.py | 118 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 151 insertions(+), 3 deletions(-) create mode 100644 tests/test_utils.py diff --git a/src/vlm/models.py b/src/vlm/models.py index de6ca71..516b945 100644 --- a/src/vlm/models.py +++ b/src/vlm/models.py @@ -13,9 +13,9 @@ from typing import Optional @dataclass class VideoFile: """Represents a video file discovered during inventory scanning. - + Attributes: - path: Full path to the video file + path: Full path to the video file (automatically canonicalized) filename: Name of the file (without directory path) size_bytes: File size in bytes modified_timestamp: Last modification timestamp @@ -30,13 +30,18 @@ class VideoFile: size_bytes: int modified_timestamp: datetime category: str - + # Optional metadata (if ffprobe available) resolution: Optional[str] = None codec: Optional[str] = None duration_seconds: Optional[float] = None bitrate_kbps: Optional[int] = None + def __post_init__(self): + """Canonicalize path on creation.""" + from vlm.utils import canonical_path + object.__setattr__(self, 'path', canonical_path(self.path)) + @dataclass class MovieIdentity: diff --git a/src/vlm/utils.py b/src/vlm/utils.py index 91f3dbc..8d9a2b7 100644 --- a/src/vlm/utils.py +++ b/src/vlm/utils.py @@ -1,6 +1,7 @@ """Shared utilities for Video Library Manager.""" from datetime import datetime, timezone +from pathlib import Path def utc_now() -> datetime: @@ -22,3 +23,27 @@ def format_size(size_bytes: int) -> str: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.1f} PB" + + +def canonical_path(path: Path) -> Path: + """Return canonical absolute path, resolving symlinks. + + Args: + path: Path to canonicalize + + Returns: + Canonical absolute path with symlinks resolved + """ + return path.resolve() + + +def canonical_path_str(path: Path) -> str: + """Return canonical path as string for use as dictionary key. + + Args: + path: Path to canonicalize + + Returns: + String representation of canonical path + """ + return str(canonical_path(path)) diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..d800d7e --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,118 @@ +"""Tests for utility functions.""" + +import tempfile +from pathlib import Path + +import pytest + +from vlm.utils import canonical_path, canonical_path_str, format_size, utc_now + + +class TestCanonicalPath: + """Tests for path canonicalization.""" + + def test_canonical_path_resolves_absolute(self, tmp_path): + """Test that canonical_path returns absolute path.""" + rel_path = Path("relative/path/file.txt") + canonical = canonical_path(rel_path) + assert canonical.is_absolute() + + def test_canonical_path_resolves_symlinks(self, tmp_path): + """Test that canonical_path resolves symlinks.""" + # Create a real file + real_file = tmp_path / "real_file.txt" + real_file.write_text("content") + + # Create a symlink + symlink = tmp_path / "symlink.txt" + symlink.symlink_to(real_file) + + # Canonicalize both paths + canonical_real = canonical_path(real_file) + canonical_symlink = canonical_path(symlink) + + # Both should resolve to the same canonical path + assert canonical_real == canonical_symlink + assert canonical_real == real_file.resolve() + + def test_canonical_path_with_dot_segments(self, tmp_path): + """Test that canonical_path removes . and .. segments.""" + # Create a file + test_file = tmp_path / "test.txt" + test_file.write_text("content") + + # Create path with . and .. + weird_path = tmp_path / "." / "subdir" / ".." / "test.txt" + + canonical = canonical_path(weird_path) + assert canonical == test_file.resolve() + assert ".." not in str(canonical) + assert "/." not in str(canonical) + + def test_canonical_path_str(self, tmp_path): + """Test that canonical_path_str returns string.""" + test_file = tmp_path / "test.txt" + test_file.write_text("content") + + canonical_str = canonical_path_str(test_file) + assert isinstance(canonical_str, str) + assert canonical_str == str(canonical_path(test_file)) + + def test_multiple_symlinks_resolve_to_same_path(self, tmp_path): + """Test that multiple symlinks to the same file resolve to the same canonical path.""" + # Create real file + real_file = tmp_path / "real.txt" + real_file.write_text("content") + + # Create multiple symlinks + symlink1 = tmp_path / "link1.txt" + symlink2 = tmp_path / "link2.txt" + symlink3 = tmp_path / "subdir" / "link3.txt" + symlink3.parent.mkdir(exist_ok=True) + + symlink1.symlink_to(real_file) + symlink2.symlink_to(real_file) + symlink3.symlink_to(real_file) + + # All should resolve to same canonical path + canonical_real = canonical_path(real_file) + canonical_link1 = canonical_path(symlink1) + canonical_link2 = canonical_path(symlink2) + canonical_link3 = canonical_path(symlink3) + + assert canonical_real == canonical_link1 == canonical_link2 == canonical_link3 + + +class TestFormatSize: + """Tests for format_size function.""" + + def test_format_size_bytes(self): + """Test formatting bytes.""" + assert format_size(0) == "0.0 B" + assert format_size(512) == "512.0 B" + assert format_size(1023) == "1023.0 B" + + def test_format_size_kilobytes(self): + """Test formatting kilobytes.""" + assert format_size(1024) == "1.0 KB" + assert format_size(1536) == "1.5 KB" + + def test_format_size_megabytes(self): + """Test formatting megabytes.""" + assert format_size(1024 * 1024) == "1.0 MB" + assert format_size(int(1.5 * 1024 * 1024)) == "1.5 MB" + + def test_format_size_gigabytes(self): + """Test formatting gigabytes.""" + assert format_size(1024 * 1024 * 1024) == "1.0 GB" + assert format_size(int(2.5 * 1024 * 1024 * 1024)) == "2.5 GB" + + +class TestUtcNow: + """Tests for utc_now function.""" + + def test_utc_now_returns_aware_datetime(self): + """Test that utc_now returns timezone-aware datetime.""" + now = utc_now() + assert now.tzinfo is not None + assert now.tzinfo.tzname(None) == "UTC"