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.
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user