2026-02-09 17:43:35 +08:00
|
|
|
"""Unit tests for the inventory scanner module."""
|
|
|
|
|
|
2026-02-09 23:55:35 +08:00
|
|
|
import csv
|
2026-06-01 15:32:03 +08:00
|
|
|
import json
|
|
|
|
|
import os
|
|
|
|
|
import subprocess
|
|
|
|
|
import threading
|
|
|
|
|
import time
|
2026-02-09 20:16:39 +08:00
|
|
|
from datetime import datetime, timezone
|
2026-02-09 17:43:35 +08:00
|
|
|
from pathlib import Path
|
2026-06-01 15:32:03 +08:00
|
|
|
from unittest.mock import MagicMock, patch
|
2026-02-09 17:43:35 +08:00
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
from vlm.config import Config
|
|
|
|
|
from vlm.models import VideoFile
|
|
|
|
|
from vlm.scanner import (
|
|
|
|
|
categorize_file,
|
|
|
|
|
extract_metadata,
|
2026-02-09 23:55:35 +08:00
|
|
|
load_inventory_csv,
|
2026-06-01 15:32:03 +08:00
|
|
|
save_inventory_csv,
|
|
|
|
|
scan_library,
|
2026-02-09 17:43:35 +08:00
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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"}
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
2026-04-07 11:00:47 +08:00
|
|
|
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)
|
|
|
|
|
|
2026-02-09 22:40:48 +08:00
|
|
|
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
|
2026-02-09 17:43:35 +08:00
|
|
|
|
|
|
|
|
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)
|
2026-02-09 20:16:39 +08:00
|
|
|
assert vf.modified_timestamp.tzinfo == timezone.utc
|
2026-02-09 17:43:35 +08:00
|
|
|
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
|
|
|
|
|
|
2026-02-09 22:40:48 +08:00
|
|
|
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)]
|
|
|
|
|
|
2026-02-13 13:36:39 +08:00
|
|
|
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
|
|
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
|
|
|
|
|
class TestCategorizeFile:
|
|
|
|
|
"""Tests for the categorize_file function."""
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "movie"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "series"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "anime"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "other"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "movie"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "movie"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
categories_config = {
|
|
|
|
|
"movie": ["movie"],
|
|
|
|
|
"series": ["series"],
|
|
|
|
|
"anime": ["anime"]
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
category = categorize_file(video_file, tmp_path, categories_config)
|
2026-02-09 17:43:35 +08:00
|
|
|
assert category == "other"
|
2026-02-09 22:40:48 +08:00
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
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()
|
2026-02-09 22:40:48 +08:00
|
|
|
|
|
|
|
|
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)
|
2026-02-09 17:43:35 +08:00
|
|
|
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
|
|
|
|
|
|
2026-02-09 23:55:35 +08:00
|
|
|
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()
|
|
|
|
|
|
2026-02-09 17:43:35 +08:00
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
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
|
|
|
|
|
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']
|
2026-02-09 20:16:39 +08:00
|
|
|
|
|
|
|
|
@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()
|
