"""Tests for the I/O layer module.""" import json from datetime import datetime, timezone from pathlib import Path import pytest from vlm.io import ( _video_file_from_record, execution_plan_from_record, execution_plan_to_record, load_analysis_json, load_execution_plan, load_identities_json, save_execution_plan, save_identities_json, ) from vlm.models import ExecutionPlan, FileOperation, VideoFile class TestVideoFileFromRecord: """Tests for _video_file_from_record function.""" def test_video_file_from_record_with_metadata_v2(self): """Test loading VideoFile from v2 record with embedded metadata.""" record = { "path": "/library/movie/The Matrix (1999).mkv", "filename": "The Matrix (1999).mkv", "category": "movie", "title": "The Matrix", "year": 1999, "confidence": 0.9, "needs_review": False, "video_metadata": { "size_bytes": 1234567890, "modified_timestamp": "2024-01-01T10:00:00+00:00", "resolution": "1920x1080", "codec": "h264", "duration_seconds": 7200.5, "bitrate_kbps": 5000, }, } vf = _video_file_from_record(record) assert vf.path == Path("/library/movie/The Matrix (1999).mkv") assert vf.filename == "The Matrix (1999).mkv" assert vf.category == "movie" assert vf.size_bytes == 1234567890 assert vf.resolution == "1920x1080" assert vf.codec == "h264" assert vf.duration_seconds == 7200.5 assert vf.bitrate_kbps == 5000 assert isinstance(vf.modified_timestamp, datetime) def test_video_file_from_record_without_metadata_v1(self): """Test loading VideoFile from v1 record without metadata (backward compatibility).""" record = { "path": "/library/movie/Inception (2010).mkv", "filename": "Inception (2010).mkv", "category": "movie", "title": "Inception", "year": 2010, "confidence": 0.9, "needs_review": False, } vf = _video_file_from_record(record) assert vf.path == Path("/library/movie/Inception (2010).mkv") assert vf.filename == "Inception (2010).mkv" assert vf.category == "movie" # v1 defaults assert vf.size_bytes == 0 assert vf.resolution is None assert vf.codec is None assert vf.duration_seconds is None assert vf.bitrate_kbps is None assert isinstance(vf.modified_timestamp, datetime) def test_video_file_from_record_partial_metadata(self): """Test loading VideoFile with partial metadata (some fields missing).""" record = { "path": "/library/series/Breaking Bad S01E01.mkv", "filename": "Breaking Bad S01E01.mkv", "category": "series", "title": "Breaking Bad", "season": 1, "episodes": [1], "confidence": 0.9, "needs_review": False, "video_metadata": { "size_bytes": 987654321, "modified_timestamp": "2024-02-01T15:30:00+00:00", "resolution": "1280x720", # codec missing # duration_seconds missing # bitrate_kbps missing }, } vf = _video_file_from_record(record) assert vf.size_bytes == 987654321 assert vf.resolution == "1280x720" assert vf.codec is None assert vf.duration_seconds is None assert vf.bitrate_kbps is None class TestExecutionPlanIo: """Tests for the validated typed execution plan boundary.""" def test_execution_plan_record_round_trip(self): plan = ExecutionPlan( plan_id="plan-123", created_at=datetime(2026, 4, 7, 12, 0, tzinfo=timezone.utc), operations=[ FileOperation( operation_type="move", source_path=Path("/library/movie/source.mkv"), destination_path=Path("/library/movie/target.mkv"), reason="move movie", has_conflict=False, conflict_reason=None, ) ], summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0}, summary_by_reason={"move movie": 1}, human_summary="计划已生成", metadata={"analysis_source": "analysis.json"}, ) record = execution_plan_to_record(plan) loaded = execution_plan_from_record(record) assert loaded.plan_id == plan.plan_id assert loaded.created_at == plan.created_at assert loaded.operations[0].source_path == plan.operations[0].source_path assert loaded.operations[0].destination_path == plan.operations[0].destination_path assert loaded.summary == plan.summary assert loaded.summary_by_reason == plan.summary_by_reason assert loaded.human_summary == plan.human_summary assert loaded.metadata == plan.metadata def test_load_execution_plan_normalizes_naive_timestamp(self, tmp_path): plan_path = tmp_path / "plan.json" plan_path.write_text( json.dumps( { "vlm_schema_version": "1.0", "plan_id": "plan-naive", "created_at": "2026-04-07T12:00:00", "operations": [ { "operation_type": "no-op", "source_path": "/library/movie/source.mkv", "destination_path": None, "reason": "manual review", "has_conflict": False, "conflict_reason": None, } ], "summary": {"total": 1, "move": 0, "rename": 0, "quarantine": 0, "no-op": 1}, "summary_by_reason": {"manual review": 1}, "human_summary": "summary", "metadata": {}, } ), encoding="utf-8", ) loaded = load_execution_plan(plan_path) assert loaded.created_at.tzinfo == timezone.utc assert loaded.created_at.isoformat() == "2026-04-07T12:00:00+00:00" def test_save_execution_plan_writes_validated_schema(self, tmp_path): plan = ExecutionPlan( plan_id="plan-save", created_at=datetime(2026, 4, 7, 13, 0, tzinfo=timezone.utc), operations=[ FileOperation( operation_type="quarantine", source_path=Path("/library/movie/duplicate.mkv"), destination_path=None, reason="duplicate", has_conflict=False, conflict_reason=None, ) ], summary={"total": 1, "move": 0, "rename": 0, "quarantine": 1, "no-op": 0}, ) output_path = tmp_path / "plan.json" save_execution_plan(plan, output_path) saved = json.loads(output_path.read_text(encoding="utf-8")) assert saved["vlm_schema_version"] == "1.0" assert saved["plan_id"] == "plan-save" assert saved["operations"][0]["operation_type"] == "quarantine" assert saved["operations"][0]["source_path"] == "/library/movie/duplicate.mkv" class TestIdentitiesJsonVersioning: """Tests for identities.json schema versioning.""" def test_load_identities_v2_with_metadata(self, tmp_path): """Test loading v2 identities.json with video metadata.""" identities_file = tmp_path / "identities.json" data = { "vlm_schema_version": "2.0", "metadata": { "generated": "2024-01-01T10:00:00", "source_inventory": "inventory.csv", "total_files": 1, }, "movies": [ { "path": "/library/movie/Test (2024).mkv", "filename": "Test (2024).mkv", "category": "movie", "title": "Test", "year": 2024, "confidence": 0.9, "needs_review": False, "video_metadata": { "size_bytes": 1000000, "modified_timestamp": "2024-01-01T00:00:00+00:00", "resolution": "1920x1080", "codec": "h264", "duration_seconds": 3600.0, "bitrate_kbps": 2500, }, } ], "series": [], "anime": [], "other": [], } save_identities_json(data, identities_file) loaded = load_identities_json(identities_file) assert loaded["vlm_schema_version"] == "2.0" assert len(loaded["movies"]) == 1 assert "video_metadata" in loaded["movies"][0] assert loaded["movies"][0]["video_metadata"]["size_bytes"] == 1000000 def test_load_identities_v1_backward_compatibility(self, tmp_path): """Test loading v1 identities.json without version field (backward compatibility).""" identities_file = tmp_path / "identities.json" data = { # No vlm_schema_version field (v1) "metadata": { "generated": "2024-01-01T10:00:00", "source_inventory": "inventory.csv", "total_files": 1, }, "movies": [ { "path": "/library/movie/Old (2023).mkv", "filename": "Old (2023).mkv", "category": "movie", "title": "Old", "year": 2023, "confidence": 0.9, "needs_review": False, # No video_metadata field } ], "series": [], "anime": [], "other": [], } save_identities_json(data, identities_file) loaded = load_identities_json(identities_file) # v1 files don't have version field assert "vlm_schema_version" not in loaded or loaded.get("vlm_schema_version") is None assert len(loaded["movies"]) == 1 assert "video_metadata" not in loaded["movies"][0] class TestIdentitiesJsonRoundTrip: """Tests for identities.json save/load round-trip.""" def test_save_and_load_v2_identities(self, tmp_path): """Test saving and loading v2 identities with metadata.""" identities_file = tmp_path / "identities_v2.json" original_data = { "vlm_schema_version": "2.0", "metadata": { "generated": "2024-02-13T12:00:00", "source_inventory": "test_inventory.csv", "total_files": 2, }, "movies": [ { "path": "/library/movie/Movie1 (2024).mkv", "filename": "Movie1 (2024).mkv", "category": "movie", "title": "Movie1", "year": 2024, "confidence": 0.9, "needs_review": False, "video_metadata": { "size_bytes": 2000000000, "modified_timestamp": "2024-02-10T08:00:00+00:00", "resolution": "3840x2160", "codec": "hevc", "duration_seconds": 7200.0, "bitrate_kbps": 8000, }, } ], "series": [ { "path": "/library/series/Show S01E01.mkv", "filename": "Show S01E01.mkv", "category": "series", "title": "Show", "season": 1, "episodes": [1], "confidence": 0.9, "needs_review": False, "video_metadata": { "size_bytes": 500000000, "modified_timestamp": "2024-02-11T10:00:00+00:00", "resolution": "1920x1080", "codec": "h264", "duration_seconds": 2700.0, "bitrate_kbps": 3000, }, } ], "anime": [], "other": [], } save_identities_json(original_data, identities_file) loaded_data = load_identities_json(identities_file) # Verify structure assert loaded_data["vlm_schema_version"] == "2.0" assert len(loaded_data["movies"]) == 1 assert len(loaded_data["series"]) == 1 # Verify movie metadata preserved movie = loaded_data["movies"][0] assert movie["video_metadata"]["size_bytes"] == 2000000000 assert movie["video_metadata"]["resolution"] == "3840x2160" assert movie["video_metadata"]["codec"] == "hevc" # Verify series metadata preserved series = loaded_data["series"][0] assert series["video_metadata"]["size_bytes"] == 500000000 assert series["video_metadata"]["resolution"] == "1920x1080" assert series["video_metadata"]["codec"] == "h264" class TestJsonSchemaValidation: """Tests for runtime schema validation on JSON artifacts.""" def test_save_identities_rejects_invalid_movie_shape(self, tmp_path): identities_file = tmp_path / "identities.json" invalid_data = { "vlm_schema_version": "2.0", "metadata": { "generated": "2024-02-13T12:00:00", "source_inventory": "test_inventory.csv", "total_files": 1, }, "movies": [ { "path": "/library/movie/Movie1 (2024).mkv", "filename": "Movie1 (2024).mkv", "category": "movie", "title": "Movie1", "year": 2024, "confidence": "high", "needs_review": False, } ], "series": [], "anime": [], "other": [], } with pytest.raises(ValueError, match="confidence"): save_identities_json(invalid_data, identities_file) def test_load_analysis_rejects_missing_required_fields(self, tmp_path): analysis_file = tmp_path / "analysis.json" analysis_file.write_text( json.dumps( { "vlm_schema_version": "1.0", "metadata": {"generated": "2024-01-01T00:00:00"}, "completeness": [ { "series_title": "Show", "season": 1, "episodes_found": [1], "episodes_missing": [2], } ], "duplicates": [ { "identity": { "type": "movie", "title": "Test", "year": 2024, "season": None, "episodes": [], }, "files": ["/library/movie/Test (2024).mkv"], "quality_comparison": [], } ], } ), encoding="utf-8", ) loaded = load_analysis_json(analysis_file) assert loaded["metadata"]["generated"] == "2024-01-01T00:00:00" assert loaded["duplicates"][0]["identity"]["title"] == "Test"