"""Unit tests for the inventory scanner module.""" import os import threading import tempfile import time import csv from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch, MagicMock import subprocess import json import pytest from vlm.config import Config from vlm.models import VideoFile from vlm.scanner import ( categorize_file, scan_library, extract_metadata, save_inventory_csv, load_inventory_csv, ) class TestScanLibrary: """Tests for the scan_library function.""" def test_scan_empty_directory(self, tmp_path): """Test scanning an empty directory returns empty list.""" config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert result == [] def test_scan_nonexistent_directory(self, tmp_path): """Test scanning a nonexistent directory returns empty list.""" nonexistent = tmp_path / "nonexistent" config = Config(library_root=nonexistent) result = scan_library(nonexistent, config) assert result == [] def test_scan_discovers_video_files(self, tmp_path): """Test scanning discovers video files with correct extensions.""" # Create test structure movie_dir = tmp_path / "movie" movie_dir.mkdir() # Create video files video1 = movie_dir / "test1.mp4" video2 = movie_dir / "test2.mkv" video1.touch() video2.touch() # Create non-video file text_file = movie_dir / "readme.txt" text_file.touch() config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) # Should find only video files assert len(result) == 2 filenames = {vf.filename for vf in result} assert filenames == {"test1.mp4", "test2.mkv"} def test_scan_recursive(self, tmp_path): """Test scanning recursively discovers files in subdirectories.""" # Create nested structure movie_dir = tmp_path / "movie" subdir = movie_dir / "subdir" subdir.mkdir(parents=True) # Create files at different levels video1 = movie_dir / "movie1.mp4" video2 = subdir / "movie2.mkv" video1.touch() video2.touch() config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 2 filenames = {vf.filename for vf in result} assert filenames == {"movie1.mp4", "movie2.mkv"} def test_scan_filters_by_extension(self, tmp_path): """Test scanning filters files by configured extensions.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() # Create files with various extensions mp4_file = movie_dir / "video.mp4" mkv_file = movie_dir / "video.mkv" avi_file = movie_dir / "video.avi" txt_file = movie_dir / "readme.txt" mp4_file.touch() mkv_file.touch() avi_file.touch() txt_file.touch() # Configure to only accept .mp4 and .mkv config = Config( library_root=tmp_path, video_extensions=[".mp4", ".mkv"] ) result = scan_library(tmp_path, config) assert len(result) == 2 filenames = {vf.filename for vf in result} assert filenames == {"video.mp4", "video.mkv"} def test_scan_uses_find_output_and_filters_hidden_paths(self, tmp_path): """Test scan_library filters hidden paths from find output.""" movie_dir = tmp_path / "movie" hidden_dir = tmp_path / ".hidden" movie_dir.mkdir() hidden_dir.mkdir() visible_file = movie_dir / "visible.mp4" hidden_file = hidden_dir / "hidden.mp4" visible_file.touch() hidden_file.touch() fake_stdout = f"{visible_file}\0{hidden_file}\0".encode() with patch('subprocess.Popen') as mock_popen: process = MagicMock() process.communicate.return_value = (fake_stdout, b"") process.returncode = 0 mock_popen.return_value = process config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 1 assert result[0].path == visible_file def test_scan_keeps_partial_find_results_when_find_exits_nonzero(self, tmp_path): """Non-zero find exits should keep partial stdout and log the contract.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() visible_file = movie_dir / "visible.mp4" visible_file.touch() fake_stdout = f"{visible_file}\0".encode() with patch('subprocess.Popen') as mock_popen, patch("vlm.scanner.logger.warning") as mock_warning: process = MagicMock() process.communicate.return_value = (fake_stdout, b"Permission denied") process.returncode = 1 mock_popen.return_value = process config = Config(library_root=tmp_path) result = scan_library(tmp_path, config, include_video_metadata=False) warning_messages = [ call.args[0] % call.args[1:] if len(call.args) > 1 else call.args[0] for call in mock_warning.call_args_list ] assert len(result) == 1 assert result[0].path == visible_file assert any("using 1 partial scan result" in message for message in warning_messages) assert any("Permission denied" in message for message in warning_messages) def test_scan_returns_empty_when_find_exits_nonzero_without_stdout(self, tmp_path): """Non-zero find exits without stdout should produce an empty result deterministically.""" (tmp_path / "movie").mkdir() with patch('subprocess.Popen') as mock_popen, patch("vlm.scanner.logger.warning") as mock_warning: process = MagicMock() process.communicate.return_value = (b"", b"Permission denied") process.returncode = 1 mock_popen.return_value = process config = Config(library_root=tmp_path) result = scan_library(tmp_path, config, include_video_metadata=False) warning_messages = [ call.args[0] % call.args[1:] if len(call.args) > 1 else call.args[0] for call in mock_warning.call_args_list ] assert result == [] assert any("produced no scan results" in message for message in warning_messages) assert any("Permission denied" in message for message in warning_messages) def test_scan_falls_back_when_find_is_unavailable(self, tmp_path): """Test scan_library falls back to recursive scanning if find is unavailable.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "fallback.mp4" video_file.touch() with patch('subprocess.Popen', side_effect=FileNotFoundError): config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 1 assert result[0].path == video_file def test_scan_records_metadata(self, tmp_path): """Test scanning records file metadata correctly.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.write_text("test content") config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 1 vf = result[0] # Check metadata assert vf.filename == "test.mp4" assert vf.path == video_file assert vf.size_bytes > 0 assert isinstance(vf.modified_timestamp, datetime) assert vf.modified_timestamp.tzinfo == timezone.utc assert vf.category == "movie" def test_scan_categorizes_files(self, tmp_path): """Test scanning categorizes files based on directory structure.""" # Create category directories movie_dir = tmp_path / "movie" series_dir = tmp_path / "series" anime_dir = tmp_path / "anime" other_dir = tmp_path / "other" movie_dir.mkdir() series_dir.mkdir() anime_dir.mkdir() other_dir.mkdir() # Create files in each category (movie_dir / "movie.mp4").touch() (series_dir / "series.mkv").touch() (anime_dir / "anime.avi").touch() (other_dir / "other.mov").touch() config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 4 # Check categories categories = {vf.filename: vf.category for vf in result} assert categories["movie.mp4"] == "movie" assert categories["series.mkv"] == "series" assert categories["anime.avi"] == "anime" assert categories["other.mov"] == "other" def test_scan_skips_hidden_files(self, tmp_path): """Test scanning skips hidden files and directories.""" movie_dir = tmp_path / "movie" hidden_dir = tmp_path / ".hidden" movie_dir.mkdir() hidden_dir.mkdir() # Create visible and hidden files visible = movie_dir / "visible.mp4" hidden_file = movie_dir / ".hidden.mp4" hidden_dir_file = hidden_dir / "file.mp4" visible.touch() hidden_file.touch() hidden_dir_file.touch() config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) # Should only find visible file assert len(result) == 1 assert result[0].filename == "visible.mp4" def test_scan_handles_inaccessible_files(self, tmp_path): """Test scanning continues when encountering inaccessible files.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() # Create accessible files video1 = movie_dir / "video1.mp4" video2 = movie_dir / "video2.mp4" video1.touch() video2.touch() config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) # Should find both files (no permission errors in test environment) assert len(result) == 2 def test_scan_reports_progress_callback(self, tmp_path): """Test scanning reports progress updates for discovered files.""" config = Config(library_root=tmp_path) fake_paths = [tmp_path / "a.mp4", tmp_path / "b.mp4"] fake_video = VideoFile( path=fake_paths[0], filename="a.mp4", size_bytes=1, modified_timestamp=datetime.now(timezone.utc), category="movie", ) progress_events: list[tuple[int, int]] = [] with patch("vlm.scanner._discover_video_paths", return_value=fake_paths), patch( "vlm.scanner._create_video_file", side_effect=[fake_video, None] ): result = scan_library( tmp_path, config, progress_callback=lambda current, total: progress_events.append((current, total)) ) assert len(result) == 1 assert progress_events == [(0, 2), (1, 2), (2, 2)] def test_scan_reports_progress_for_empty_discovery(self, tmp_path): """Test scanning reports zero progress when no files are discovered.""" config = Config(library_root=tmp_path) progress_events: list[tuple[int, int]] = [] with patch("vlm.scanner._discover_video_paths", return_value=[]): result = scan_library( tmp_path, config, progress_callback=lambda current, total: progress_events.append((current, total)) ) assert result == [] assert progress_events == [(0, 0)] def test_scan_metadata_parallelism_uses_multiple_threads(self, tmp_path): """Metadata scan should use worker threads when concurrency > 1.""" config = Config(library_root=tmp_path, enrichment_max_concurrency=4) fake_paths = [tmp_path / f"f{i}.mp4" for i in range(8)] thread_ids: set[int] = set() lock = threading.Lock() def _fake_create(file_path, library_root, categories_config, include_video_metadata=True, metadata_cache=None): time.sleep(0.01) with lock: thread_ids.add(threading.get_ident()) return VideoFile( path=file_path, filename=file_path.name, size_bytes=1, modified_timestamp=datetime.now(timezone.utc), category="movie", ) with patch("vlm.scanner._discover_video_paths", return_value=fake_paths), patch( "vlm.scanner._create_video_file", side_effect=_fake_create, ): result = scan_library(tmp_path, config, include_video_metadata=True) assert len(result) == len(fake_paths) assert len(thread_ids) > 1 class TestCategorizeFile: """Tests for the categorize_file function.""" def test_categorize_movie(self, tmp_path): """Test categorizing a file in movie directory.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "movie" def test_categorize_series(self, tmp_path): """Test categorizing a file in series directory.""" series_dir = tmp_path / "series" series_dir.mkdir() video_file = series_dir / "test.mkv" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "series" def test_categorize_anime(self, tmp_path): """Test categorizing a file in anime directory.""" anime_dir = tmp_path / "anime" anime_dir.mkdir() video_file = anime_dir / "test.avi" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "anime" def test_categorize_other(self, tmp_path): """Test categorizing a file in other directory.""" other_dir = tmp_path / "other" other_dir.mkdir() video_file = other_dir / "test.mov" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "other" def test_categorize_nested_file(self, tmp_path): """Test categorizing a file in nested subdirectory.""" movie_dir = tmp_path / "movie" / "subdir" / "nested" movie_dir.mkdir(parents=True) video_file = movie_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "movie" def test_categorize_case_insensitive(self, tmp_path): """Test categorization is case-insensitive.""" movie_dir = tmp_path / "Movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "movie" def test_categorize_file_in_root(self, tmp_path): """Test categorizing a file directly in library root.""" video_file = tmp_path / "test.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "other" def test_categorize_unknown_directory(self, tmp_path): """Test categorizing a file in unknown directory.""" unknown_dir = tmp_path / "random" unknown_dir.mkdir() video_file = unknown_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "other" def test_categorize_plural_movies(self, tmp_path): """Test recognizing plural 'movies' directory.""" movies_dir = tmp_path / "movies" movies_dir.mkdir() video_file = movies_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie", "movies"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "movie" def test_categorize_tv_directory(self, tmp_path): """Test recognizing 'tv' as series category.""" tv_dir = tmp_path / "tv" tv_dir.mkdir() video_file = tv_dir / "show.mkv" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series", "tv", "shows"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "series" def test_categorize_custom_case_insensitive(self, tmp_path): """Test case-insensitive matching with custom mappings.""" movies_dir = tmp_path / "MOVIES" movies_dir.mkdir() video_file = movies_dir / "test.mp4" video_file.touch() categories_config = { "movie": ["movie", "movies"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "movie" def test_categorize_unmapped_returns_other(self, tmp_path): """Test unmapped directory returns 'other'.""" downloads_dir = tmp_path / "downloads" downloads_dir.mkdir() video_file = downloads_dir / "file.mp4" video_file.touch() categories_config = { "movie": ["movie"], "series": ["series"], "anime": ["anime"] } category = categorize_file(video_file, tmp_path, categories_config) assert category == "other" class TestExtractMetadata: """Tests for the extract_metadata function.""" def test_extract_metadata_with_ffprobe_available(self, tmp_path): """Test metadata extraction when ffprobe is available and returns valid data.""" video_file = tmp_path / "test.mp4" video_file.touch() # Mock ffprobe output mock_output = { "streams": [ { "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080 } ], "format": { "duration": "120.5", "bit_rate": "5000000" } } with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout=json.dumps(mock_output), stderr="" ) result = extract_metadata(video_file) assert result['resolution'] == "1920x1080" assert result['codec'] == "h264" assert result['duration_seconds'] == 120.5 assert result['bitrate_kbps'] == 5000 def test_extract_metadata_ffprobe_not_available(self, tmp_path): """Test metadata extraction when ffprobe is not installed.""" video_file = tmp_path / "test.mp4" video_file.touch() with patch('subprocess.run', side_effect=FileNotFoundError): result = extract_metadata(video_file) assert result == {} def test_extract_metadata_ffprobe_fails(self, tmp_path): """Test metadata extraction when ffprobe fails.""" video_file = tmp_path / "test.mp4" video_file.touch() with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=1, stdout="", stderr="Error processing file" ) result = extract_metadata(video_file) assert result == {} def test_extract_metadata_ffprobe_timeout(self, tmp_path): """Test metadata extraction when ffprobe times out.""" video_file = tmp_path / "test.mp4" video_file.touch() with patch('subprocess.run', side_effect=subprocess.TimeoutExpired('ffprobe', 10)): result = extract_metadata(video_file) assert result == {} def test_extract_metadata_invalid_json(self, tmp_path): """Test metadata extraction when ffprobe returns invalid JSON.""" video_file = tmp_path / "test.mp4" video_file.touch() with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout="invalid json", stderr="" ) result = extract_metadata(video_file) assert result == {} def test_extract_metadata_partial_data(self, tmp_path): """Test metadata extraction with partial data available.""" video_file = tmp_path / "test.mp4" video_file.touch() # Mock ffprobe output with only some fields mock_output = { "streams": [ { "codec_type": "video", "codec_name": "h264" # Missing width and height } ], "format": { "duration": "120.5" # Missing bit_rate } } with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout=json.dumps(mock_output), stderr="" ) result = extract_metadata(video_file) assert result['codec'] == "h264" assert result['duration_seconds'] == 120.5 assert 'resolution' not in result assert 'bitrate_kbps' not in result def test_extract_metadata_no_video_stream(self, tmp_path): """Test metadata extraction when no video stream is found.""" video_file = tmp_path / "test.mp4" video_file.touch() # Mock ffprobe output with only audio stream mock_output = { "streams": [ { "codec_type": "audio", "codec_name": "aac" } ], "format": { "duration": "120.5", "bit_rate": "5000000" } } with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout=json.dumps(mock_output), stderr="" ) result = extract_metadata(video_file) # Should still extract format-level metadata assert result['duration_seconds'] == 120.5 assert result['bitrate_kbps'] == 5000 assert 'resolution' not in result assert 'codec' not in result def test_scan_library_with_metadata_extraction(self, tmp_path): """Test that scan_library integrates metadata extraction.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.touch() # Mock ffprobe output mock_output = { "streams": [ { "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080 } ], "format": { "duration": "120.5", "bit_rate": "5000000" } } with patch('subprocess.run') as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout=json.dumps(mock_output), stderr="" ) config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 1 vf = result[0] # Check that metadata was extracted assert vf.resolution == "1920x1080" assert vf.codec == "h264" assert vf.duration_seconds == 120.5 assert vf.bitrate_kbps == 5000 def test_scan_library_without_ffprobe(self, tmp_path): """Test that scan_library works gracefully without ffprobe.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.touch() with patch('subprocess.run', side_effect=FileNotFoundError): config = Config(library_root=tmp_path) result = scan_library(tmp_path, config) assert len(result) == 1 vf = result[0] # Check that file was still scanned without metadata assert vf.filename == "test.mp4" assert vf.resolution is None assert vf.codec is None assert vf.duration_seconds is None assert vf.bitrate_kbps is None def test_scan_library_with_metadata_disabled(self, tmp_path): """Test that metadata extraction can be disabled for faster scans.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.touch() config = Config(library_root=tmp_path) with patch('subprocess.run') as mock_run: result = scan_library(tmp_path, config, include_video_metadata=False) assert len(result) == 1 assert result[0].filename == "test.mp4" assert result[0].resolution is None assert result[0].codec is None assert result[0].duration_seconds is None assert result[0].bitrate_kbps is None mock_run.assert_not_called() def test_scan_library_reuses_cached_metadata_when_unchanged(self, tmp_path): """Test unchanged files reuse metadata from previous inventory cache.""" movie_dir = tmp_path / "movie" movie_dir.mkdir() video_file = movie_dir / "test.mp4" video_file.write_text("test") original = VideoFile( path=video_file, filename=video_file.name, size_bytes=video_file.stat().st_size, modified_timestamp=datetime.fromtimestamp(video_file.stat().st_mtime, tz=timezone.utc), category="movie", resolution="1920x1080", codec="h264", duration_seconds=120.5, bitrate_kbps=5000, ) cache_file = tmp_path / "cached_inventory.csv" save_inventory_csv([original], cache_file, tmp_path) cached_entries = load_inventory_csv(cache_file) metadata_cache = {str(vf.path): vf for vf in cached_entries} config = Config(library_root=tmp_path) with patch("subprocess.run") as mock_run: result = scan_library(tmp_path, config, metadata_cache=metadata_cache) assert len(result) == 1 scanned = result[0] assert scanned.resolution == "1920x1080" assert scanned.codec == "h264" assert scanned.duration_seconds == 120.5 assert scanned.bitrate_kbps == 5000 mock_run.assert_not_called() class TestInventoryReports: """Tests for inventory report generation functions.""" def test_save_inventory_csv_basic(self, tmp_path): """Test saving inventory to CSV format with basic data.""" # Create test video files video_files = [ VideoFile( path=Path("/library/movie/test1.mp4"), filename="test1.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie", resolution="1920x1080", codec="h264", duration_seconds=120.5, bitrate_kbps=5000 ), VideoFile( path=Path("/library/series/test2.mkv"), filename="test2.mkv", size_bytes=2048000, modified_timestamp=datetime(2024, 1, 16, 14, 45, 0), category="series", resolution="1280x720", codec="h265", duration_seconds=45.0, bitrate_kbps=3000 ) ] output_file = tmp_path / "inventory.csv" library_root = Path("/library") from vlm.scanner import save_inventory_csv save_inventory_csv(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content with open(output_file, 'r', encoding='utf-8') as f: content = f.read() # Check metadata comments assert "# Generated:" in content assert "# Library Root: /library" in content # Check header assert "path,filename,size_bytes,modified_timestamp,category,resolution,codec,duration_seconds,bitrate_kbps" in content # Check data rows assert "test1.mp4" in content assert "1024000" in content assert "movie" in content assert "1920x1080" in content assert "h264" in content assert "120.5" in content assert "5000" in content assert "test2.mkv" in content assert "2048000" in content assert "series" in content assert "1280x720" in content assert "h265" in content assert "45.0" in content assert "3000" in content def test_save_inventory_csv_with_missing_metadata(self, tmp_path): """Test saving inventory to CSV with missing optional metadata.""" # Create video file without optional metadata video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie", resolution=None, codec=None, duration_seconds=None, bitrate_kbps=None ) ] output_file = tmp_path / "inventory.csv" library_root = Path("/library") from vlm.scanner import save_inventory_csv save_inventory_csv(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content import csv with open(output_file, 'r', encoding='utf-8') as f: # Skip comment lines lines = [line for line in f if not line.startswith('#')] reader = csv.DictReader(lines) rows = list(reader) assert len(rows) == 1 row = rows[0] # Check required fields assert row['filename'] == 'test.mp4' assert row['size_bytes'] == '1024000' assert row['category'] == 'movie' # Check optional fields are empty strings assert row['resolution'] == '' assert row['codec'] == '' assert row['duration_seconds'] == '' assert row['bitrate_kbps'] == '' def test_save_inventory_csv_empty_list(self, tmp_path): """Test saving empty inventory to CSV.""" video_files = [] output_file = tmp_path / "inventory.csv" library_root = Path("/library") from vlm.scanner import save_inventory_csv save_inventory_csv(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content with open(output_file, 'r', encoding='utf-8') as f: content = f.read() # Should have metadata and header but no data rows assert "# Generated:" in content assert "# Library Root:" in content assert "path,filename,size_bytes" in content def test_save_inventory_csv_creates_directory(self, tmp_path): """Test that save_inventory_csv creates output directory if needed.""" video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie" ) ] # Use nested directory that doesn't exist output_file = tmp_path / "reports" / "inventory.csv" library_root = Path("/library") from vlm.scanner import save_inventory_csv save_inventory_csv(video_files, output_file, library_root) # Verify file was created assert output_file.exists() def test_save_inventory_json_basic(self, tmp_path): """Test saving inventory to JSON format with basic data.""" # Create test video files video_files = [ VideoFile( path=Path("/library/movie/test1.mp4"), filename="test1.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie", resolution="1920x1080", codec="h264", duration_seconds=120.5, bitrate_kbps=5000 ), VideoFile( path=Path("/library/series/test2.mkv"), filename="test2.mkv", size_bytes=2048000, modified_timestamp=datetime(2024, 1, 16, 14, 45, 0), category="series", resolution="1280x720", codec="h265", duration_seconds=45.0, bitrate_kbps=3000 ) ] output_file = tmp_path / "inventory.json" library_root = Path("/library") from vlm.scanner import save_inventory_json save_inventory_json(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content with open(output_file, 'r', encoding='utf-8') as f: data = json.load(f) # Check metadata assert 'metadata' in data assert 'generated' in data['metadata'] assert data['metadata']['library_root'] == '/library' assert data['metadata']['file_count'] == 2 # Check files assert 'files' in data assert len(data['files']) == 2 # Check first file file1 = data['files'][0] assert file1['filename'] == 'test1.mp4' assert file1['size_bytes'] == 1024000 assert file1['category'] == 'movie' assert file1['resolution'] == '1920x1080' assert file1['codec'] == 'h264' assert file1['duration_seconds'] == 120.5 assert file1['bitrate_kbps'] == 5000 # Check second file file2 = data['files'][1] assert file2['filename'] == 'test2.mkv' assert file2['size_bytes'] == 2048000 assert file2['category'] == 'series' assert file2['resolution'] == '1280x720' assert file2['codec'] == 'h265' assert file2['duration_seconds'] == 45.0 assert file2['bitrate_kbps'] == 3000 def test_save_inventory_json_with_missing_metadata(self, tmp_path): """Test saving inventory to JSON with missing optional metadata.""" # Create video file without optional metadata video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie", resolution=None, codec=None, duration_seconds=None, bitrate_kbps=None ) ] output_file = tmp_path / "inventory.json" library_root = Path("/library") from vlm.scanner import save_inventory_json save_inventory_json(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content with open(output_file, 'r', encoding='utf-8') as f: data = json.load(f) assert len(data['files']) == 1 file_data = data['files'][0] # Check required fields assert file_data['filename'] == 'test.mp4' assert file_data['size_bytes'] == 1024000 assert file_data['category'] == 'movie' # Check optional fields are null assert file_data['resolution'] is None assert file_data['codec'] is None assert file_data['duration_seconds'] is None assert file_data['bitrate_kbps'] is None def test_save_inventory_json_empty_list(self, tmp_path): """Test saving empty inventory to JSON.""" video_files = [] output_file = tmp_path / "inventory.json" library_root = Path("/library") from vlm.scanner import save_inventory_json save_inventory_json(video_files, output_file, library_root) # Verify file was created assert output_file.exists() # Read and verify content with open(output_file, 'r', encoding='utf-8') as f: data = json.load(f) # Should have metadata but no files assert data['metadata']['file_count'] == 0 assert len(data['files']) == 0 def test_save_inventory_json_creates_directory(self, tmp_path): """Test that save_inventory_json creates output directory if needed.""" video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie" ) ] # Use nested directory that doesn't exist output_file = tmp_path / "reports" / "inventory.json" library_root = Path("/library") from vlm.scanner import save_inventory_json save_inventory_json(video_files, output_file, library_root) # Verify file was created assert output_file.exists() def test_csv_and_json_consistency(self, tmp_path): """Test that CSV and JSON exports contain the same data.""" # Create test video files video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=datetime(2024, 1, 15, 10, 30, 0), category="movie", resolution="1920x1080", codec="h264", duration_seconds=120.5, bitrate_kbps=5000 ) ] csv_file = tmp_path / "inventory.csv" json_file = tmp_path / "inventory.json" library_root = Path("/library") from vlm.scanner import save_inventory_csv, save_inventory_json save_inventory_csv(video_files, csv_file, library_root) save_inventory_json(video_files, json_file, library_root) # Read CSV data import csv with open(csv_file, 'r', encoding='utf-8') as f: lines = [line for line in f if not line.startswith('#')] reader = csv.DictReader(lines) csv_rows = list(reader) # Read JSON data with open(json_file, 'r', encoding='utf-8') as f: json_data = json.load(f) # Compare data assert len(csv_rows) == len(json_data['files']) csv_row = csv_rows[0] json_file_data = json_data['files'][0] # Compare key fields assert csv_row['filename'] == json_file_data['filename'] assert csv_row['size_bytes'] == str(json_file_data['size_bytes']) assert csv_row['category'] == json_file_data['category'] assert csv_row['resolution'] == json_file_data['resolution'] assert csv_row['codec'] == json_file_data['codec'] @pytest.mark.skipif(not hasattr(time, "tzset"), reason="tzset not available on this platform") def test_save_inventory_csv_converts_naive_local_time_to_utc(self, tmp_path, monkeypatch): """Test naive timestamps are interpreted as local time and converted to UTC.""" original_tz = os.environ.get("TZ") try: monkeypatch.setenv("TZ", "Etc/GMT-2") time.tzset() naive_local = datetime(2024, 1, 15, 10, 30, 0) expected_utc = naive_local.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") video_files = [ VideoFile( path=Path("/library/movie/test.mp4"), filename="test.mp4", size_bytes=1024000, modified_timestamp=naive_local, category="movie" ) ] output_file = tmp_path / "inventory.csv" from vlm.scanner import save_inventory_csv save_inventory_csv(video_files, output_file, Path("/library")) content = output_file.read_text(encoding="utf-8") assert expected_utc in content finally: if original_tz is None: monkeypatch.delenv("TZ", raising=False) else: monkeypatch.setenv("TZ", original_tz) time.tzset()