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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.5
parent
d6c8852e1e
commit
e3069a287e
+8
-3
@@ -13,9 +13,9 @@ from typing import Optional
|
|||||||
@dataclass
|
@dataclass
|
||||||
class VideoFile:
|
class VideoFile:
|
||||||
"""Represents a video file discovered during inventory scanning.
|
"""Represents a video file discovered during inventory scanning.
|
||||||
|
|
||||||
Attributes:
|
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)
|
filename: Name of the file (without directory path)
|
||||||
size_bytes: File size in bytes
|
size_bytes: File size in bytes
|
||||||
modified_timestamp: Last modification timestamp
|
modified_timestamp: Last modification timestamp
|
||||||
@@ -30,13 +30,18 @@ class VideoFile:
|
|||||||
size_bytes: int
|
size_bytes: int
|
||||||
modified_timestamp: datetime
|
modified_timestamp: datetime
|
||||||
category: str
|
category: str
|
||||||
|
|
||||||
# Optional metadata (if ffprobe available)
|
# Optional metadata (if ffprobe available)
|
||||||
resolution: Optional[str] = None
|
resolution: Optional[str] = None
|
||||||
codec: Optional[str] = None
|
codec: Optional[str] = None
|
||||||
duration_seconds: Optional[float] = None
|
duration_seconds: Optional[float] = None
|
||||||
bitrate_kbps: Optional[int] = 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
|
@dataclass
|
||||||
class MovieIdentity:
|
class MovieIdentity:
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
"""Shared utilities for Video Library Manager."""
|
"""Shared utilities for Video Library Manager."""
|
||||||
|
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
def utc_now() -> datetime:
|
def utc_now() -> datetime:
|
||||||
@@ -22,3 +23,27 @@ def format_size(size_bytes: int) -> str:
|
|||||||
return f"{size_bytes:.1f} {unit}"
|
return f"{size_bytes:.1f} {unit}"
|
||||||
size_bytes /= 1024.0
|
size_bytes /= 1024.0
|
||||||
return f"{size_bytes:.1f} PB"
|
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))
|
||||||
|
|||||||
@@ -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"
|
||||||
Reference in New Issue
Block a user