dan/tvh2jellyfin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: TVH2Jellyfin application

Browse Source 
A Docker-based Python application that transfers completed recordings
from TVHeadend to Jellyfin media storage via SCP.

Features:
- Monitors TVHeadend for completed recordings via JSON API
- Transfers files via SCP with SSH key authentication
- Organizes files by show name
- Deletes recordings from TVHeadend after successful transfer
- Tracks transferred files to avoid duplicates
- Dry run mode for testing without making changes
This commit is contained in:
Dan Head
2026-01-07 17:43:09 +00:00
commit a8b247a83f
25 changed files with 2962 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/__init__.py Normal file
View File

117
tests/test_episode_resolver.py Normal file
View File

@@ -0,0 +1,117 @@
"""Tests for episode resolver."""
from unittest.mock import Mock
import pytest
from tvh2jellyfin.episode_resolver import EpisodeResolver
from tvh2jellyfin.tvdb import EpisodeInfo
class TestEpisodeResolver:
"""Tests for EpisodeResolver."""
@pytest.fixture
def mock_episode_info(self):
return EpisodeInfo(
series_id=456,
series_name="Test Show",
season=1,
episode=5,
episode_name="Pilot",
air_date="2024-01-15",
)
def test_lookup_tvdb_success(self, mock_episode_info):
"""Test successful lookup from TVDB."""
mock_tvdb = Mock()
mock_tvdb.lookup_episode.return_value = mock_episode_info
resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=None)
result, source = resolver.lookup("Test Show", 1705312800)
assert result is not None
assert result.season == 1
assert result.episode == 5
assert source == "tvdb"
def test_lookup_tvmaze_fallback(self, mock_episode_info):
"""Test fallback to TVmaze when TVDB fails."""
mock_tvdb = Mock()
mock_tvdb.lookup_episode.return_value = None
mock_tvmaze = Mock()
mock_tvmaze.lookup_episode.return_value = mock_episode_info
resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze)
result, source = resolver.lookup("Test Show", 1705312800)
assert result is not None
assert source == "tvmaze"
def test_lookup_tvdb_exception_fallback(self, mock_episode_info):
"""Test fallback to TVmaze when TVDB raises exception."""
mock_tvdb = Mock()
mock_tvdb.lookup_episode.side_effect = Exception("API Error")
mock_tvmaze = Mock()
mock_tvmaze.lookup_episode.return_value = mock_episode_info
resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze)
result, source = resolver.lookup("Test Show", 1705312800)
assert result is not None
assert source == "tvmaze"
def test_lookup_both_fail(self):
"""Test when both sources fail."""
mock_tvdb = Mock()
mock_tvdb.lookup_episode.return_value = None
mock_tvmaze = Mock()
mock_tvmaze.lookup_episode.return_value = None
resolver = EpisodeResolver(tvdb_client=mock_tvdb, tvmaze_client=mock_tvmaze)
result, source = resolver.lookup("Unknown Show", 1705312800)
assert result is None
assert source == "none"
def test_lookup_no_clients(self):
"""Test when no clients are available."""
resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=None)
result, source = resolver.lookup("Test Show", 1705312800)
assert result is None
assert source == "none"
def test_lookup_tvmaze_only(self, mock_episode_info):
"""Test with only TVmaze client."""
mock_tvmaze = Mock()
mock_tvmaze.lookup_episode.return_value = mock_episode_info
resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=mock_tvmaze)
result, source = resolver.lookup("Test Show", 1705312800)
assert result is not None
assert source == "tvmaze"
def test_is_available_with_tvdb(self):
"""Test is_available with TVDB client."""
resolver = EpisodeResolver(tvdb_client=Mock(), tvmaze_client=None)
assert resolver.is_available() is True
def test_is_available_with_tvmaze(self):
"""Test is_available with TVmaze client."""
resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=Mock())
assert resolver.is_available() is True
def test_is_available_with_both(self):
"""Test is_available with both clients."""
resolver = EpisodeResolver(tvdb_client=Mock(), tvmaze_client=Mock())
assert resolver.is_available() is True
def test_is_available_with_none(self):
"""Test is_available with no clients."""
resolver = EpisodeResolver(tvdb_client=None, tvmaze_client=None)
assert resolver.is_available() is False

150
tests/test_jellyfin.py Normal file
View File

@@ -0,0 +1,150 @@
"""Tests for Jellyfin client."""
from unittest.mock import Mock, patch
import pytest
import requests
from tvh2jellyfin.jellyfin import JellyfinClient, JellyfinConfig
class TestJellyfinConfig:
"""Tests for JellyfinConfig dataclass."""
def test_config_defaults(self):
"""Test default values."""
config = JellyfinConfig(
url="http://localhost:8096",
api_key="test_key",
)
assert config.library_id is None
assert config.enabled is True
def test_config_with_library_id(self):
"""Test config with specific library ID."""
config = JellyfinConfig(
url="http://localhost:8096",
api_key="test_key",
library_id="abc123",
)
assert config.library_id == "abc123"
class TestJellyfinClient:
"""Tests for Jellyfin API client."""
@pytest.fixture
def config(self):
return JellyfinConfig(
url="http://localhost:8096",
api_key="test_api_key",
)
@pytest.fixture
def config_with_library(self):
return JellyfinConfig(
url="http://localhost:8096",
api_key="test_api_key",
library_id="lib123",
)
@pytest.fixture
def client(self, config):
return JellyfinClient(config)
@pytest.fixture
def client_with_library(self, config_with_library):
return JellyfinClient(config_with_library)
def test_url_normalization(self):
"""Test that trailing slashes are removed from URL."""
config = JellyfinConfig(
url="http://localhost:8096/",
api_key="test_key",
)
client = JellyfinClient(config)
assert client.base_url == "http://localhost:8096"
def test_headers_set(self, client):
"""Test that auth headers are set correctly."""
assert client.session.headers["X-MediaBrowser-Token"] == "test_api_key"
assert client.session.headers["Content-Type"] == "application/json"
@patch("requests.Session.get")
def test_test_connection_success(self, mock_get, client):
"""Test successful connection test."""
mock_get.return_value.json.return_value = {
"ServerName": "Test Server",
"Version": "10.8.0",
}
mock_get.return_value.raise_for_status = Mock()
result = client.test_connection()
assert result is True
mock_get.assert_called_once()
assert "/System/Info" in mock_get.call_args[0][0]
@patch("requests.Session.get")
def test_test_connection_failure(self, mock_get, client):
"""Test connection test failure."""
mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused")
result = client.test_connection()
assert result is False
@patch("requests.Session.post")
def test_trigger_library_scan_full(self, mock_post, client):
"""Test triggering full library scan."""
mock_post.return_value.raise_for_status = Mock()
result = client.trigger_library_scan()
assert result is True
mock_post.assert_called_once()
assert "/Library/Refresh" in mock_post.call_args[0][0]
@patch("requests.Session.post")
def test_trigger_library_scan_specific_library(self, mock_post, client_with_library):
"""Test triggering scan for specific library."""
mock_post.return_value.raise_for_status = Mock()
result = client_with_library.trigger_library_scan()
assert result is True
mock_post.assert_called_once()
assert "/Items/lib123/Refresh" in mock_post.call_args[0][0]
@patch("requests.Session.post")
def test_trigger_library_scan_failure(self, mock_post, client):
"""Test library scan trigger failure."""
mock_post.side_effect = requests.exceptions.RequestException("API error")
result = client.trigger_library_scan()
assert result is False
@patch("requests.Session.get")
def test_get_libraries_success(self, mock_get, client):
"""Test getting list of libraries."""
mock_get.return_value.json.return_value = [
{"Name": "TV Shows", "ItemId": "tv123"},
{"Name": "Movies", "ItemId": "mov456"},
]
mock_get.return_value.raise_for_status = Mock()
result = client.get_libraries()
assert len(result) == 2
assert result[0] == {"name": "TV Shows", "id": "tv123"}
assert result[1] == {"name": "Movies", "id": "mov456"}
@patch("requests.Session.get")
def test_get_libraries_failure(self, mock_get, client):
"""Test getting libraries when API fails."""
mock_get.side_effect = requests.exceptions.RequestException("API error")
result = client.get_libraries()
assert result == []

216
tests/test_organizer.py Normal file
View File

@@ -0,0 +1,216 @@
"""Tests for organizer module."""
import pytest
from tvh2jellyfin.organizer import (
generate_destination_path,
generate_destination_path_with_episode,
sanitize_directory_name,
sanitize_filename,
)
from tvh2jellyfin.tvdb import EpisodeInfo
from tvh2jellyfin.tvheadend import Recording
class TestSanitizeFilename:
"""Tests for filename sanitization."""
def test_removes_invalid_characters(self):
assert sanitize_filename('Test: "Show"') == "Test Show"
# Multiple spaces get normalized to single space
assert sanitize_filename("Show <1> | Part 2") == "Show 1 Part 2"
def test_normalizes_spaces(self):
assert sanitize_filename("Test Show") == "Test Show"
assert sanitize_filename("Test_Show") == "Test Show"
def test_strips_whitespace(self):
assert sanitize_filename(" Test Show ") == "Test Show"
def test_empty_returns_unknown(self):
assert sanitize_filename("") == "Unknown"
assert sanitize_filename(" ") == "Unknown"
class TestSanitizeDirectoryName:
"""Tests for directory name sanitization."""
def test_removes_trailing_dots(self):
assert sanitize_directory_name("Test Show.") == "Test Show"
assert sanitize_directory_name("Test Show...") == "Test Show"
def test_handles_empty(self):
assert sanitize_directory_name("") == "Unknown"
class TestGenerateDestinationPath:
"""Tests for destination path generation."""
@pytest.fixture
def recording_with_subtitle(self):
return Recording(
uuid="123",
title="Test Show",
subtitle="Episode Title",
filename="/recordings/test.ts",
filesize=1000000,
channel="BBC One",
start=1705312800,
stop=1705316400,
status="Completed OK",
)
@pytest.fixture
def recording_without_subtitle(self):
return Recording(
uuid="456",
title="Another Show",
subtitle="",
filename="/recordings/another.ts",
filesize=2000000,
channel="ITV",
start=1705312800,
stop=1705316400,
status="Completed OK",
)
def test_with_subtitle(self, recording_with_subtitle):
dest_dir, filename = generate_destination_path(
recording_with_subtitle, "/media/tv"
)
assert dest_dir == "/media/tv/Test Show"
assert filename == "Test Show - Episode Title.ts"
def test_without_subtitle(self, recording_without_subtitle):
dest_dir, filename = generate_destination_path(
recording_without_subtitle, "/media/tv"
)
assert dest_dir == "/media/tv/Another Show"
# Should include timestamp
assert filename.startswith("Another Show - ")
assert filename.endswith(".ts")
class TestGenerateDestinationPathWithEpisode:
"""Tests for destination path generation with episode info."""
@pytest.fixture
def recording(self):
return Recording(
uuid="123",
title="Coronation Street",
subtitle="",
filename="/recordings/Coronation-Street/Coronation Street2026-01-06.ts",
filesize=1000000,
channel="ITV",
start=1767744000, # 2026-01-06
stop=1767747600,
status="Completed OK",
)
@pytest.fixture
def episode_info(self):
return EpisodeInfo(
series_id=456,
series_name="Coronation Street",
season=67,
episode=4,
episode_name="Episode 4",
air_date="2026-01-06",
)
def test_with_episode_info(self, recording, episode_info):
"""Test Jellyfin naming format with episode info."""
dest_dir, filename = generate_destination_path_with_episode(
recording, "/media/tv", episode_info
)
assert dest_dir == "/media/tv/Coronation Street/Season 67"
assert filename == "Coronation Street - S67E04.ts"
def test_without_episode_info(self, recording):
"""Test fallback when episode info not available."""
dest_dir, filename = generate_destination_path_with_episode(
recording, "/media/tv", None
)
assert dest_dir == "/media/tv/Coronation Street"
# Should use fallback logic (timestamp)
assert filename.startswith("Coronation Street - ")
assert filename.endswith(".ts")
def test_uses_canonical_series_name(self, recording):
"""Test that canonical series name from TVDB is used."""
episode_info = EpisodeInfo(
series_id=456,
series_name="Doctor Who (2005)", # TVDB canonical name
season=14,
episode=1,
episode_name="The Church on Ruby Road",
air_date="2023-12-25",
)
# Recording might have different name
recording_with_diff_name = Recording(
uuid="789",
title="Doctor Who", # TVHeadend name
subtitle="",
filename="/recordings/doctor-who.ts",
filesize=3000000,
channel="BBC One",
start=1703512800,
stop=1703516400,
status="Completed OK",
)
dest_dir, filename = generate_destination_path_with_episode(
recording_with_diff_name, "/media/tv", episode_info
)
# Should use TVDB canonical name with season folder
assert dest_dir == "/media/tv/Doctor Who (2005)/Season 14"
assert filename == "Doctor Who (2005) - S14E01.ts"
def test_large_episode_numbers(self, recording):
"""Test with season/episode numbers > 99."""
episode_info = EpisodeInfo(
series_id=456,
series_name="Long Running Show",
season=67,
episode=125,
episode_name="Episode",
air_date="2026-01-06",
)
dest_dir, filename = generate_destination_path_with_episode(
recording, "/media/tv", episode_info
)
assert filename == "Long Running Show - S67E125.ts"
def test_preserves_file_extension(self, recording):
"""Test that file extension is preserved."""
recording_mkv = Recording(
uuid="123",
title="Test Show",
subtitle="",
filename="/recordings/test.mkv",
filesize=1000000,
channel="BBC",
start=1705312800,
stop=1705316400,
status="Completed OK",
)
episode_info = EpisodeInfo(
series_id=456,
series_name="Test Show",
season=1,
episode=1,
episode_name="Pilot",
air_date="2024-01-15",
)
_, filename = generate_destination_path_with_episode(
recording_mkv, "/media/tv", episode_info
)
assert filename.endswith(".mkv")

190
tests/test_tvdb.py Normal file
View File

@@ -0,0 +1,190 @@
"""Tests for TheTVDB client."""
from unittest.mock import Mock, patch
import pytest
from tvh2jellyfin.tvdb import EpisodeInfo, TVDBClient, TVDBConfig
class TestEpisodeInfo:
"""Tests for EpisodeInfo dataclass."""
def test_formatted_episode_single_digit(self):
"""Test SxxExx formatting with single digit numbers."""
info = EpisodeInfo(
series_id=123,
series_name="Test Show",
season=1,
episode=5,
episode_name="Pilot",
air_date="2024-01-15",
)
assert info.formatted_episode == "S01E05"
def test_formatted_episode_double_digit(self):
"""Test SxxExx formatting with double digit numbers."""
info = EpisodeInfo(
series_id=123,
series_name="Test Show",
season=12,
episode=25,
episode_name="Episode",
air_date="2024-01-15",
)
assert info.formatted_episode == "S12E25"
def test_formatted_episode_large_numbers(self):
"""Test formatting with season/episode > 99."""
info = EpisodeInfo(
series_id=123,
series_name="Long Running Show",
season=67,
episode=125,
episode_name="Episode",
air_date="2024-01-15",
)
assert info.formatted_episode == "S67E125"
class TestTVDBClient:
"""Tests for TVDB API client."""
@pytest.fixture
def config(self):
return TVDBConfig(
api_key="test_key",
pin="",
cache_ttl=3600,
show_mapping_file=None,
enabled=True,
)
@pytest.fixture
def client(self, config):
with patch.object(TVDBClient, "_authenticate"):
return TVDBClient(config)
def test_cache_valid(self, client):
"""Test cache validity check."""
import time
# Recent timestamp should be valid
assert client._is_cache_valid(time.time() - 100)
# Old timestamp should be invalid
assert not client._is_cache_valid(time.time() - 7200)
@patch("requests.Session.post")
def test_authenticate(self, mock_post, config):
"""Test authentication flow."""
mock_post.return_value.json.return_value = {"data": {"token": "test_token"}}
mock_post.return_value.raise_for_status = Mock()
client = TVDBClient(config)
client._authenticate()
assert client._token == "test_token"
mock_post.assert_called_once()
@patch("requests.Session.get")
@patch("requests.Session.post")
def test_search_series_success(self, mock_post, mock_get, config):
"""Test successful series search."""
mock_post.return_value.json.return_value = {"data": {"token": "test_token"}}
mock_post.return_value.raise_for_status = Mock()
mock_get.return_value.json.return_value = {
"data": [{"tvdb_id": 456, "name": "The Test Show"}]
}
mock_get.return_value.raise_for_status = Mock()
client = TVDBClient(config)
result = client.search_series("Test Show")
assert result is not None
assert result[0] == 456
assert result[1] == "The Test Show"
@patch("requests.Session.get")
@patch("requests.Session.post")
def test_search_series_not_found(self, mock_post, mock_get, config):
"""Test series search with no results."""
mock_post.return_value.json.return_value = {"data": {"token": "test_token"}}
mock_post.return_value.raise_for_status = Mock()
mock_get.return_value.json.return_value = {"data": []}
mock_get.return_value.raise_for_status = Mock()
client = TVDBClient(config)
result = client.search_series("Nonexistent Show")
assert result is None
@patch("tvh2jellyfin.tvdb.requests.Session")
def test_lookup_episode_success(self, mock_session_class, config):
"""Test successful episode lookup."""
from datetime import datetime
mock_session = Mock()
mock_session_class.return_value = mock_session
# Use a timestamp and derive the expected air date from it
timestamp = 1767744000
expected_air_date = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d")
# Mock authentication
mock_session.post.return_value.json.return_value = {"data": {"token": "test_token"}}
mock_session.post.return_value.raise_for_status = Mock()
# Mock search and episode responses
mock_session.get.return_value.raise_for_status = Mock()
mock_session.get.return_value.json.side_effect = [
{"data": [{"tvdb_id": 456, "name": "Coronation Street"}]},
{
"data": {
"name": "Coronation Street",
"episodes": [
{
"seasonNumber": 67,
"number": 4,
"name": "Episode 4",
"aired": expected_air_date,
}
],
},
"links": {},
},
]
client = TVDBClient(config)
result = client.lookup_episode("Coronation Street", timestamp)
assert result is not None
assert result.season == 67
assert result.episode == 4
assert result.formatted_episode == "S67E04"
def test_show_mapping_file(self, tmp_path, config):
"""Test show name mapping from file."""
mapping_file = tmp_path / "mapping.json"
mapping_file.write_text('{"Doctor Who": "Doctor Who (2005)"}')
config.show_mapping_file = str(mapping_file)
with patch.object(TVDBClient, "_authenticate"):
client = TVDBClient(config)
assert client._show_mapping.get("Doctor Who") == "Doctor Who (2005)"
def test_caching_prevents_duplicate_calls(self, client):
"""Test that cache prevents duplicate API calls."""
import time
# Manually populate cache
client._series_cache["test show"] = (123, "Test Show", time.time())
# Search should use cache, not make API call
result = client.search_series("test show")
assert result == (123, "Test Show")

121
tests/test_tvmaze.py Normal file
View File

@@ -0,0 +1,121 @@
"""Tests for TVmaze client."""
from unittest.mock import Mock, patch
import pytest
from tvh2jellyfin.tvmaze import TVMazeClient, TVMazeConfig
class TestTVMazeClient:
"""Tests for TVmaze API client."""
@pytest.fixture
def config(self):
return TVMazeConfig(cache_ttl=3600, enabled=True)
@pytest.fixture
def client(self, config):
return TVMazeClient(config)
@patch("requests.Session.get")
def test_search_series_success(self, mock_get, client):
"""Test successful series search."""
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"id": 789,
"name": "Coronation Street",
}
mock_get.return_value.raise_for_status = Mock()
result = client.search_series("Coronation Street")
assert result is not None
assert result[0] == 789
assert result[1] == "Coronation Street"
@patch("requests.Session.get")
def test_search_series_not_found(self, mock_get, client):
"""Test series search with 404 response."""
mock_get.return_value.status_code = 404
result = client.search_series("Nonexistent Show")
assert result is None
@patch("requests.Session.get")
def test_get_episode_by_date_success(self, mock_get, client):
"""Test successful episode lookup by date."""
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = [
{"season": 67, "number": 4, "name": "Episode 4"}
]
mock_get.return_value.raise_for_status = Mock()
result = client.get_episode_by_date(789, "2026-01-06", "Coronation Street")
assert result is not None
assert result.season == 67
assert result.episode == 4
assert result.series_name == "Coronation Street"
@patch("requests.Session.get")
def test_get_episode_by_date_not_found(self, mock_get, client):
"""Test episode lookup with no results."""
mock_get.return_value.status_code = 404
result = client.get_episode_by_date(789, "2026-01-01", "Coronation Street")
assert result is None
@patch("requests.Session.get")
def test_lookup_episode_full_flow(self, mock_get, client):
"""Test full episode lookup flow."""
# Set up responses for search and episode lookup
mock_get.return_value.status_code = 200
mock_get.return_value.json.side_effect = [
{"id": 789, "name": "Coronation Street"}, # Search response
[{"season": 67, "number": 4, "name": "Episode 4"}], # Episode response
]
mock_get.return_value.raise_for_status = Mock()
# Timestamp for 2026-01-06
result = client.lookup_episode("Coronation Street", 1767744000)
assert result is not None
assert result.series_name == "Coronation Street"
assert result.formatted_episode == "S67E04"
def test_caching_series(self, client):
"""Test that series cache prevents duplicate calls."""
import time
# Manually populate cache
client._series_cache["coronation street"] = (789, "Coronation Street", time.time())
# Should use cache
result = client.search_series("coronation street")
assert result == (789, "Coronation Street")
def test_caching_episodes(self, client):
"""Test that episode cache prevents duplicate calls."""
import time
from tvh2jellyfin.tvdb import EpisodeInfo
cached_episode = EpisodeInfo(
series_id=789,
series_name="Coronation Street",
season=67,
episode=4,
episode_name="Episode 4",
air_date="2026-01-06",
)
client._episode_cache["789:2026-01-06"] = (cached_episode, time.time())
# Should use cache
result = client.get_episode_by_date(789, "2026-01-06", "Coronation Street")
assert result is not None
assert result.season == 67