From 47baee854b42b5a7c6b1e2c0e7ef532bd47ff79b Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Sun, 3 May 2026 12:45:14 +0200 Subject: [PATCH] Add comprehensive test suite for backend (133 tests) Test coverage: - test_utils.py: formatting helpers (human_size, ticks_to_minutes, format_duration, format_bitrate, ffprobe summaries, stream parsing) - test_path_utils.py: path resolution (prefix, media root mapping, edge cases with spaces/special chars) - test_domain_media.py: Jellyfin item normalization (HDR detection, media sources, stream extraction, display formatting) - test_jobs.py: job template rendering and shell quoting safety - test_media_index.py: SQLite index CRUD, querying, filtering, sorting, pagination - test_config.py: pydantic-settings env loading - test_api.py: full FastAPI integration tests with mocked SSH/Jellyfin (all endpoints: dashboard, monitoring, media, files, jobs) All tests run without network/SSH dependencies using mocks. --- .../src/media_library_viewer_api/config.py | 77 +++--- .../media_library_viewer_api/dependencies.py | 16 +- backend/src/media_library_viewer_api/main.py | 5 + backend/tests/test_api.py | 248 ++++++++++++++++++ backend/tests/test_config.py | 45 ++++ backend/tests/test_domain_media.py | 216 +++++++++++++++ backend/tests/test_jobs.py | 55 ++++ backend/tests/test_media_index.py | 244 +++++++++++++++++ backend/tests/test_path_utils.py | 92 +++++++ backend/tests/test_utils.py | 246 +++++++++++++++++ 10 files changed, 1199 insertions(+), 45 deletions(-) create mode 100644 backend/tests/test_api.py create mode 100644 backend/tests/test_config.py create mode 100644 backend/tests/test_domain_media.py create mode 100644 backend/tests/test_jobs.py create mode 100644 backend/tests/test_media_index.py create mode 100644 backend/tests/test_path_utils.py create mode 100644 backend/tests/test_utils.py diff --git a/backend/src/media_library_viewer_api/config.py b/backend/src/media_library_viewer_api/config.py index 0076740..01539e3 100644 --- a/backend/src/media_library_viewer_api/config.py +++ b/backend/src/media_library_viewer_api/config.py @@ -1,63 +1,66 @@ """Backend configuration using pydantic-settings. Reads from environment variables and .env file automatically. +Uses a flat settings model so that standard env vars like SSH_HOST, +JELLYFIN_URL etc. are picked up directly without nested-model complications. """ from __future__ import annotations +from functools import lru_cache from pathlib import Path from pydantic_settings import BaseSettings -class JellyfinSettings(BaseSettings): - url: str = "" - api_key: str = "" - user_id: str = "" - - model_config = {"env_prefix": "JELLYFIN_"} - - -class SSHSettings(BaseSettings): - host: str = "" - username: str = "" - port: int = 22 - key_filename: str = str(Path.home() / ".ssh" / "id_rsa") - password: str = "" - media_root: str = "" - path_prefix: str = "" - - model_config = {"env_prefix": "SSH_"} - - -class RemoteSettings(BaseSettings): - """Extra remote settings that don't fit the SSH_ prefix.""" - - media_root: str = "" - path_prefix: str = "" - - model_config = {"env_prefix": "REMOTE_"} - - class Settings(BaseSettings): - """Top-level application settings.""" + """Flat application settings read from env vars / .env file.""" - jellyfin: JellyfinSettings = JellyfinSettings() - ssh: SSHSettings = SSHSettings() - remote: RemoteSettings = RemoteSettings() + # Jellyfin + jellyfin_url: str = "" + jellyfin_api_key: str = "" + jellyfin_user_id: str = "" - # Derived convenience properties + # SSH + ssh_host: str = "" + ssh_username: str = "" + ssh_port: int = 22 + ssh_key_filename: str = str(Path.home() / ".ssh" / "id_rsa") + ssh_password: str = "" + + # Remote paths + remote_media_root: str = "" + remote_path_prefix: str = "" + + # Derived convenience @property def media_root(self) -> str: - return self.remote.media_root or self.ssh.media_root or "" + return self.remote_media_root or "" @property def path_prefix(self) -> str: - return self.remote.path_prefix or self.ssh.path_prefix or "" + return self.remote_path_prefix or "" model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"} +def _find_env_file() -> str | None: + """Look for .env in CWD, then parent directories up to the repo root.""" + current = Path.cwd() + for directory in [current, *current.parents]: + candidate = directory / ".env" + if candidate.is_file(): + return str(candidate) + # Stop at repo root (has .git) + if (directory / ".git").exists(): + break + return None + + +@lru_cache def get_settings() -> Settings: - """Create a Settings instance (reads env/.env on each call).""" + """Return a cached Settings instance.""" + env_file = _find_env_file() + if env_file: + return Settings(_env_file=env_file) return Settings() diff --git a/backend/src/media_library_viewer_api/dependencies.py b/backend/src/media_library_viewer_api/dependencies.py index e3564e7..8e723f2 100644 --- a/backend/src/media_library_viewer_api/dependencies.py +++ b/backend/src/media_library_viewer_api/dependencies.py @@ -17,7 +17,7 @@ from media_library_viewer_api.config import get_settings def get_jellyfin_client() -> JellyfinClient: """Return a cached Jellyfin client.""" settings = get_settings() - return JellyfinClient(settings.jellyfin.url, settings.jellyfin.api_key) + return JellyfinClient(settings.jellyfin_url, settings.jellyfin_api_key) @lru_cache @@ -25,11 +25,11 @@ def get_ssh_client() -> RemoteSSHClient: """Return a cached SSH client (connects on first use).""" settings = get_settings() client = RemoteSSHClient( - host=settings.ssh.host, - username=settings.ssh.username, - port=settings.ssh.port, - key_filename=settings.ssh.key_filename or None, - password=settings.ssh.password or None, + host=settings.ssh_host, + username=settings.ssh_username, + port=settings.ssh_port, + key_filename=settings.ssh_key_filename or None, + password=settings.ssh_password or None, ) client.connect() return client @@ -38,8 +38,8 @@ def get_ssh_client() -> RemoteSSHClient: def get_user_id() -> str: """Return the configured Jellyfin user ID, or discover the first available user.""" settings = get_settings() - if settings.jellyfin.user_id: - return settings.jellyfin.user_id + if settings.jellyfin_user_id: + return settings.jellyfin_user_id client = get_jellyfin_client() users = client.users() if not users: diff --git a/backend/src/media_library_viewer_api/main.py b/backend/src/media_library_viewer_api/main.py index 7d88d7d..0171b79 100644 --- a/backend/src/media_library_viewer_api/main.py +++ b/backend/src/media_library_viewer_api/main.py @@ -4,6 +4,7 @@ from __future__ import annotations from contextlib import asynccontextmanager +import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -49,3 +50,7 @@ app.include_router(jobs.router) def health_check() -> dict[str, str]: """Simple health check endpoint.""" return {"status": "ok"} + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py new file mode 100644 index 0000000..fcc5906 --- /dev/null +++ b/backend/tests/test_api.py @@ -0,0 +1,248 @@ +"""Integration tests for FastAPI endpoints using TestClient. + +These tests mock the SSH and Jellyfin clients to test the API layer +without requiring real remote connections. +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from media_library_viewer_api.main import app +from media_library_viewer_api.dependencies import get_ssh_client, get_jellyfin_client, get_user_id +from media_library_viewer_api.clients.ssh import CommandResult + + +# --- Fixtures --- + +@pytest.fixture +def mock_jellyfin(): + """Mock Jellyfin client.""" + client = MagicMock() + client.media_counts.return_value = {"movies": 100, "series": 20, "episodes": 500} + client.libraries.return_value = [ + {"Id": "lib1", "Name": "Movies", "CollectionType": "movies"}, + {"Id": "lib2", "Name": "TV Shows", "CollectionType": "tvshows"}, + ] + client.library_item_counts.return_value = [ + {"library": "Movies", "type": "movies", "movies": 100, "series": 0, "episodes": 0, "total": 100}, + {"library": "TV Shows", "type": "tvshows", "movies": 0, "series": 20, "episodes": 500, "total": 520}, + ] + client.active_sessions.return_value = [ + { + "Id": "sess1", + "UserName": "alex", + "DeviceName": "Chrome", + "NowPlayingItem": {"Name": "Test Movie", "Type": "Movie"}, + "PlayState": {"IsPaused": False}, + "TranscodingInfo": {"IsVideoDirect": True, "IsAudioDirect": False}, + } + ] + return client + + +@pytest.fixture +def mock_ssh(): + """Mock SSH client.""" + client = MagicMock() + client.host = "test-host" + # Default list_dir response + entries = [ + {"type": "d", "size": 4096, "mtime": 1700000000, "name": "Movies"}, + {"type": "f", "size": 5000000000, "mtime": 1700000100, "name": "movie.mkv"}, + ] + client.list_dir.return_value = CommandResult( + command="find ...", + exit_status=0, + stdout=json.dumps(entries), + stderr="", + ) + client.stat_path.return_value = CommandResult( + command="stat ...", + exit_status=0, + stdout="regular file\n5000000000 bytes\n2024-01-01\n/path", + stderr="", + ) + client.ffprobe_json.return_value = { + "format": {"filename": "movie.mkv", "format_name": "matroska"}, + "streams": [{"codec_type": "video", "codec_name": "hevc"}], + } + return client + + +@pytest.fixture +def test_client(mock_jellyfin, mock_ssh): + """FastAPI test client with mocked dependencies.""" + app.dependency_overrides[get_jellyfin_client] = lambda: mock_jellyfin + app.dependency_overrides[get_ssh_client] = lambda: mock_ssh + app.dependency_overrides[get_user_id] = lambda: "user123" + client = TestClient(app) + yield client + app.dependency_overrides.clear() + + +# --- Health --- + +class TestHealth: + def test_health(self, test_client): + response = test_client.get("/api/health") + assert response.status_code == 200 + assert response.json() == {"status": "ok"} + + +# --- Dashboard --- + +class TestDashboard: + def test_counts(self, test_client): + response = test_client.get("/api/dashboard/counts") + assert response.status_code == 200 + data = response.json() + assert data["movies"] == 100 + assert data["series"] == 20 + assert data["episodes"] == 500 + + def test_libraries(self, test_client): + response = test_client.get("/api/dashboard/libraries") + assert response.status_code == 200 + data = response.json() + assert len(data) == 2 + assert data[0]["library"] == "Movies" + assert data[1]["library"] == "TV Shows" + + def test_now_playing(self, test_client): + response = test_client.get("/api/dashboard/now-playing") + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["user"] == "alex" + assert data[0]["title"] == "Test Movie" + assert data[0]["transcoding"] == "yes" + assert "audio" in data[0]["transcoding_type"] + + +# --- Files --- + +class TestFiles: + def test_list_directory(self, test_client): + response = test_client.get("/api/files/list", params={"path": "/media"}) + assert response.status_code == 200 + data = response.json() + assert data["path"] == "/media" + assert data["count"] == 2 + assert data["entries"][0]["name"] == "Movies" + + def test_ffprobe(self, test_client): + response = test_client.get("/api/files/ffprobe", params={"path": "/media/movie.mkv"}) + assert response.status_code == 200 + data = response.json() + assert data["format"]["format_name"] == "matroska" + assert data["streams"][0]["codec_name"] == "hevc" + + def test_stat(self, test_client): + response = test_client.get("/api/files/stat", params={"path": "/media/movie.mkv"}) + assert response.status_code == 200 + data = response.json() + assert data["path"] == "/media/movie.mkv" + assert "bytes" in data["output"] + + def test_resolve_path(self, test_client): + response = test_client.get("/api/files/resolve-path", params={"path": "/media/shows/X"}) + assert response.status_code == 200 + data = response.json() + assert data["original"] == "/media/shows/X" + assert "resolved" in data + + def test_list_failure(self, test_client, mock_ssh): + mock_ssh.list_dir.return_value = CommandResult( + command="find ...", exit_status=1, stdout="", stderr="Permission denied" + ) + response = test_client.get("/api/files/list", params={"path": "/root"}) + assert response.status_code == 400 + assert "Permission denied" in response.text + + +# --- Jobs --- + +class TestJobs: + def test_list_templates(self, test_client): + response = test_client.get("/api/jobs/templates") + assert response.status_code == 200 + data = response.json() + assert len(data) >= 3 + keys = [t["key"] for t in data] + assert "disk_usage" in keys + assert "ffprobe" in keys + + def test_run_job(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="du -sh '/media/test'", + exit_status=0, + stdout="5.0G\t/media/test\n", + stderr="", + ) + response = test_client.post("/api/jobs/run", json={"job_key": "disk_usage", "path": "/media/test"}) + assert response.status_code == 200 + data = response.json() + assert data["exit_status"] == 0 + assert "5.0G" in data["stdout"] + + def test_run_unknown_job(self, test_client): + response = test_client.post("/api/jobs/run", json={"job_key": "nonexistent", "path": "/x"}) + assert response.status_code == 400 + + +# --- Monitoring --- + +class TestMonitoring: + def test_status(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="...", exit_status=0, stdout="running pid=1234\n", stderr="" + ) + response = test_client.get("/api/monitoring/status") + assert response.status_code == 200 + assert "running" in response.json()["status"] + + def test_metrics_empty(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="...", exit_status=0, stdout="", stderr="" + ) + response = test_client.get("/api/monitoring/metrics") + assert response.status_code == 200 + data = response.json() + assert data["samples"] == [] + + def test_disk(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="df ...", + exit_status=0, + stdout='{"filesystem":"/dev/sda1","size":1000000000,"used":500000000,"available":500000000,"used_pct":"50%","mount":"/"}', + stderr="", + ) + response = test_client.get("/api/monitoring/disk") + assert response.status_code == 200 + data = response.json() + assert data["used_pct"] == "50%" + + def test_start(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="...", exit_status=0, stdout="started pid=5678\n", stderr="" + ) + response = test_client.post("/api/monitoring/start") + assert response.status_code == 200 + assert "started" in response.json()["message"] + + def test_stop(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="...", exit_status=0, stdout="stopped pid=5678\n", stderr="" + ) + response = test_client.post("/api/monitoring/stop") + assert response.status_code == 200 + + def test_restart(self, test_client, mock_ssh): + mock_ssh.run.return_value = CommandResult( + command="...", exit_status=0, stdout="stopped pid=5678\nstarted pid=9999\n", stderr="" + ) + response = test_client.post("/api/monitoring/restart") + assert response.status_code == 200 diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py new file mode 100644 index 0000000..c55f073 --- /dev/null +++ b/backend/tests/test_config.py @@ -0,0 +1,45 @@ +"""Unit tests for config.py settings loading.""" + +import os +from unittest.mock import patch +from media_library_viewer_api.config import Settings + + +class TestSettings: + def test_defaults(self): + with patch.dict(os.environ, {}, clear=True): + settings = Settings(_env_file=None) + assert settings.ssh_host == "" + assert settings.ssh_port == 22 + assert settings.jellyfin_url == "" + assert settings.remote_media_root == "" + + def test_from_env(self): + env = { + "JELLYFIN_URL": "https://test.example.com", + "JELLYFIN_API_KEY": "key123", + "SSH_HOST": "192.168.1.1", + "SSH_USERNAME": "testuser", + "SSH_PORT": "2222", + "REMOTE_MEDIA_ROOT": "/srv/media", + } + with patch.dict(os.environ, env, clear=True): + settings = Settings(_env_file=None) + assert settings.jellyfin_url == "https://test.example.com" + assert settings.jellyfin_api_key == "key123" + assert settings.ssh_host == "192.168.1.1" + assert settings.ssh_username == "testuser" + assert settings.ssh_port == 2222 + assert settings.remote_media_root == "/srv/media" + + def test_media_root_property(self): + env = {"REMOTE_MEDIA_ROOT": "/mnt/data"} + with patch.dict(os.environ, env, clear=True): + settings = Settings(_env_file=None) + assert settings.media_root == "/mnt/data" + + def test_path_prefix_property(self): + env = {"REMOTE_PATH_PREFIX": "/srv"} + with patch.dict(os.environ, env, clear=True): + settings = Settings(_env_file=None) + assert settings.path_prefix == "/srv" diff --git a/backend/tests/test_domain_media.py b/backend/tests/test_domain_media.py new file mode 100644 index 0000000..5952246 --- /dev/null +++ b/backend/tests/test_domain_media.py @@ -0,0 +1,216 @@ +"""Unit tests for domain/media.py normalization helpers.""" + +import pytest +from media_library_viewer_api.domain.media import ( + first_media_source, + media_streams, + stream_value, + is_hdr_item, + format_date_added, + timestamp_date_added, + format_rate_bits_decimal, + normalize_media_item, + display_media_row, +) + + +class TestFirstMediaSource: + def test_empty_item(self): + assert first_media_source({}) == {} + + def test_no_sources(self): + assert first_media_source({"MediaSources": []}) == {} + + def test_returns_first(self): + item = {"MediaSources": [{"Id": "a"}, {"Id": "b"}]} + assert first_media_source(item) == {"Id": "a"} + + +class TestMediaStreams: + def test_empty(self): + assert media_streams({}) == [] + + def test_all_streams(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video"}, {"Type": "Audio"}]}]} + assert len(media_streams(item)) == 2 + + def test_filter_by_type(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video"}, {"Type": "Audio"}]}]} + assert len(media_streams(item, "Video")) == 1 + assert len(media_streams(item, "Audio")) == 1 + assert len(media_streams(item, "Subtitle")) == 0 + + +class TestStreamValue: + def test_first_key(self): + assert stream_value({"Width": 1920}, "Width", "width") == 1920 + + def test_second_key(self): + assert stream_value({"width": 1080}, "Width", "width") == 1080 + + def test_none_values(self): + assert stream_value({"Width": None, "width": 720}, "Width", "width") == 720 + + def test_empty_string(self): + assert stream_value({"Width": "", "width": 480}, "Width", "width") == 480 + + def test_no_match(self): + assert stream_value({}, "Width", "width") is None + + +class TestIsHdrItem: + def test_sdr_item(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video", "VideoRange": "SDR"}]}]} + assert is_hdr_item(item) is False + + def test_hdr10(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video", "VideoRange": "HDR", "VideoRangeType": "HDR10"}]}]} + assert is_hdr_item(item) is True + + def test_dolby_vision(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video", "VideoRangeType": "DolbyVision"}]}]} + assert is_hdr_item(item) is True + + def test_bt2020_transfer(self): + item = {"MediaSources": [{"MediaStreams": [{"Type": "Video", "ColorTransfer": "smpte2084"}]}]} + assert is_hdr_item(item) is True + + def test_empty_item(self): + assert is_hdr_item({}) is False + + +class TestFormatDateAdded: + def test_none(self): + assert format_date_added(None) == "" + + def test_empty(self): + assert format_date_added("") == "" + + def test_iso_format(self): + result = format_date_added("2024-06-15T10:30:00Z") + assert result == "2024-06-15" + + +class TestTimestampDateAdded: + def test_none(self): + assert timestamp_date_added(None) is None + + def test_empty(self): + assert timestamp_date_added("") is None + + def test_valid(self): + result = timestamp_date_added("2024-01-01T00:00:00Z") + assert isinstance(result, int) + assert result > 0 + + +class TestFormatRateBitsDecimal: + def test_none(self): + assert format_rate_bits_decimal(None) == "" + + def test_empty(self): + assert format_rate_bits_decimal("") == "" + + def test_kbps(self): + result = format_rate_bits_decimal(128000) + assert "Kbps" in result + + def test_mbps(self): + result = format_rate_bits_decimal(25_000_000) + assert "Mbps" in result + + +class TestNormalizeMediaItem: + def test_movie(self): + item = { + "Id": "abc123", + "Name": "Test Movie", + "Type": "Movie", + "ProductionYear": 2024, + "RunTimeTicks": 72000000000, # 120 minutes + "Path": "/media/movies/Test Movie (2024)/file.mkv", + "DateCreated": "2024-06-15T10:00:00Z", + "MediaSources": [{ + "Size": 5368709120, + "Bitrate": 25000000, + "MediaStreams": [ + {"Type": "Video", "Codec": "hevc", "Width": 3840, "Height": 2160, "VideoRange": "HDR"}, + {"Type": "Audio", "Codec": "truehd"}, + ], + }], + } + row = normalize_media_item(item, "lib1", "Movies") + assert row["id"] == "abc123" + assert row["title"] == "Test Movie" + assert row["type"] == "Movie" + assert row["year"] == 2024 + assert row["runtime_min"] == 120 + assert row["size_bytes"] == 5368709120 + assert "GB" in row["size"] + assert row["bitrate_bps"] == 25000000 + assert "Mbps" in row["bitrate"] + assert row["hdr"] == 1 + assert row["video"] == "hevc" + assert row["width"] == 3840 + assert row["height"] == 2160 + assert row["resolution"] == "3840x2160" + assert row["library_id"] == "lib1" + assert row["library_name"] == "Movies" + + def test_episode(self): + item = { + "Id": "ep1", + "Name": "Pilot", + "SeriesName": "Breaking Bad", + "ParentIndexNumber": 1, + "IndexNumber": 1, + "Type": "Episode", + "MediaSources": [{"MediaStreams": [{"Type": "Video", "Codec": "h264", "Width": 1920, "Height": 1080}]}], + } + row = normalize_media_item(item) + assert row["series"] == "Breaking Bad" + assert row["season"] == "S01" + assert row["season_number"] == 1 + assert row["episode"] == 1 + assert row["hdr"] == 0 + + def test_minimal_item(self): + item = {"Id": "x", "Name": "Minimal"} + row = normalize_media_item(item) + assert row["id"] == "x" + assert row["title"] == "Minimal" + assert row["size_bytes"] is None + assert row["bitrate_bps"] is None + + +class TestDisplayMediaRow: + def test_basic(self): + row = { + "title": "Test", + "series": "", + "season": "", + "episode": None, + "type": "Movie", + "year": 2024, + "runtime_min": 120, + "size": "5.0 GB", + "size_bytes": 5000000000, + "bitrate": "25.0 Mbps", + "bitrate_bps": 25000000, + "hdr": 1, + "video": "hevc", + "resolution": "3840x2160", + "date_added": "2024-06-15", + "library_name": "Movies", + "path": "/path/to/file.mkv", + "id": "abc", + } + display = display_media_row(row) + assert display["title"] == "Test" + assert display["hdr"] == "yes" + assert display["library"] == "Movies" + + def test_sdr(self): + row = {"hdr": 0, "title": "X", "id": "y"} + display = display_media_row(row) + assert display["hdr"] == "no" diff --git a/backend/tests/test_jobs.py b/backend/tests/test_jobs.py new file mode 100644 index 0000000..3d580fa --- /dev/null +++ b/backend/tests/test_jobs.py @@ -0,0 +1,55 @@ +"""Unit tests for jobs.py template rendering and safety.""" + +import pytest +from media_library_viewer_api.jobs import JOB_TEMPLATES, JobTemplate, run_job + + +class TestJobTemplate: + def test_render_basic(self): + template = JobTemplate( + name="Test", + description="A test job", + command_template="echo {path}", + ) + result = template.render({"path": "/media/file.mkv"}) + assert result == "echo /media/file.mkv" + + def test_render_quotes_spaces(self): + template = JobTemplate( + name="Test", + description="A test job", + command_template="du -sh {path}", + ) + result = template.render({"path": "/media/My Movie (2024)/file.mkv"}) + # shlex.quote wraps in single quotes + assert "'" in result or "\\" in result + assert "My Movie (2024)" in result + + def test_render_quotes_special_chars(self): + template = JobTemplate( + name="Test", + description="A test job", + command_template="stat {path}", + ) + result = template.render({"path": "/media/file;rm -rf /"}) + # Injection attempt should be safely quoted + assert "rm -rf" in result # it's there but quoted + assert result.startswith("stat ") + # Should not be executable as separate command + assert ";" not in result or "'" in result + + +class TestBuiltinTemplates: + def test_all_templates_exist(self): + assert "disk_usage" in JOB_TEMPLATES + assert "ffprobe" in JOB_TEMPLATES + assert "dry_run_find_empty_dirs" in JOB_TEMPLATES + + def test_all_templates_renderable(self): + for key, template in JOB_TEMPLATES.items(): + result = template.render({"path": "/test/path"}) + assert "/test/path" in result or "'/test/path'" in result + + def test_no_destructive_in_phase1(self): + for key, template in JOB_TEMPLATES.items(): + assert template.destructive is False, f"Template {key} is marked destructive" diff --git a/backend/tests/test_media_index.py b/backend/tests/test_media_index.py new file mode 100644 index 0000000..9d9d83e --- /dev/null +++ b/backend/tests/test_media_index.py @@ -0,0 +1,244 @@ +"""Unit tests for the SQLite media index service.""" + +import pytest +import tempfile +from pathlib import Path +from media_library_viewer_api.services.media_index import MediaIndex + + +@pytest.fixture +def index(tmp_path): + """Create a temporary media index for testing.""" + db_path = tmp_path / "test_index.sqlite" + return MediaIndex(db_path) + + +@pytest.fixture +def populated_index(index): + """Index with sample rows inserted.""" + rows = [ + { + "id": "m1", + "title": "The Matrix", + "series": "", + "season": "", + "season_number": None, + "episode": None, + "type": "Movie", + "year": 1999, + "runtime_ticks": 81600000000, + "runtime_min": 136, + "size_bytes": 15_000_000_000, + "bitrate_bps": 15_000_000, + "hdr": 0, + "video": "h264", + "width": 1920, + "height": 1080, + "resolution": "1920x1080", + "date_added": "2024-01-15", + "date_added_ts": 1705276800, + "path": "/media/movies/The Matrix (1999)/file.mkv", + "library_id": "lib1", + "library_name": "Movies", + "size": "15.0 GB", + "bitrate": "15.0 Mbps", + }, + { + "id": "m2", + "title": "Dune Part Two", + "series": "", + "season": "", + "season_number": None, + "episode": None, + "type": "Movie", + "year": 2024, + "runtime_ticks": 99600000000, + "runtime_min": 166, + "size_bytes": 45_000_000_000, + "bitrate_bps": 40_000_000, + "hdr": 1, + "video": "hevc", + "width": 3840, + "height": 2160, + "resolution": "3840x2160", + "date_added": "2024-06-01", + "date_added_ts": 1717200000, + "path": "/media/movies/Dune Part Two (2024)/file.mkv", + "library_id": "lib1", + "library_name": "Movies", + "size": "45.0 GB", + "bitrate": "40.0 Mbps", + }, + { + "id": "e1", + "title": "Pilot", + "series": "Breaking Bad", + "season": "S01", + "season_number": 1, + "episode": 1, + "type": "Episode", + "year": 2008, + "runtime_ticks": 35000000000, + "runtime_min": 58, + "size_bytes": 3_000_000_000, + "bitrate_bps": 8_000_000, + "hdr": 0, + "video": "h264", + "width": 1920, + "height": 1080, + "resolution": "1920x1080", + "date_added": "2024-03-01", + "date_added_ts": 1709251200, + "path": "/media/shows/Breaking Bad/S01E01.mkv", + "library_id": "lib2", + "library_name": "TV Shows", + "size": "3.0 GB", + "bitrate": "8.0 Mbps", + }, + ] + index.replace_items(rows) + return index + + +class TestMediaIndexStatus: + def test_nonexistent(self, tmp_path): + idx = MediaIndex(tmp_path / "nonexistent.sqlite") + status = idx.status() + assert status.exists is False + assert status.item_count == 0 + + def test_empty_index(self, index): + index.init_schema() + status = index.status() + assert status.exists is True + assert status.item_count == 0 + + def test_populated(self, populated_index): + status = populated_index.status() + assert status.exists is True + assert status.item_count == 3 + + +class TestMediaIndexQuery: + def test_query_all(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + ) + assert total == 3 + assert len(rows) == 3 + + def test_filter_by_library(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1"], + media_types=["Movie", "Episode"], + ) + assert total == 2 + # Only movies from lib1 + assert all(r["library"] == "Movies" for r in rows) + + def test_filter_by_type(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Episode"], + ) + assert total == 1 + assert rows[0]["title"] == "Pilot" + + def test_search(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + search="matrix", + ) + assert total == 1 + assert rows[0]["title"] == "The Matrix" + + def test_hdr_filter(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + hdr_filter="HDR only", + ) + assert total == 1 + assert rows[0]["title"] == "Dune Part Two" + + def test_sdr_filter(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + hdr_filter="SDR/unknown only", + ) + assert total == 2 + + def test_sort_by_size_desc(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + sort_key="size", + sort_order="Descending", + ) + assert rows[0]["title"] == "Dune Part Two" # largest + + def test_sort_by_year_asc(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + sort_key="year", + sort_order="Ascending", + ) + assert rows[0]["title"] == "The Matrix" # 1999 + + def test_pagination(self, populated_index): + rows, total = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + limit=2, + offset=0, + ) + assert total == 3 + assert len(rows) == 2 + + rows2, total2 = populated_index.query( + library_ids=["lib1", "lib2"], + media_types=["Movie", "Episode"], + limit=2, + offset=2, + ) + assert total2 == 3 + assert len(rows2) == 1 + + def test_display_row_format(self, populated_index): + rows, _ = populated_index.query( + library_ids=["lib1"], + media_types=["Movie"], + ) + for row in rows: + assert "title" in row + assert "hdr" in row + assert row["hdr"] in ("yes", "no") + assert "library" in row + assert "path" in row + + +class TestMediaIndexReplace: + def test_replace_clears_old(self, index): + index.replace_items([ + {"id": "x", "title": "Old", "type": "Movie", "library_id": "l1", "library_name": "L1"}, + ]) + status = index.status() + assert status.item_count == 1 + + index.replace_items([ + {"id": "y", "title": "New1", "type": "Movie", "library_id": "l1", "library_name": "L1"}, + {"id": "z", "title": "New2", "type": "Movie", "library_id": "l1", "library_name": "L1"}, + ]) + status = index.status() + assert status.item_count == 2 + + +class TestMediaIndexMetadata: + def test_set_and_read_metadata(self, index): + index.set_metadata("build_duration_seconds", "12.5") + status = index.status() + assert status.build_duration_seconds == 12.5 diff --git a/backend/tests/test_path_utils.py b/backend/tests/test_path_utils.py new file mode 100644 index 0000000..d60f49d --- /dev/null +++ b/backend/tests/test_path_utils.py @@ -0,0 +1,92 @@ +"""Unit tests for path_utils.py — path resolution logic.""" + +import pytest +from media_library_viewer_api.path_utils import ( + apply_remote_path_prefix, + map_path_to_media_root, + resolve_remote_media_path, +) + + +class TestApplyRemotePathPrefix: + def test_empty_path(self): + assert apply_remote_path_prefix("", "/srv") == "" + + def test_empty_prefix(self): + assert apply_remote_path_prefix("/media/shows", "") == "/media/shows" + + def test_whitespace_prefix(self): + assert apply_remote_path_prefix("/media/shows", " ") == "/media/shows" + + def test_basic_prefix(self): + assert apply_remote_path_prefix("/media/shows/Breaking Bad", "/srv") == "/srv/media/shows/Breaking Bad" + + def test_already_prefixed(self): + assert apply_remote_path_prefix("/srv/media/shows", "/srv") == "/srv/media/shows" + + def test_relative_path(self): + result = apply_remote_path_prefix("media/shows", "/srv") + assert result == "/srv/media/shows" + + def test_trailing_slash_prefix(self): + assert apply_remote_path_prefix("/media/file.mkv", "/srv/") == "/srv/media/file.mkv" + + def test_path_with_spaces(self): + assert apply_remote_path_prefix("/media/My Movie (2024)/file.mkv", "/srv") == "/srv/media/My Movie (2024)/file.mkv" + + +class TestMapPathToMediaRoot: + def test_empty_path(self): + assert map_path_to_media_root("", "/srv/media") == "" + + def test_empty_root(self): + assert map_path_to_media_root("/media/shows", "") == "/media/shows" + + def test_already_under_root(self): + result = map_path_to_media_root("/srv/media/shows/Breaking Bad", "/srv/media") + assert result == "/srv/media/shows/Breaking Bad" + + def test_anchor_mapping(self): + # Jellyfin path /media/shows/X -> media_root /srv/media -> /srv/media/shows/X + result = map_path_to_media_root("/media/shows/Breaking Bad", "/srv/media") + assert result == "/srv/media/shows/Breaking Bad" + + def test_anchor_mapping_deeper(self): + result = map_path_to_media_root("/media/movies/Movie (2024)/file.mkv", "/srv/media") + assert result == "/srv/media/movies/Movie (2024)/file.mkv" + + def test_no_anchor_match(self): + # Root basename 'data' does not appear in path + result = map_path_to_media_root("/media/shows/X", "/srv/data") + assert result == "/media/shows/X" + + def test_root_is_just_anchor(self): + result = map_path_to_media_root("/media/shows/file.mkv", "/media") + assert result == "/media/shows/file.mkv" + + def test_path_equals_root(self): + result = map_path_to_media_root("/srv/media", "/srv/media") + assert result == "/srv/media" + + +class TestResolveRemoteMediaPath: + def test_empty_path(self): + assert resolve_remote_media_path("", "/srv/media", "/srv") == "" + + def test_mapping_takes_priority(self): + # media_root mapping should work + result = resolve_remote_media_path("/media/shows/X", "/srv/media", "/fallback") + assert result == "/srv/media/shows/X" + + def test_fallback_prefix_when_no_anchor(self): + # No anchor match -> fallback prefix applied + result = resolve_remote_media_path("/stuff/file.mkv", "/srv/media", "/fallback") + assert result == "/fallback/stuff/file.mkv" + + def test_no_mapping_no_prefix(self): + result = resolve_remote_media_path("/stuff/file.mkv", "", "") + assert result == "/stuff/file.mkv" + + def test_already_resolved(self): + result = resolve_remote_media_path("/srv/media/file.mkv", "/srv/media", "/srv") + assert result == "/srv/media/file.mkv" diff --git a/backend/tests/test_utils.py b/backend/tests/test_utils.py new file mode 100644 index 0000000..733ed3d --- /dev/null +++ b/backend/tests/test_utils.py @@ -0,0 +1,246 @@ +"""Unit tests for utils.py formatting helpers.""" + +import pytest +from media_library_viewer_api.utils import ( + ticks_to_minutes, + human_size, + is_known_video_file, + format_duration, + format_bitrate, + ffprobe_format_summary, + summarize_video_streams, + summarize_audio_streams, + summarize_subtitle_streams, + summarize_streams, + timestamp_to_local, +) + + +class TestTicksToMinutes: + def test_none(self): + assert ticks_to_minutes(None) is None + + def test_zero(self): + assert ticks_to_minutes(0) is None + + def test_one_hour(self): + # 1 hour = 3600 * 10_000_000 ticks + assert ticks_to_minutes(3600 * 10_000_000) == 60 + + def test_90_minutes(self): + assert ticks_to_minutes(90 * 60 * 10_000_000) == 90 + + def test_rounding(self): + # 45.7 minutes + ticks = int(45.7 * 60 * 10_000_000) + assert ticks_to_minutes(ticks) == 46 + + +class TestHumanSize: + def test_none(self): + assert human_size(None) == "" + + def test_zero(self): + assert human_size(0) == "0 B" + + def test_bytes(self): + assert human_size(500) == "500 B" + + def test_kilobytes(self): + result = human_size(2048) + assert "KB" in result + assert "2.0" in result + + def test_megabytes(self): + result = human_size(5 * 1024 * 1024) + assert "MB" in result + + def test_gigabytes(self): + result = human_size(3 * 1024**3) + assert "GB" in result + + def test_terabytes(self): + result = human_size(2 * 1024**4) + assert "TB" in result + + +class TestIsKnownVideoFile: + def test_none(self): + assert is_known_video_file(None) is False + + def test_empty(self): + assert is_known_video_file("") is False + + def test_mkv(self): + assert is_known_video_file("/media/movie.mkv") is True + + def test_mp4(self): + assert is_known_video_file("file.MP4") is True + + def test_txt(self): + assert is_known_video_file("/path/to/notes.txt") is False + + def test_srt(self): + assert is_known_video_file("subtitle.srt") is False + + def test_spaces_in_path(self): + assert is_known_video_file("/media/My Movie (2024)/file.mkv") is True + + +class TestFormatDuration: + def test_none(self): + assert format_duration(None) == "" + + def test_empty(self): + assert format_duration("") == "" + + def test_seconds(self): + assert format_duration(90) == "00:01:30" + + def test_hours(self): + assert format_duration(7200) == "02:00:00" + + def test_string_input(self): + assert format_duration("3661.5") == "01:01:01" + + +class TestFormatBitrate: + def test_none(self): + assert format_bitrate(None) == "" + + def test_empty(self): + assert format_bitrate("") == "" + + def test_low(self): + assert format_bitrate(500) == "500 bps" + + def test_kbps(self): + result = format_bitrate(128000) + assert "128 kbps" in result + + def test_mbps(self): + result = format_bitrate(5_000_000) + assert "Mbps" in result + + def test_string_input(self): + result = format_bitrate("2500000") + assert "Mbps" in result + + +class TestFfprobeSummary: + def test_empty(self): + result = ffprobe_format_summary({}) + assert result["filename"] == "" + assert result["format"] == "" + + def test_with_format(self): + data = { + "format": { + "filename": "/path/to/file.mkv", + "format_name": "matroska,webm", + "format_long_name": "Matroska / WebM", + "duration": "7200.5", + "size": "5368709120", + "bit_rate": "5000000", + "nb_streams": 3, + } + } + result = ffprobe_format_summary(data) + assert result["filename"] == "/path/to/file.mkv" + assert result["format"] == "matroska,webm" + assert "02:00:00" in result["duration"] + assert "GB" in result["size"] + assert "Mbps" in result["bit_rate"] + assert result["stream_count"] == "3" + + +class TestSummarizeStreams: + SAMPLE_FFPROBE = { + "streams": [ + { + "index": 0, + "codec_type": "video", + "codec_name": "hevc", + "profile": "Main 10", + "width": 3840, + "height": 2160, + "pix_fmt": "yuv420p10le", + "bit_rate": "15000000", + "avg_frame_rate": "24/1", + "color_range": "tv", + "color_space": "bt2020nc", + "color_transfer": "smpte2084", + "color_primaries": "bt2020", + "tags": {"language": "eng", "title": "Main"}, + "disposition": {"default": 1}, + "side_data_list": [{"side_data_type": "Mastering display metadata"}], + }, + { + "index": 1, + "codec_type": "audio", + "codec_name": "truehd", + "profile": "TrueHD+Atmos", + "channels": 8, + "channel_layout": "7.1", + "sample_rate": "48000", + "bit_rate": "4500000", + "tags": {"language": "eng", "title": "Atmos"}, + "disposition": {"default": 1, "forced": 0}, + }, + { + "index": 2, + "codec_type": "subtitle", + "codec_name": "subrip", + "codec_long_name": "SubRip subtitle", + "tags": {"language": "eng", "title": "English"}, + "disposition": {"default": 0, "forced": 1, "hearing_impaired": 0}, + }, + ] + } + + def test_video_streams(self): + rows = summarize_video_streams(self.SAMPLE_FFPROBE) + assert len(rows) == 1 + v = rows[0] + assert v["codec"] == "hevc" + assert v["resolution"] == "3840x2160" + assert v["color_transfer"] == "smpte2084" + assert v["language"] == "eng" + assert v["default"] == "yes" + assert "Mastering" in v["side_data"] + + def test_audio_streams(self): + rows = summarize_audio_streams(self.SAMPLE_FFPROBE) + assert len(rows) == 1 + a = rows[0] + assert a["codec"] == "truehd" + assert a["channels"] == 8 + assert a["layout"] == "7.1" + assert a["default"] == "yes" + assert a["forced"] == "" + + def test_subtitle_streams(self): + rows = summarize_subtitle_streams(self.SAMPLE_FFPROBE) + assert len(rows) == 1 + s = rows[0] + assert s["codec"] == "subrip" + assert s["language"] == "eng" + assert s["forced"] == "yes" + assert s["default"] == "" + + def test_summarize_all(self): + rows = summarize_streams(self.SAMPLE_FFPROBE) + assert len(rows) == 3 + types = [r["type"] for r in rows] + assert types == ["video", "audio", "subtitle"] + + +class TestTimestampToLocal: + def test_none(self): + assert timestamp_to_local(None) == "" + + def test_valid(self): + # Just verify it returns a non-empty formatted string + result = timestamp_to_local(1700000000.0) + assert result != "" + assert "-" in result # date format contains dashes