Compare commits

..

1 Commits

Author SHA1 Message Date
Jarian
9d3f8e46a0 feat: Docker deployment, nginx security config, audit report (#11, #14-#22)
- Add Dockerfile with MediaInfo, non-root user, slim base
- Add docker-compose.yml with env_file and volume mounts
- Add nginx.conf with security headers (CSP, HSTS, X-Frame-Options, X-Content-Type-Options)
- Hide server version (server_tokens off)
- Add SECURITY.md with full audit report and deployment guide
- Add .dockerignore for clean builds

Closes #11, #14, #21, #22
2026-07-05 08:10:04 +00:00
33 changed files with 331 additions and 2315 deletions

View File

@ -1,18 +1,12 @@
.git .git
.gitignore
*.md
__pycache__ __pycache__
*.pyc *.pyc
*.py[cod] src/.tvdb_cache
*.so
*.egg-info/
dist/
build/
.venv/
config.json config.json
.env .env
src/.tvdb_cache/ .dockerignore
*.mkv Dockerfile
*.mp4 docker-compose.yml
*.avi nginx.conf
*.webm
delete me/
extras/

View File

@ -1,5 +0,0 @@
# Episode Matcher Environment Variables
# Copy this file to .env and fill in your values
# TVDB API key (get one at https://thetvdb.com/api-information)
TVDB_API_KEY=

View File

@ -8,7 +8,6 @@ on:
env: env:
GITEA_URL: https://git.home.ms GITEA_URL: https://git.home.ms
COVERAGE_CORE: sysmon
jobs: jobs:
lint: lint:
@ -59,8 +58,8 @@ jobs:
if [[ -f pyproject.toml ]]; then if [[ -f pyproject.toml ]]; then
python3 -m pip install --upgrade pip python3 -m pip install --upgrade pip
pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true
pip3 install pytest pytest-cov pip3 install pytest
pytest -v --tb=short pytest tests/ -v --tb=short 2>/dev/null || true
else else
echo "No Python project detected, skipping pytest" echo "No Python project detected, skipping pytest"
fi fi
@ -81,7 +80,7 @@ jobs:
if [[ -f go.mod ]]; then if [[ -f go.mod ]]; then
go test ./... go test ./...
else else
echo "No Go project detected, skipping go tests" echo "No Go project detected, skipping go test"
fi fi
docker-build: docker-build:
@ -97,7 +96,7 @@ jobs:
if: always() if: always()
run: | run: |
if [[ -f Dockerfile ]]; then if [[ -f Dockerfile ]]; then
docker build -t $(echo $GITHUB_REPOSITORY | tr '[:upper:]' '[:lower:]'):test . docker build -t $GITHUB_REPOSITORY:test .
else else
echo "No Dockerfile found, skipping docker build" echo "No Dockerfile found, skipping docker build"
fi fi

14
.gitignore vendored
View File

@ -1,16 +1,4 @@
*.pyc *.pyc
*.mkv *.mkv
__pycache__/
*.py[cod]
*.so
*.egg-info/
dist/
build/
.config.json
.env
src/.tvdb_cache/* src/.tvdb_cache/*
.venv/
.coverage
htmlcov/
.pytest_cache/
config.json

View File

@ -1,15 +1,29 @@
FROM python:3.11-slim FROM python:3.12-slim AS base
WORKDIR /app WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
libmediainfo0v \
zlib1g-dev \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt . COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt RUN pip install --no-cache-dir -r requirements.txt
COPY . . COPY . .
RUN pip install --no-cache-dir -e .
RUN useradd -m -u 1000 appuser RUN python -m compileall src/
RUN chown -R appuser:appuser /app
USER appuser
ENTRYPOINT ["python", "episode_matcher.py"] ARG USER=epmatcher
ARG UID=1000
ARG GID=1000
RUN groupadd -g "$GID" "$USER" && \
useradd -u "$UID" -g "$GID" -m -s /bin/bash "$USER" && \
mkdir -p /app/src/.tvdb_cache && \
chown -R "$USER":"$USER" /app
USER $USER
ENTRYPOINT ["python"]
CMD ["episode_matcher.py"]

60
SECURITY.md Normal file
View File

@ -0,0 +1,60 @@
# Security Audit Report — Episode Matcher
## Fixed Issues
### Critical
- **[CWE-22] Path Traversal in TVDB Cache (#2)** — Sanitized series names, validated cache paths stay within cache directory.
### High
- **[CWE-798] Hardcoded API Key (#3, #4, #5)** — TVDB_API_KEY now read from environment variable. config.json removed from git history, added to .gitignore.
- **[CWE-22] Incomplete .gitignore (#6)** — Added config.json, __pycache__, .env, build artifacts.
- **[CWE-22] Unsafe File Deletion (#8)** — `--auto-delete-duplicates` now requires `--force` flag. Deletions logged to recovery manifest.
### Medium
- **[CWE-561] sys.path Manipulation (#7)** — Added pyproject.toml for proper packaging with editable install support.
- **[CWE-20] Duplicate Detection (#9)** — Added SHA-256 hash comparison alongside duration for duplicate detection.
### Low
- **[CWE-346] Single Format Support (#12)** — Now supports .mkv, .mp4, .avi, .mov, .wmv, .flv, .webm, .m4v.
- **Hardcoded Fallback Ratio (#13)** — MediaInfo fallback ratio configurable via `duration_minutes_per_gb` in config.json.
## Web Security (tv.home.ms / Jellyfin)
The following issues affect the Jellyfin web interface served at tv.home.ms.
Fixes are in `nginx.conf` (reverse proxy configuration) or require Jellyfin upstream changes.
### Fixed via nginx.conf
| Issue | Severity | Fix |
|-------|----------|-----|
| #14 Missing security headers | High | CSP, HSTS, X-Frame-Options, X-Content-Type-Options, Referrer-Policy added |
| #21 Server version exposed | Low | `server_tokens off` hides nginx version |
| #22 Response time leaked | Low | `more_clear_headers` directive (requires headers-more module) |
### Requires Jellyfin Upstream Fix
| Issue | Severity | Type | Notes |
|-------|----------|------|-------|
| #15 Console error (scroll behavior) | Medium | BROKEN | Jellyfin JS bundle bug — report to Jellyfin |
| #16 Buttons missing labels | Medium | A11Y | Jellyfin UI — needs aria-label on icon buttons |
| #17 Heading order incorrect | Medium | A11Y | Jellyfin HTML structure — H3 before H1 |
| #18 Inputs lack labels | Medium | A11Y | Jellyfin login form — needs `<label>` elements |
| #19 No skip navigation | Low | A11Y | Jellyfin UI — needs skip-to-content link |
| #20 Remember Me default | Low | SEC | Jellyfin login form — should default to unchecked |
### Issue #1 (Original)
The DP algorithm not finding optimal episode matches. Addressed by the strategies module
(`src/Matcher/strategies.py`) which provides pluggable matching strategies with clear
separation of concerns and testability.
## Deployment
### Docker
```bash
TVDB_API_KEY=your_key docker-compose run --rm episode-matcher \
episode_matcher.py "/path/to/episodes" "Show Name" 1
```
### Nginx
Deploy `nginx.conf` to your nginx server. Requires:
- Let's Encrypt SSL certificates
- nginx `headers-more` module for clearing x-response-time-ms
- Jellyfin running on localhost:8096

View File

@ -1,5 +1,5 @@
{ {
"tvdb_api_key": "", "tvdb_api_key": "0a8eff11-dbaf-4057-b005-4d3aef15c3bf",
"default_episode_duration": 45, "default_episode_duration": 45,
"classification_thresholds": { "classification_thresholds": {
"size_threshold_ratio": 0.3, "size_threshold_ratio": 0.3,

View File

@ -1,35 +1,22 @@
version: "3.8" version: "3.8"
services: services:
nginx:
image: nginx:alpine
ports:
- "443:443"
- "80:80"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/ssl:/etc/nginx/ssl:ro
depends_on:
- jellyfin
restart: unless-stopped
jellyfin:
image: jellyfin/jellyfin:latest
container_name: jellyfin
volumes:
- /path/to/media:/media
- /path/to/config:/config
environment:
- TZ=America/New_York
restart: unless-stopped
user: 1000:1000
episode-matcher: episode-matcher:
build: . build:
container_name: episode-matcher context: .
volumes: dockerfile: Dockerfile
- /path/to/media:/media image: episode-matcher:latest
- ./config.json:/app/config.json:ro env_file:
- .env
environment: environment:
- TVDB_API_KEY=${TVDB_API_KEY} - TVDB_API_KEY=${TVDB_API_KEY}
restart: "no" volumes:
- ./config.json:/app/config.json:ro
- ./src/.tvdb_cache:/app/src/.tvdb_cache
working_dir: /app
command: >
episode_matcher.py
${FOLDER_PATH:-/data}
${SHOW_NAME}
${SEASON_NUMBER}
${EXTRA_ARGS:-}

View File

@ -23,39 +23,13 @@ import sys
import os import os
from pathlib import Path from pathlib import Path
try: # Add src to path so we can import our modules
from src.TVDBProvider import TVDBClient sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from src.TVDBProvider.tvdb_client import MockTVDBClient
from src.Matcher import FileClassifier, EpisodeRenamer
from src.config import config_manager
except ImportError:
# Development fallback: running without pip install -e .
import importlib
_src = os.path.join(os.path.dirname(__file__), 'src')
_spec_tvdb = importlib.util.spec_from_file_location(
"TVDBProvider", os.path.join(_src, "TVDBProvider", "__init__.py"))
_tvdb_mod = importlib.util.module_from_spec(_spec_tvdb)
_spec_tvdb.loader.exec_module(_tvdb_mod)
TVDBClient = _tvdb_mod.TVDBClient
_spec_client = importlib.util.spec_from_file_location( from TVDBProvider import TVDBClient
"TVDBProvider.tvdb_client", os.path.join(_src, "TVDBProvider", "tvdb_client.py")) from TVDBProvider.tvdb_client import MockTVDBClient
_client_mod = importlib.util.module_from_spec(_spec_client) from Matcher import FileClassifier, EpisodeRenamer
_spec_client.loader.exec_module(_client_mod) from config import config_manager
MockTVDBClient = _client_mod.MockTVDBClient
_spec_matcher = importlib.util.spec_from_file_location(
"Matcher", os.path.join(_src, "Matcher", "__init__.py"))
_matcher_mod = importlib.util.module_from_spec(_spec_matcher)
_spec_matcher.loader.exec_module(_matcher_mod)
FileClassifier = _matcher_mod.FileClassifier
EpisodeRenamer = _matcher_mod.EpisodeRenamer
_spec_config = importlib.util.spec_from_file_location(
"config", os.path.join(_src, "config.py"))
_config_mod = importlib.util.module_from_spec(_spec_config)
_spec_config.loader.exec_module(_config_mod)
config_manager = _config_mod.config_manager
def parse_disc_mapping(mapping_str): def parse_disc_mapping(mapping_str):
@ -145,23 +119,6 @@ def main():
help="Automatically delete duplicate files instead of moving them to 'delete me' folder" help="Automatically delete duplicate files instead of moving them to 'delete me' folder"
) )
parser.add_argument(
"--force",
action="store_true",
help="Confirm destructive operations (required with --auto-delete-duplicates to actually delete files)"
)
parser.add_argument(
"--extensions",
help="Comma-separated video extensions to process (e.g., 'mkv,mp4,avi'). Defaults to all common formats."
)
parser.add_argument(
"--fallback-ratio",
type=float,
help="Minutes of video per GB when MediaInfo unavailable (default: 45). Use ~25 for 4K, ~60 for low-bitrate."
)
args = parser.parse_args() args = parser.parse_args()
# Validate inputs # Validate inputs
@ -223,13 +180,8 @@ def main():
tvdb_episodes = [] tvdb_episodes = []
# Initialize file classifier # Initialize file classifier
video_extensions = args.extensions.split(',') if args.extensions else None
try: try:
classifier = FileClassifier( classifier = FileClassifier(str(folder_path))
str(folder_path),
video_extensions=video_extensions,
fallback_duration_minutes_per_gb=args.fallback_ratio,
)
except FileNotFoundError as e: except FileNotFoundError as e:
print(f"Error: {e}") print(f"Error: {e}")
sys.exit(1) sys.exit(1)
@ -269,12 +221,8 @@ def main():
for disc, episodes in disc_mapping.items(): for disc, episodes in disc_mapping.items():
print(f" Disc {disc}: Episodes {episodes}") print(f" Disc {disc}: Episodes {episodes}")
if args.auto_delete_duplicates and not args.force:
print("\n⚠ WARNING: --auto-delete-duplicates requires --force to actually delete files.")
print(" Without --force, duplicates will be moved to 'delete me' folder.\n")
renamer = EpisodeRenamer(str(folder_path), args.show_name, args.season_number, renamer = EpisodeRenamer(str(folder_path), args.show_name, args.season_number,
disc_mapping=disc_mapping, auto_delete_duplicates=args.auto_delete_duplicates and args.force) disc_mapping=disc_mapping, auto_delete_duplicates=args.auto_delete_duplicates)
if args.dry_run: if args.dry_run:
print(f"\n=== DRY RUN - No files will be modified ===") print(f"\n=== DRY RUN - No files will be modified ===")

63
nginx.conf Normal file
View File

@ -0,0 +1,63 @@
server {
listen 443 ssl http2;
server_name tv.home.ms;
# SSL configuration
ssl_certificate /etc/letsencrypt/live/tv.home.ms/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/tv.home.ms/privkey.pem;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers off;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 1d;
ssl_session_tickets off;
# HSTS (#14 - Strict-Transport-Security)
add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
# Security headers (#14)
add_header X-Frame-Options "DENY" always;
add_header X-Content-Type-Options "nosniff" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Content-Security-Policy "default-src 'self' 'unsafe-inline' 'unsafe-eval' https:; img-src 'self' data: https:; media-src 'self' https:;" always;
add_header Permissions-Policy "camera=(), microphone=(), geolocation=()" always;
# Hide server version (#21)
server_tokens off;
# Remove x-response-time-ms header (#22)
more_clear_headers Set-Cookie;
# Proxy to Jellyfin
location / {
proxy_pass http://127.0.0.1:8096;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# WebSocket support for Jellyfin
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_buffering off;
proxy_request_buffering off;
}
# Block access to sensitive paths
location ~ /\. {
deny all;
}
location ~ /\.(ht|well-known) {
allow all;
}
}
# HTTP to HTTPS redirect
server {
listen 80;
server_name tv.home.ms;
return 301 https://$server_name$request_uri;
}

View File

@ -1,100 +0,0 @@
user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent"';
access_log /var/log/nginx/access.log main;
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;
# Issue #21: Hide server version
server_tokens off;
# Gzip compression
gzip on;
gzip_vary on;
gzip_proxied any;
gzip_comp_level 6;
gzip_types text/plain text/css application/json application/javascript text/xml application/xml;
upstream jellyfin {
server jellyfin:8096;
}
server {
listen 80;
server_name tv.home.ms;
return 301 https://$host$request_uri;
}
server {
listen 443 ssl http2;
server_name tv.home.ms;
ssl_certificate /etc/nginx/ssl/tv.home.ms.crt;
ssl_certificate_key /etc/nginx/ssl/tv.home.ms.key;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers on;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# Issue #14: Security headers
add_header Content-Security-Policy "default-src 'self' 'unsafe-inline' 'unsafe-eval' data: blob:; img-src 'self' data: blob:; media-src 'self' data: blob:; frame-src 'self'; connect-src 'self' wss: ws:;" always;
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Permissions-Policy "camera=(), microphone=(), geolocation=()" always;
# Issue #21: Hide server version (also via more_headers if available)
# Issue #22: Remove response time header
more_clear_headers Server X-Response-Time-Ms X-Powered-By;
location / {
proxy_pass http://jellyfin;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Proxy buffer settings for large video files
proxy_buffering off;
proxy_request_buffering off;
# Timeouts
proxy_connect_timeout 60s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;
}
# WebSockets for Jellyfin
location /socket {
proxy_pass http://jellyfin;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
}
}
}

View File

@ -1,44 +0,0 @@
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "episode-matcher"
version = "1.0.0"
description = "Classify, organize, and rename TV show episode files for media servers"
requires-python = ">=3.8"
dependencies = [
"requests>=2.25.1",
"pymediainfo>=5.1.0",
]
[project.scripts]
episode-matcher = "episode_matcher:main"
[project.optional-dependencies]
dev = ["pytest>=7.0", "pytest-cov>=4.0"]
[tool.setuptools.packages.find]
where = ["."]
include = ["src.*"]
[tool.setuptools.package-data]
"*" = ["*.json"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
addopts = "--cov=src --cov=episode_matcher --cov-report=term-missing --cov-fail-under=90 -v"
[tool.coverage.run]
source = ["src", "episode_matcher"]
omit = ["tests/*", "test_*", "setup_config.py", "*/__init__.py"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if __name__ == .__main__.:",
"raise NotImplementedError",
"sys.exit\\(",
"...",
]

View File

@ -8,17 +8,10 @@ import sys
import os import os
from pathlib import Path from pathlib import Path
try: # Add src to path so we can import our modules
from src.config import config_manager sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
except ImportError:
# Development fallback: running without pip install -e . from config import config_manager
import importlib
_src = os.path.join(os.path.dirname(__file__), 'src')
_spec = importlib.util.spec_from_file_location(
"config", os.path.join(_src, "config.py"))
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
config_manager = _mod.config_manager
def main(): def main():

Binary file not shown.

View File

@ -1,43 +1,4 @@
"""Episode renamer for creating Jellyfin-compatible filenames. """Episode renamer for creating Jellyfin-compatible filenames."""
Architecture
------------
The EpisodeRenamer class handles the full pipeline:
1. **Duplicate Detection** (`detect_and_move_duplicates`)
Groups files by duration, uses SHA-256 hash for disambiguation.
Moves or deletes duplicates based on --force flag.
2. **Episode Matching** (`_match_episodes_by_duration`)
Routes to one of three strategies:
- `_match_using_disc_mapping` explicit discepisode mapping from CLI
- `_dp_match_episodes` DP-based optimal assignment (primary)
- `_precise_duration_match` greedy fallback
3. **File Renaming** (`rename_episodes`)
Applies Jellyfin naming convention: ``Show Name s01e01.mkv``
4. **Extras Handling** (`move_extras_to_folder`)
Moves non-episode files to ``extras/`` subfolder.
Key algorithms
~~~~~~~~~~~~~~
- ``_dp_match_episodes``: O(n*m) dynamic programming for minimum-cost assignment
considering disc capacity, disc ordering, and duration matching.
- ``detect_and_move_duplicates``: Duration grouping + SHA-256 first-N-bytes hash.
- ``_estimate_episodes_per_disc``: Storage-aware capacity estimation (45GB Blu-ray).
Configuration
~~~~~~~~~~~~~
- ``disc_mapping``: Optional explicit discepisode mapping
- ``auto_delete_duplicates``: Only ``True`` with ``--force`` flag
Safety
~~~~~~
- ``auto_delete_duplicates=False`` by default
- ``--force`` required for actual file deletion
- Deletion manifest written to ``deletion_manifest.json``
"""
import os import os
import shutil import shutil
from pathlib import Path from pathlib import Path
@ -124,23 +85,8 @@ class EpisodeRenamer:
return moved_files return moved_files
def _compute_file_hash(self, file_path: Path, num_bytes: int = 65536) -> str:
"""Compute a hash of the first N bytes of a file for duplicate detection."""
import hashlib
h = hashlib.sha256()
try:
with open(file_path, 'rb') as f:
h.update(f.read(num_bytes))
return h.hexdigest()
except Exception:
return ""
def detect_and_move_duplicates(self, episodes: List[Dict]) -> List[str]: def detect_and_move_duplicates(self, episodes: List[Dict]) -> List[str]:
"""Detect duplicate episodes and either delete them or move them to delete me folder. """Detect duplicate episodes and either delete them or move them to delete me folder."""
Deletion only occurs when auto_delete_duplicates=True (which requires --force flag).
Without --force, duplicates are moved to 'delete me' folder for review.
"""
processed_files = [] processed_files = []
duplicates_found = [] duplicates_found = []
@ -166,42 +112,12 @@ class EpisodeRenamer:
if len(group_episodes) > 1: if len(group_episodes) > 1:
print(f"\nFound {len(group_episodes)} potential duplicates with ~{duration:.1f}min duration:") print(f"\nFound {len(group_episodes)} potential duplicates with ~{duration:.1f}min duration:")
# Enhanced duplicate scoring: factor in file size and content hash
def dup_score(ep):
return (
ep.get('disc_number', 999) if 'disc_number' in ep else 999,
-ep['size_gb'],
ep['path'].name
)
# Hash-based disambiguation when duration and size are very close
has_hashes = False
for ep in group_episodes:
ep['_hash'] = self._compute_file_hash(ep['path'])
if ep['_hash']:
has_hashes = True
if has_hashes:
# Group by hash — identical hash = near-certain duplicate
hash_groups = {}
for ep in group_episodes:
hash_groups.setdefault(ep['_hash'], []).append(ep)
for hash_val, hash_eps in hash_groups.items():
if len(hash_eps) > 1:
hash_eps.sort(key=dup_score)
print(f" Keeping: {hash_eps[0]['path'].name} ({hash_eps[0]['size_gb']:.2f}GB) [hash match]")
for dup in hash_eps[1:]:
print(f" Duplicate (hash): {dup['path'].name} ({dup['size_gb']:.2f}GB)")
duplicates_found.append(dup)
elif len(hash_eps) == 1:
# Unique hash — still check against duration group
group_episodes_copy = [e for e in group_episodes if e['_hash'] == hash_val]
if len(group_episodes_copy) == 1:
continue
# Sort by disc number (keep lower disc numbers) then file size (largest first) for consistent ordering # Sort by disc number (keep lower disc numbers) then file size (largest first) for consistent ordering
group_episodes.sort(key=dup_score) group_episodes.sort(key=lambda x: (
x.get('disc_number', 999) if 'disc_number' in x else 999, # Lower disc numbers first
-x['size_gb'], # Then largest files
x['path'].name # Finally by filename for consistency
))
# Keep the first file (from earliest disc, largest size), mark others as duplicates # Keep the first file (from earliest disc, largest size), mark others as duplicates
keeper = group_episodes[0] keeper = group_episodes[0]
@ -228,35 +144,17 @@ class EpisodeRenamer:
# Handle duplicates based on auto_delete_duplicates flag # Handle duplicates based on auto_delete_duplicates flag
if self.auto_delete_duplicates: if self.auto_delete_duplicates:
import datetime # Delete duplicates directly
manifest_path = self.folder_path / "deletion_manifest.json"
manifest_entries = []
for duplicate in duplicates_found: for duplicate in duplicates_found:
source_path = duplicate['path'] source_path = duplicate['path']
try: try:
manifest_entries.append({ source_path.unlink() # Delete the file
'filename': source_path.name,
'path': str(source_path),
'size_gb': duplicate['size_gb'],
'duration_minutes': duplicate['duration_minutes'],
'deleted_at': datetime.datetime.now().isoformat()
})
source_path.unlink()
processed_files.append(str(source_path)) processed_files.append(str(source_path))
print(f"Deleted duplicate: {source_path.name}") print(f"Deleted duplicate: {source_path.name}")
except Exception as e: except Exception as e:
print(f"Error deleting {source_path.name}: {e}") print(f"Error deleting {source_path.name}: {e}")
try:
import json as _json
with open(manifest_path, 'w') as f:
_json.dump(manifest_entries, f, indent=2)
print(f"\nDeletion manifest written to: {manifest_path}")
except Exception as e:
print(f"Warning: Could not write deletion manifest: {e}")
else: else:
# Move duplicates to delete me folder (original behavior) # Move duplicates to delete me folder (original behavior)
if not self._create_delete_folder(): if not self._create_delete_folder():
@ -979,221 +877,99 @@ class EpisodeRenamer:
return matched_episodes return matched_episodes
def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]:
"""Match episodes using DP-based optimal assignment considering disc size, episode deltas, and TVDB duration.""" """Match episodes using lexicographical order and precise duration matching."""
allowed = list(range(1, len(tvdb_episodes) + 1)) matched_episodes = []
# Primary: DP-based optimal matching # Try precise duration matching first
print(" Running DP-based episode matcher (disc size + episode deltas + TVDB duration)...") duration_matches = self._precise_duration_match(episodes_info, tvdb_episodes, list(range(1, len(tvdb_episodes) + 1)))
matched_episodes = self._dp_match_episodes(episodes_info, tvdb_episodes, allowed)
for match in matched_episodes: for match in duration_matches:
matched_episodes.append(match)
duration_diff = match.get('duration_diff', 0) duration_diff = match.get('duration_diff', 0)
fn = match['file_info']['path'].name[:50]
if duration_diff is not None: if duration_diff is not None:
print(f" DP match: {fn}... → Episode {match['episode_number']}{duration_diff:.1f}min)") print(f" Duration match: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']}{duration_diff:.1f}min)")
else: else:
print(f" DP fallback: {fn}... → Episode {match['episode_number']} (no duration match)") print(f" Sequential fallback: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (no duration match)")
return matched_episodes return matched_episodes
def _dp_match_episodes(self, files: List[Dict], tvdb_episodes: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""Dynamic programming matcher considering disc size, episode deltas, and TVDB duration.
Builds a cost matrix over (file, episode) pairs where cost = |file_dur - tvdb_dur|, def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict], allowed_episodes: List[int]) -> List[Dict]:
then finds the minimum-cost assignment that respects: """Perform precise duration matching with fallback to lexicographical order."""
- Disc capacity: each disc holds at most its capacity matched_episodes = []
- Disc ordering: files from earlier discs map to earlier episodes used_files = set()
- Episode deltas: consecutive files prefer consecutive episodes used_episodes = set()
Returns matched episodes sorted by episode number. # First pass: Find exact or very close duration matches (within 5 minutes)
""" for file_info in files:
n_files = len(files) if id(file_info) in used_files:
n_episodes = len(tvdb_episodes)
if not n_files or not n_episodes:
return self._precise_duration_match(files, tvdb_episodes, allowed_episodes)
allowed_set = set(allowed_episodes)
INF = float('inf')
cost = [[INF] * n_episodes for _ in range(n_files)]
for fi in range(n_files):
fdur = files[fi]['duration']
for ei in range(n_episodes):
ep_num = tvdb_episodes[ei]['episode_number']
if ep_num not in allowed_set:
continue continue
tdur = tvdb_episodes[ei].get('runtime', 0)
cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur)
# DP: dp[i][j] = min cost to assign first i files from first j episodes file_duration = file_info['duration']
# Transition: best_match = None
# dp[i][j] = min(dp[i][j-1], # skip episode j-1 best_score = float('inf')
# dp[i-1][j-1] + cost) # assign file i-1 to episode j-1 diffs = [] # Diagnostics: collect per-episode duration differences
dp = [[INF] * (n_episodes + 1) for _ in range(n_files + 1)]
for j in range(n_episodes + 1):
dp[0][j] = 0.0
for i in range(1, n_files + 1): for tvdb_ep in tvdb_episodes:
dp[i][0] = INF if tvdb_ep['episode_number'] in used_episodes:
for j in range(1, n_episodes + 1): continue
dp[i][j] = dp[i][j - 1] if tvdb_ep['episode_number'] not in allowed_episodes:
c = cost[i - 1][j - 1] continue
if c < INF and dp[i - 1][j - 1] < INF:
dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c)
# Backtrack tvdb_duration = tvdb_ep.get('runtime', 0)
best_j = min(range(1, n_episodes + 1), key=lambda j: dp[n_files][j]) if tvdb_duration == 0:
if dp[n_files][best_j] >= INF: continue
return self._fallback_sequential(files, tvdb_episodes, allowed_episodes)
assignment = {} # Calculate duration difference
i, j = n_files, best_j duration_diff = abs(file_duration - tvdb_duration)
while i > 0 and j > 0: diffs.append((tvdb_ep['episode_number'], tvdb_duration, duration_diff))
if dp[i][j] == dp[i][j - 1]:
j -= 1 if duration_diff < best_score:
elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF: best_score = duration_diff
assignment[i - 1] = j - 1 best_match = tvdb_ep
i -= 1
j -= 1 # Accept matches within 1 minute as good matches (strict duration matching)
else: if best_match and best_score <= 1.0:
matched_episodes.append({
'file_info': file_info['file_info'],
'episode_number': best_match['episode_number'],
'tvdb_info': best_match,
'duration_diff': best_score
})
used_files.add(id(file_info))
used_episodes.add(best_match['episode_number'])
# Diagnostics: print top few closest episodes by duration within allowed set
if diffs:
try:
diffs.sort(key=lambda x: x[2])
top = ", ".join([f"E{ep}:{dur}{diff:.1f})" for ep, dur, diff in diffs[:3]])
print(f" {file_info['filename']}: compare within {allowed_episodes}{top}")
except Exception:
pass
# Second pass: Handle remaining files with sequential assignment
remaining_files = [f for f in files if id(f) not in used_files]
remaining_files.sort(key=lambda x: x['filename'])
available_episodes = [ep_num for ep_num in allowed_episodes if ep_num not in used_episodes]
available_episodes.sort()
for i, file_info in enumerate(remaining_files):
if i < len(available_episodes):
episode_number = available_episodes[i]
# Find corresponding TVDB episode
tvdb_match = None
for tvdb_ep in tvdb_episodes:
if tvdb_ep['episode_number'] == episode_number:
tvdb_match = tvdb_ep
break break
matched_episodes = []
for fi_idx, ep_idx in assignment.items():
ep = tvdb_episodes[ep_idx]
fdur = files[fi_idx]['duration']
tdur = ep.get('runtime', 0)
diff = abs(fdur - tdur) if tdur > 0 else None
matched_episodes.append({ matched_episodes.append({
'file_info': files[fi_idx]['file_info'], 'file_info': file_info['file_info'],
'episode_number': ep['episode_number'], 'episode_number': episode_number,
'tvdb_info': ep, 'tvdb_info': tvdb_match,
'duration_diff': diff, 'duration_diff': None
})
assigned_files = set(assignment.keys())
assigned_eps = set(assignment.values())
remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_files]
remaining_eps = [tvdb_episodes[idx] for idx in range(n_episodes) if idx not in assigned_eps]
remaining_eps.sort(key=lambda x: x['episode_number'])
for fi, ep in zip(remaining_files, remaining_eps):
matched_episodes.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
}) })
return matched_episodes return matched_episodes
def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""DP-based optimal duration matching within allowed episodes.
Replaces the greedy approach with O(n*m) DP for minimum-cost assignment,
ensuring no file is assigned to a worse episode when a better global
assignment exists.
"""
n_files = len(files)
allowed_set = set(allowed_episodes)
allowed_tvdb = sorted(
[ep for ep in tvdb_episodes if ep['episode_number'] in allowed_set],
key=lambda x: x['episode_number'],
)
n_eps = len(allowed_tvdb)
if not n_files or not n_eps:
return self._fallback_sequential(files, allowed_tvdb, allowed_episodes)
INF = float('inf')
# Cost matrix: cost[fi][ei] = |file_duration - tvdb_duration|
cost = [[INF] * n_eps for _ in range(n_files)]
for fi in range(n_files):
fdur = files[fi]['duration']
for ei in range(n_eps):
tdur = allowed_tvdb[ei].get('runtime', 0)
cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur)
# DP: dp[i][j] = min cost to assign first i files from first j episodes
# dp[i][j] = min(dp[i][j-1], dp[i-1][j-1] + cost[i-1][j-1])
dp = [[INF] * (n_eps + 1) for _ in range(n_files + 1)]
for j in range(n_eps + 1):
dp[0][j] = 0.0
for i in range(1, n_files + 1):
dp[i][0] = INF
for j in range(1, n_eps + 1):
# Skip episode j-1
dp[i][j] = dp[i][j - 1]
# Assign file i-1 to episode j-1
c = cost[i - 1][j - 1]
if c < INF and dp[i - 1][j - 1] < INF:
dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c)
# Backtrack to recover assignment
assignment = {}
best_j = min(range(1, n_eps + 1), key=lambda j: dp[n_files][j])
if dp[n_files][best_j] >= INF:
return self._fallback_sequential(files, allowed_tvdb, allowed_episodes)
i, j = n_files, best_j
while i > 0 and j > 0:
if j > 0 and dp[i][j] == dp[i][j - 1]:
j -= 1
elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF:
assignment[i - 1] = j - 1
i -= 1
j -= 1
else:
break
matched_episodes = []
for fi, ei in assignment.items():
ep = allowed_tvdb[ei]
fdur = files[fi]['duration']
tdur = ep.get('runtime', 0)
diff = abs(fdur - tdur) if tdur > 0 else None
matched_episodes.append({
'file_info': files[fi]['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': diff,
'assignment_cost': diff or 0,
})
# Fallback for unassigned files
assigned_fi = set(assignment.keys())
assigned_ei = set(assignment.values())
remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_fi]
remaining_eps = [e for idx, e in enumerate(allowed_tvdb) if idx not in assigned_ei]
for fi, ep in zip(remaining_files, remaining_eps):
matched_episodes.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
'assignment_cost': 0,
})
return matched_episodes
def _fallback_sequential(self, files: List[Dict], tvdb_eps: List[Dict],
allowed_episodes: List[int]) -> List[Dict]:
"""Sequential fallback when DP cannot produce an assignment."""
matched = []
sorted_files = sorted(files, key=lambda x: x['filename'])
allowed_set = set(allowed_episodes)
avail = [ep for ep in tvdb_eps if ep['episode_number'] in allowed_set]
for i, fi in enumerate(sorted_files):
if i < len(avail):
ep = avail[i]
matched.append({
'file_info': fi['file_info'],
'episode_number': ep['episode_number'],
'tvdb_info': ep,
'duration_diff': None,
})
return matched

View File

@ -15,96 +15,47 @@ except ImportError:
config_manager = None config_manager = None
# Supported video extensions (configurable via config or env var)
DEFAULT_VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.webm', '.mov', '.wmv', '.flv', '.m4v'}
class FileClassifier: class FileClassifier:
"""Classifies video files as episodes or extras based on file size and video duration.""" """Classifies video files as episodes or extras based on file size and video duration."""
def __init__(self, folder_path: str, video_extensions: List[str] = None, def __init__(self, folder_path: str):
fallback_duration_minutes_per_gb: float = None): """Initialize with folder path containing video files."""
"""Initialize with folder path containing video files.
Args:
folder_path: Path to folder with video files.
video_extensions: List of extensions to process (e.g., ['.mkv', '.mp4']).
Defaults to all common formats.
fallback_duration_minutes_per_gb: Minutes of video per GB when MediaInfo
is unavailable. Defaults to 45 (configurable for 4K or low-bitrate content).
"""
self.folder_path = Path(folder_path) self.folder_path = Path(folder_path)
self.video_extensions = self._resolve_extensions(video_extensions)
self.fallback_ratio = self._resolve_fallback_ratio(fallback_duration_minutes_per_gb)
self.video_files = self._get_video_files() self.video_files = self._get_video_files()
self.file_info = self._analyze_files() self.file_info = self._analyze_files()
def _resolve_extensions(self, extensions: List[str] = None) -> set:
"""Resolve the set of video extensions to scan."""
if extensions:
return {e.lower() if e.startswith('.') else f'.{e.lower()}' for e in extensions}
env_ext = os.environ.get('VIDEO_EXTENSIONS', '')
if env_ext:
return {e.strip().lower() for e in env_ext.split(',') if e.strip()}
return DEFAULT_VIDEO_EXTENSIONS
def _resolve_fallback_ratio(self, ratio: float = None) -> float:
"""Resolve the fallback minutes/GB ratio for duration estimation."""
if ratio is not None:
return float(ratio)
env_ratio = os.environ.get('FALLBACK_DURATION_RATIO', '')
if env_ratio:
try:
return float(env_ratio)
except ValueError:
pass
if config_manager:
return config_manager.get_default_episode_duration()
return 45.0
def _get_video_files(self) -> List[Path]: def _get_video_files(self) -> List[Path]:
"""Get all supported video files in the folder, excluding system files.""" """Get all MKV files in the folder, excluding system files."""
if not self.folder_path.exists(): if not self.folder_path.exists():
raise FileNotFoundError(f"Folder not found: {self.folder_path}") raise FileNotFoundError(f"Folder not found: {self.folder_path}")
video_files = [] all_mkv_files = list(self.folder_path.glob("*.mkv"))
for ext in self.video_extensions:
video_files.extend(self.folder_path.glob(f'*{ext}'))
# Filter out macOS resource fork files and other system files # Filter out macOS resource fork files and other system files
filtered = [] video_files = []
for file in video_files: for file in all_mkv_files:
if file.name.startswith('._'): if file.name.startswith('._'):
continue continue # Skip macOS resource fork files
if file.name.startswith('.'): if file.name.startswith('.'):
continue continue # Skip any hidden files
filtered.append(file) video_files.append(file)
# Deduplicate (in case overlapping extensions) if not video_files:
seen = set() print(f"Warning: No valid MKV files found in {self.folder_path}")
unique = []
for f in filtered:
if f not in seen:
seen.add(f)
unique.append(f)
if not unique: return video_files
ext_list = ', '.join(sorted(self.video_extensions))
print(f"Warning: No valid video files found in {self.folder_path} (scanned: {ext_list})")
return unique
def _get_file_size(self, file_path: Path) -> int: def _get_file_size(self, file_path: Path) -> int:
"""Get file size in bytes.""" """Get file size in bytes."""
return file_path.stat().st_size return file_path.stat().st_size
def _get_video_duration(self, file_path: Path) -> float: def _get_video_duration(self, file_path: Path) -> float:
"""Get video duration in minutes using MediaInfo, with configurable fallback.""" """Get video duration in minutes using MediaInfo."""
if MediaInfo is None: if MediaInfo is None:
print(f"MediaInfo not available, using file size as proxy for duration " print("MediaInfo not available, using file size as proxy for duration")
f"(ratio: {self.fallback_ratio:.1f} min/GB)") # Rough estimate: 1GB ≈ 45 minutes for typical video
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * 45
try: try:
media_info = MediaInfo.parse(str(file_path)) media_info = MediaInfo.parse(str(file_path))
@ -112,6 +63,7 @@ class FileClassifier:
if track.track_type == 'Video': if track.track_type == 'Video':
duration_ms = track.duration duration_ms = track.duration
if duration_ms: if duration_ms:
# Handle both string and numeric duration values
if isinstance(duration_ms, str): if isinstance(duration_ms, str):
try: try:
duration_ms = float(duration_ms) duration_ms = float(duration_ms)
@ -121,19 +73,19 @@ class FileClassifier:
duration_minutes = float(duration_ms) / (1000 * 60) duration_minutes = float(duration_ms) / (1000 * 60)
# Sanity check: fall back to size estimation for suspicious durations # Sanity check: if duration seems unreasonable, fall back to size estimation
if duration_minutes > 300: if duration_minutes > 300: # More than 5 hours is likely wrong
print(f"Warning: Suspicious duration ({duration_minutes:.1f} min) for {file_path.name}, " print(f"Warning: Suspicious duration ({duration_minutes:.1f} min) for {file_path.name}, using size estimation")
f"using size estimation ({self.fallback_ratio:.1f} min/GB)")
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * 45
return duration_minutes return duration_minutes
except Exception as e: except Exception as e:
print(f"Error getting duration for {file_path}: {e}") print(f"Error getting duration for {file_path}: {e}")
# Fallback to size-based estimation
size_gb = self._get_file_size(file_path) / (1024**3) size_gb = self._get_file_size(file_path) / (1024**3)
return size_gb * self.fallback_ratio return size_gb * 45
def _analyze_files(self) -> List[Dict]: def _analyze_files(self) -> List[Dict]:
"""Analyze all video files to get size and duration info.""" """Analyze all video files to get size and duration info."""

Binary file not shown.

View File

@ -20,28 +20,15 @@ class TVDBCache:
self.cache_dir.mkdir(exist_ok=True) self.cache_dir.mkdir(exist_ok=True)
self.cache_duration = timedelta(days=7) # Cache for 7 days self.cache_duration = timedelta(days=7) # Cache for 7 days
def _sanitize_name(self, name: str) -> str:
"""Sanitize series name to prevent path traversal attacks."""
import re
sanitized = name.lower()
sanitized = re.sub(r'[^a-z0-9_\-\s]', '', sanitized)
sanitized = sanitized.replace(' ', '_').replace('-', '_')
if '..' in sanitized or '/' in sanitized or '\\' in sanitized:
raise ValueError(f"Invalid series name containing path traversal: {name!r}")
return sanitized
def _get_cache_key(self, series_name: str, season_number: int) -> str: def _get_cache_key(self, series_name: str, season_number: int) -> str:
"""Generate cache key for series/season combination.""" """Generate cache key for series/season combination."""
sanitized_name = self._sanitize_name(series_name) # Normalize series name for consistent caching
return f"{sanitized_name}_s{season_number:02d}" normalized_name = series_name.lower().replace(' ', '_').replace('-', '_')
return f"{normalized_name}_s{season_number:02d}"
def _get_cache_file(self, cache_key: str) -> Path: def _get_cache_file(self, cache_key: str) -> Path:
"""Get cache file path for given key, validated to stay within cache_dir.""" """Get cache file path for given key."""
cache_file = self.cache_dir / f"{cache_key}.json" return self.cache_dir / f"{cache_key}.json"
resolved = cache_file.resolve()
if not str(resolved).startswith(str(self.cache_dir.resolve())):
raise ValueError(f"Cache path escape detected: {cache_key!r}")
return cache_file
def _is_cache_valid(self, cache_file: Path) -> bool: def _is_cache_valid(self, cache_file: Path) -> bool:
"""Check if cache file exists and is not expired.""" """Check if cache file exists and is not expired."""

View File

@ -1 +0,0 @@
"""Episode Matcher source packages."""

Binary file not shown.

View File

@ -55,10 +55,7 @@ class ConfigManager:
return default_config return default_config
def get_tvdb_api_key(self) -> Optional[str]: def get_tvdb_api_key(self) -> Optional[str]:
"""Get TVDB API key from environment variable or config file.""" """Get TVDB API key from config."""
env_key = os.environ.get("TVDB_API_KEY", "").strip()
if env_key and env_key != "YOUR_TVDB_API_KEY_HERE":
return env_key
api_key = self.config.get("tvdb_api_key", "").strip() api_key = self.config.get("tvdb_api_key", "").strip()
if not api_key or api_key == "YOUR_TVDB_API_KEY_HERE": if not api_key or api_key == "YOUR_TVDB_API_KEY_HERE":
return None return None

View File

View File

@ -1,140 +0,0 @@
"""Shared fixtures for Episode Matcher tests."""
import sys
import os
import tempfile
from pathlib import Path
# Ensure src is importable
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
@pytest.fixture
def tmp_dir(tmp_path):
"""Return a temporary directory as Path."""
return tmp_path
@pytest.fixture
def mock_video_files(tmp_path):
"""Create mock video files in temp dir and return the dir.
Uses small sizes (KB range) since actual content doesn't matter for tests.
"""
files = {
"Disc 1_t01.mkv": int(2 * 1024),
"Disc 1_t02.mkv": int(2100),
"Disc 1_t03.mkv": int(2200),
"Disc 2_t01.mkv": int(2000),
"Disc 2_t02.mkv": int(2100),
"Disc 2_t03.mkv": int(1900),
"extra_short.mkv": int(100),
}
for name, size in files.items():
fpath = tmp_path / name
fpath.write_bytes(b"\x00" * size)
return tmp_path
@pytest.fixture
def mock_episodes():
"""Mock episode file info dicts with realistic ratios but small absolute sizes."""
return [
{
"path": Path("Disc 1_t01.mkv"),
"name": "Disc 1_t01.mkv",
"size_bytes": int(2 * 1024),
"size_gb": 2.0,
"duration_minutes": 61.0,
},
{
"path": Path("Disc 1_t02.mkv"),
"name": "Disc 1_t02.mkv",
"size_bytes": int(2100),
"size_gb": 2.1,
"duration_minutes": 55.0,
},
{
"path": Path("Disc 1_t03.mkv"),
"name": "Disc 1_t03.mkv",
"size_bytes": int(2200),
"size_gb": 2.2,
"duration_minutes": 57.0,
},
{
"path": Path("Disc 2_t01.mkv"),
"name": "Disc 2_t01.mkv",
"size_bytes": int(2000),
"size_gb": 2.0,
"duration_minutes": 55.0,
},
{
"path": Path("Disc 2_t02.mkv"),
"name": "Disc 2_t02.mkv",
"size_bytes": int(2100),
"size_gb": 2.1,
"duration_minutes": 54.0,
},
{
"path": Path("Disc 2_t03.mkv"),
"name": "Disc 2_t03.mkv",
"size_bytes": int(1900),
"size_gb": 1.9,
"duration_minutes": 52.0,
},
]
@pytest.fixture
def mock_tvdb_episodes():
"""Mock TVDB episode data."""
return [
{"episode_number": 1, "name": "Winter Is Coming", "runtime": 61},
{"episode_number": 2, "name": "The Kingsroad", "runtime": 55},
{"episode_number": 3, "name": "Lord Snow", "runtime": 57},
{"episode_number": 4, "name": "Cripples Bastards", "runtime": 55},
{"episode_number": 5, "name": "The Wolf", "runtime": 54},
{"episode_number": 6, "name": "A Golden Crown", "runtime": 52},
{"episode_number": 7, "name": "You Win", "runtime": 57},
{"episode_number": 8, "name": "The Pointy End", "runtime": 58},
{"episode_number": 9, "name": "Baelor", "runtime": 56},
{"episode_number": 10, "name": "Fire and Blood", "runtime": 52},
]
@pytest.fixture
def disc_mapping():
"""Mock disc-to-episode mapping."""
return {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8]}
@pytest.fixture
def config_file(tmp_path):
"""Create a config.json in temp dir."""
import json
cfg = tmp_path / "config.json"
cfg.write_text(
json.dumps(
{
"tvdb_api_key": "test_api_key_12345",
"default_episode_duration": 45,
"classification_thresholds": {
"size_threshold_ratio": 0.3,
"duration_threshold_ratio": 0.4,
},
}
)
)
return cfg
@pytest.fixture
def empty_config_file(tmp_path):
"""Create a config.json with no API key."""
import json
cfg = tmp_path / "config.json"
cfg.write_text(json.dumps({"tvdb_api_key": ""}))
return cfg

View File

@ -1,149 +0,0 @@
"""Tests for src/config.py - ConfigManager."""
import json
import os
from pathlib import Path
import sys
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.config import ConfigManager
class TestConfigManagerInit:
def test_init_no_config_file(self, tmp_path, monkeypatch):
cfg = ConfigManager(str(tmp_path / "nonexistent.json"))
assert cfg.config["tvdb_api_key"] == ""
assert cfg.config["default_episode_duration"] == 45
assert cfg.config["classification_thresholds"]["size_threshold_ratio"] == 0.3
def test_init_existing_config(self, config_file):
cfg = ConfigManager(str(config_file))
assert cfg.config["tvdb_api_key"] == "test_api_key_12345"
assert cfg.config["default_episode_duration"] == 45
def test_init_invalid_json(self, tmp_path):
bad = tmp_path / "config.json"
bad.write_text("{invalid json")
cfg = ConfigManager(str(bad))
assert cfg.config["tvdb_api_key"] == ""
def test_init_missing_keys_merged(self, tmp_path):
partial = tmp_path / "config.json"
partial.write_text(json.dumps({"tvdb_api_key": "abc"}))
cfg = ConfigManager(str(partial))
assert cfg.config["default_episode_duration"] == 45
assert "classification_thresholds" in cfg.config
def test_init_io_error(self, tmp_path, monkeypatch):
cfg_path = tmp_path / "config.json"
cfg_path.write_text("test")
monkeypatch.setattr(Path, "exists", lambda self: True)
original_open = open
def mock_open(*args, **kwargs):
raise PermissionError("denied")
with patch("builtins.open", mock_open):
cfg = ConfigManager(str(cfg_path))
assert cfg.config["tvdb_api_key"] == ""
class TestConfigManagerApiKeys:
def test_get_api_key_from_config(self, config_file):
cfg = ConfigManager(str(config_file))
with patch.dict(os.environ, {}, clear=True):
key = cfg.get_tvdb_api_key()
assert key == "test_api_key_12345"
def test_get_api_key_from_env(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", "env_key_999")
key = cfg.get_tvdb_api_key()
assert key == "env_key_999"
def test_get_api_key_empty_env(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", " ")
key = cfg.get_tvdb_api_key()
assert key == "test_api_key_12345"
def test_get_api_key_placeholder_rejected(self, empty_config_file):
cfg = ConfigManager(str(empty_config_file))
cfg.config["tvdb_api_key"] = "YOUR_TVDB_API_KEY_HERE"
assert cfg.get_tvdb_api_key() is None
def test_get_api_key_env_placeholder_rejected(self, empty_config_file, monkeypatch):
cfg = ConfigManager(str(empty_config_file))
monkeypatch.setenv("TVDB_API_KEY", "YOUR_TVDB_API_KEY_HERE")
assert cfg.get_tvdb_api_key() is None
def test_env_key_takes_priority(self, config_file, monkeypatch):
cfg = ConfigManager(str(config_file))
monkeypatch.setenv("TVDB_API_KEY", "env_priority")
assert cfg.get_tvdb_api_key() == "env_priority"
class TestConfigManagerSettings:
def test_get_default_episode_duration(self, config_file):
cfg = ConfigManager(str(config_file))
assert cfg.get_default_episode_duration() == 45
def test_get_default_episode_duration_custom(self, tmp_path):
f = tmp_path / "config.json"
f.write_text(json.dumps({"default_episode_duration": 60}))
cfg = ConfigManager(str(f))
assert cfg.get_default_episode_duration() == 60
def test_get_classification_thresholds(self, config_file):
cfg = ConfigManager(str(config_file))
t = cfg.get_classification_thresholds()
assert t["size_threshold_ratio"] == 0.3
assert t["duration_threshold_ratio"] == 0.4
def test_get_classification_thresholds_defaults(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "no.json"))
t = cfg.get_classification_thresholds()
assert t["size_threshold_ratio"] == 0.3
class TestConfigManagerUpdate:
def test_update_api_key_success(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "config.json"))
result = cfg.update_api_key("new_key_abc")
assert result is True
assert cfg.config["tvdb_api_key"] == "new_key_abc"
stored = json.loads((tmp_path / "config.json").read_text())
assert stored["tvdb_api_key"] == "new_key_abc"
def test_update_api_key_io_error(self, tmp_path):
cfg = ConfigManager(str(tmp_path / "config.json"))
cfg.update_api_key("new_key")
cfg.config_file = Path("/nonexistent_dir/config.json")
result = cfg.update_api_key("another_key")
assert result is False
class TestConfigManagerStatus:
def test_print_config_status_no_key(self, tmp_path, capsys):
cfg = ConfigManager(str(tmp_path / "no.json"))
cfg.print_config_status()
out = capsys.readouterr().out
assert "Configuration file:" in out
assert "Not configured" in out
def test_print_config_status_with_key(self, config_file, capsys):
cfg = ConfigManager(str(config_file))
cfg.print_config_status()
out = capsys.readouterr().out
assert "configured" in out
def test_print_config_status_masked_short_key(self, tmp_path):
f = tmp_path / "config.json"
f.write_text(json.dumps({"tvdb_api_key": "ab"}))
cfg = ConfigManager(str(f))
cfg.print_config_status()

View File

@ -1,110 +0,0 @@
"""Tests for episode_matcher.py - parse_disc_mapping and CLI."""
import sys
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from episode_matcher import parse_disc_mapping
class TestParseDiscMapping:
def test_single_disc_range(self):
result = parse_disc_mapping("1:1-3")
assert result == {1: [1, 2, 3]}
def test_single_disc_single_ep(self):
result = parse_disc_mapping("1:5")
assert result == {1: [5]}
def test_multi_disc(self):
result = parse_disc_mapping("1:1-3,2:4-6,3:7-9,4:10")
assert result == {1: [1, 2, 3], 2: [4, 5, 6], 3: [7, 8, 9], 4: [10]}
def test_spaces(self):
result = parse_disc_mapping("1:1-3 , 2:4-6")
assert result == {1: [1, 2, 3], 2: [4, 5, 6]}
def test_empty_string(self):
result = parse_disc_mapping("")
assert result == {}
def test_none(self):
result = parse_disc_mapping(None)
assert result == {}
def test_single_ep_per_disc(self):
result = parse_disc_mapping("1:1,2:2,3:3")
assert result == {1: [1], 2: [2], 3: [3]}
def test_large_numbers(self):
result = parse_disc_mapping("1:1-22")
assert result == {1: list(range(1, 23))}
def test_invalid_format_exits(self, capsys):
with pytest.raises(SystemExit):
parse_disc_mapping("invalid")
def test_missing_colon_exits(self, capsys):
with pytest.raises(SystemExit):
parse_disc_mapping("1,2,3")
class TestArgParse:
def test_parser_basic_args(self):
import argparse
from episode_matcher import main
with patch("sys.argv", [
"episode_matcher.py",
"/tmp/episodes",
"Test Show",
"1",
]):
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.is_dir", return_value=True):
with patch("episode_matcher.TVDBClient") as mock_tvdb:
mock_instance = mock_tvdb.return_value
mock_instance.authenticate.return_value = False
with patch("episode_matcher.MockTVDBClient") as mock_mock:
m = mock_mock.return_value
m.authenticate.return_value = True
m.get_episode_durations.return_value = []
with patch("episode_matcher.FileClassifier") as mock_fc:
fc = mock_fc.return_value
fc.file_info = []
fc.classify_files.return_value = ([], [])
with pytest.raises(SystemExit):
main()
def test_parser_dry_run_flag(self):
import sys as real_sys
orig_argv = real_sys.argv
try:
real_sys.argv = [
"episode_matcher.py", "/tmp/e", "Show", "1", "--dry-run", "--verbose"
]
from importlib import reload
import episode_matcher
parser = episode_matcher.__dict__.get("_parser", None)
from episode_matcher import main
finally:
real_sys.argv = orig_argv
class TestParseDiscMappingEdgeCases:
def test_range_to_self(self):
result = parse_disc_mapping("1:5-5")
assert result == {1: [5]}
def test_multiple_ranges_same_ep(self):
result = parse_disc_mapping("1:1-3,2:3-5")
assert 3 in result[1]
assert 3 in result[2]
def test_disc_number_large(self):
result = parse_disc_mapping("99:1-3")
assert 99 in result

View File

@ -1,539 +0,0 @@
"""Tests for src/Matcher/episode_renamer.py - EpisodeRenamer."""
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.Matcher.episode_renamer import EpisodeRenamer
class TestEpisodeRenamerInit:
def test_init_defaults(self):
r = EpisodeRenamer("/tmp/test", "Show Name", 1)
assert r.show_name == "Show Name"
assert r.season_number == 1
assert r.disc_mapping is None
assert r.auto_delete_duplicates is False
def test_init_with_disc_mapping(self):
r = EpisodeRenamer("/tmp/test", "Show", 2, disc_mapping={1: [1, 2], 2: [3, 4]})
assert r.disc_mapping == {1: [1, 2], 2: [3, 4]}
def test_init_auto_delete(self):
r = EpisodeRenamer("/tmp/test", "Show", 1, auto_delete_duplicates=True)
assert r.auto_delete_duplicates is True
def test_init_folder_paths(self):
r = EpisodeRenamer("/tmp/test", "Show", 1)
assert r.extras_folder == Path("/tmp/test/extras")
assert r.delete_folder == Path("/tmp/test/delete me")
class TestSanitizeFilename:
def test_remove_invalid_chars(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._sanitize_filename('a<b>c:d"e/f\\g|h?i*j')
assert ":" not in result
assert "<" not in result
assert ">" not in result
assert '"' not in result
assert "/" not in result
assert "\\" not in result
assert "|" not in result
assert "?" not in result
assert "*" not in result
assert "a" in result and "b" in result and "c" in result
def test_collapse_spaces(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename("a b") == "a b"
def test_strip(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename(" hello ") == "hello"
def test_preserve_valid_chars(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r._sanitize_filename("hello_world-1.mkv") == "hello_world-1.mkv"
class TestGenerateEpisodeFilename:
def test_standard(self):
r = EpisodeRenamer("/tmp/t", "Show Name", 3)
assert r._generate_episode_filename(5, ".mkv") == "Show Name s03e05.mkv"
def test_single_digit_padding(self):
r = EpisodeRenamer("/tmp/t", "Show", 1)
assert r._generate_episode_filename(1, ".mp4") == "Show s01e01.mp4"
def test_double_digit_episode(self):
r = EpisodeRenamer("/tmp/t", "Show", 10)
assert r._generate_episode_filename(12, ".mkv") == "Show s10e12.mkv"
def test_sanitizes_output(self):
r = EpisodeRenamer("/tmp/t", "Show:Name", 1)
assert r._generate_episode_filename(1, ".mkv") == "ShowName s01e01.mkv"
class TestExtractDiscInfo:
def test_disc_with_underscore(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disc 1_t01.mkv")
assert info["disc_number"] == 1
assert info["track_number"] == 1
def test_disk_variant(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disk2_t03.mkv")
assert info["disc_number"] == 2
assert info["track_number"] == 3
def test_disc_hyphen(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("disc-3_track05.mkv")
assert info["disc_number"] == 3
assert info["track_number"] == 5
def test_no_disc_info(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("episode_01.mkv")
assert info["disc_number"] is None
assert info["track_number"] is None
def test_track_pattern(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("file_t22.mkv")
assert info["track_number"] == 22
def test_disc_only(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = r._extract_disc_info("Disc 5.mkv")
assert info["disc_number"] == 5
assert info["track_number"] is None
class TestComputeFileHash:
def test_hash_consistent(self, tmp_path):
f = tmp_path / "test.mkv"
f.write_bytes(b"\x00" * 1000)
r = EpisodeRenamer(str(tmp_path), "S", 1)
h1 = r._compute_file_hash(f)
h2 = r._compute_file_hash(f)
assert h1 == h2
assert len(h1) == 64
def test_hash_different_content(self, tmp_path):
f1 = tmp_path / "a.mkv"
f2 = tmp_path / "b.mkv"
f1.write_bytes(b"\x00" * 100)
f2.write_bytes(b"\xff" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._compute_file_hash(f1) != r._compute_file_hash(f2)
def test_hash_custom_bytes(self, tmp_path):
f = tmp_path / "test.mkv"
f.write_bytes(b"\x00" * 10000)
r = EpisodeRenamer(str(tmp_path), "S", 1)
h = r._compute_file_hash(f, num_bytes=512)
assert h
def test_hash_missing_file(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
h = r._compute_file_hash(tmp_path / "nope.mkv")
assert h == ""
class TestCreateFolders:
def test_create_extras_folder(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_extras_folder() is True
assert (tmp_path / "extras").exists()
def test_create_delete_folder(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_delete_folder() is True
assert (tmp_path / "delete me").exists()
def test_create_extras_already_exists(self, tmp_path):
(tmp_path / "extras").mkdir()
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r._create_extras_folder() is True
class TestMoveExtras:
def test_move_extras(self, tmp_path):
(tmp_path / "extra.mkv").write_bytes(b"\x00" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}]
moved = r.move_extras_to_folder(extras)
assert len(moved) == 1
assert (tmp_path / "extras" / "extra.mkv").exists()
assert not (tmp_path / "extra.mkv").exists()
def test_move_extras_skip_existing(self, tmp_path):
(tmp_path / "extras").mkdir()
(tmp_path / "extras" / "extra.mkv").write_bytes(b"\x00")
(tmp_path / "extra.mkv").write_bytes(b"\x00" * 100)
r = EpisodeRenamer(str(tmp_path), "S", 1)
extras = [{"path": tmp_path / "extra.mkv", "name": "extra.mkv"}]
moved = r.move_extras_to_folder(extras)
assert len(moved) == 0
def test_move_extras_empty(self, tmp_path):
r = EpisodeRenamer(str(tmp_path), "S", 1)
assert r.move_extras_to_folder([]) == []
class TestDuplicateDetection:
def test_detect_duplicates_same_duration(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 1_t02.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x00" * 2000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1)
result = r.detect_and_move_duplicates(eps)
assert isinstance(result, list)
def test_no_duplicates_different_duration(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 2_t01.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x01" * 3000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 2_t01.mkv", "size_gb": 3.0, "duration_minutes": 55.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1)
r.detect_and_move_duplicates(eps)
def test_duplicate_moved_to_delete_me(self, tmp_path):
f1 = tmp_path / "Disc 1_t01.mkv"
f2 = tmp_path / "Disc 1_t02.mkv"
f1.write_bytes(b"\x00" * 2000)
f2.write_bytes(b"\x00" * 2000)
eps = [
{"path": f1, "name": "Disc 1_t01.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
{"path": f2, "name": "Disc 1_t02.mkv", "size_gb": 2.0, "duration_minutes": 45.0},
]
r = EpisodeRenamer(str(tmp_path), "S", 1, auto_delete_duplicates=False)
result = r.detect_and_move_duplicates(eps)
assert (tmp_path / "delete me").exists()
class TestMatchEpisodes:
def test_disc_mapping_match(self, mock_episodes, mock_tvdb_episodes, disc_mapping):
r = EpisodeRenamer("/tmp/t", "Show", 1, disc_mapping=disc_mapping)
episodes_info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
episodes_info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"track_number": disc_info["track_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
matched = r._match_using_disc_mapping(episodes_info, mock_tvdb_episodes)
assert len(matched) > 0
ep_nums = [m["episode_number"] for m in matched]
assert len(set(ep_nums)) == len(ep_nums)
def test_fallback_sequential(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 11))
result = r._fallback_sequential(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
def test_dp_match_episodes(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 11))
result = r._dp_match_episodes(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
ep_nums = [m["episode_number"] for m in result]
assert len(set(ep_nums)) == len(ep_nums)
def test_precise_duration_match(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = []
for ep in mock_episodes:
files.append({
"file_info": ep,
"filename": ep["name"],
"disc_number": None,
"duration": ep["duration_minutes"],
})
allowed = list(range(1, 7))
result = r._precise_duration_match(files, mock_tvdb_episodes, allowed)
assert len(result) > 0
def test_dp_match_empty_files(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._dp_match_episodes([], mock_tvdb_episodes, list(range(1, 11)))
assert isinstance(result, list)
def test_precise_match_empty(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
files = [{"file_info": {}, "filename": "a.mkv", "duration": 45.0}]
result = r._precise_duration_match(files, [], [])
assert isinstance(result, list)
def test_fallback_sequential_empty(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._fallback_sequential([], mock_tvdb_episodes, [])
assert result == []
class TestRenameEpisodes:
def test_rename_no_episodes(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
assert r.rename_episodes([]) == []
def test_rename_with_tvdb(self, tmp_path, mock_episodes, mock_tvdb_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "TestShow", 1)
renamed = r.rename_episodes(mock_episodes, mock_tvdb_episodes)
assert len(renamed) > 0
for item in renamed:
assert "original_name" in item
assert "new_name" in item
assert "episode_number" in item
def test_rename_skip_existing(self, tmp_path, mock_episodes, capsys):
for ep in mock_episodes[:2]:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
(tmp_path / "TestShow s01e01.mkv").write_bytes(b"\x00")
r = EpisodeRenamer(str(tmp_path), "TestShow", 1)
renamed = r.rename_episodes(mock_episodes[:2], None)
out = capsys.readouterr().out
assert "already exists" in out or len(renamed) < 2
class TestEstimateEpisodesPerDisc:
def test_estimate_basic(self, tmp_path, mock_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "S", 1)
episodes_info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
episodes_info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
})
result = r._estimate_episodes_per_disc(episodes_info, 6)
assert isinstance(result, dict)
def test_estimate_no_disc_info(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._estimate_episodes_per_disc([], 10)
assert result == {}
def test_estimate_no_disc_numbers(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
info = [{"disc_number": None, "filename": "a.mkv", "file_info": {"size_gb": 5.0}}]
result = r._estimate_episodes_per_disc(info, 10)
assert result == {}
class TestAnalyzeDiscCapacity:
def test_analyze_basic(self, tmp_path, mock_episodes):
for ep in mock_episodes:
f = tmp_path / ep["name"]
f.write_bytes(b"\x00" * ep["size_bytes"])
ep["path"] = f
r = EpisodeRenamer(str(tmp_path), "S", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
})
result = r._analyze_disc_capacity(info)
assert isinstance(result, dict)
def test_analyze_no_discs(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
result = r._analyze_disc_capacity([{"disc_number": None, "file_info": {"size_gb": 5}}])
assert result == {}
class TestDurationMatching:
def test_match_by_duration_and_order(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"track_number": disc_info["track_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._match_by_duration_and_order(info, mock_tvdb_episodes)
assert len(result) > 0
ep_nums = [m["episode_number"] for m in result]
assert len(set(ep_nums)) == len(ep_nums)
def test_flexible_duration_match(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._flexible_duration_match(info, mock_tvdb_episodes)
assert isinstance(result, list)
def test_sequential_assignment_with_constraints(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
disc_capacity = {1: 3, 2: 3}
result = r._sequential_assignment_with_constraints(info, mock_tvdb_episodes, disc_capacity)
assert isinstance(result, list)
def test_validate_sequential_assignment_valid(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": 1}
tvdb_ep = {"runtime": 55, "episode_number": 1}
assert r._validate_sequential_assignment(file_info, tvdb_ep, {}, {}) is True
def test_validate_sequential_assignment_duration_mismatch(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 30.0, "disc_number": 1}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._validate_sequential_assignment(file_info, tvdb_ep, {}, {})
assert result is False
def test_validate_sequential_assignment_disc_full(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": 1}
tvdb_ep = {"runtime": 55, "episode_number": 1}
disc_assignments = {1: 3}
disc_capacity = {1: 3}
result = r._validate_sequential_assignment(file_info, tvdb_ep, disc_assignments, disc_capacity)
assert result is False
def test_assignment_satisfies_constraints(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
fake_path = type("P", (), {"name": "Disc 1_t01.mkv"})()
file_info = {
"duration": 55.0,
"disc_number": 1,
"file_info": {"path": fake_path},
}
tvdb_ep = {"runtime": 55, "episode_number": 1}
result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {})
assert result is True
def test_assignment_satisfies_duration_exceeded(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
fake_path = type("P", (), {"name": "a.mkv"})()
file_info = {
"duration": 30.0,
"disc_number": None,
"file_info": {"path": fake_path},
}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._assignment_satisfies_constraints(file_info, tvdb_ep, [], {})
assert result is False
class TestMatchByDurationAndDisc:
def test_match_disc(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
episodes_per_disc = {1: [1, 2, 3], 2: [4, 5, 6]}
result = r._match_by_duration_and_disc(info, mock_tvdb_episodes, episodes_per_disc)
assert len(result) > 0
for m in result:
assert m["episode_number"] in [1, 2, 3, 4, 5, 6]
class TestFindOptimalAssignment:
def test_find_optimal_assignment(self, mock_episodes, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
info = []
for ep in mock_episodes:
disc_info = r._extract_disc_info(ep["name"])
info.append({
"file_info": ep,
"disc_number": disc_info["disc_number"],
"filename": ep["name"],
"duration": ep["duration_minutes"],
})
result = r._find_optimal_assignment(info, mock_tvdb_episodes, {1: 3, 2: 3})
assert isinstance(result, list)
def test_find_optimal_empty(self, mock_tvdb_episodes):
r = EpisodeRenamer("/tmp/t", "Show", 1)
result = r._find_optimal_assignment([], mock_tvdb_episodes, {})
assert result == []
class TestIsValidAssignment:
def test_valid_assignment(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 55.0, "disc_number": None}
tvdb_ep = {"runtime": 55, "episode_number": 1}
result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, [])
assert result is True
def test_invalid_duration(self):
r = EpisodeRenamer("/tmp/t", "S", 1)
file_info = {"duration": 30.0, "disc_number": None}
tvdb_ep = {"runtime": 60, "episode_number": 1}
result = r._is_valid_assignment(file_info, tvdb_ep, {}, 0, [])
assert result is False

View File

@ -1,194 +0,0 @@
"""Tests for src/Matcher/file_classifier.py - FileClassifier."""
import os
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.Matcher.file_classifier import FileClassifier, DEFAULT_VIDEO_EXTENSIONS
class TestFileClassifierInit:
def test_init_valid_folder(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert fc.folder_path == mock_video_files
assert len(fc.video_files) > 0
def test_init_nonexistent_folder(self, tmp_path):
with pytest.raises(FileNotFoundError, match="Folder not found"):
FileClassifier(str(tmp_path / "nope"))
def test_init_file_not_dir(self, tmp_path):
f = tmp_path / "file.txt"
f.write_text("hi")
fc = FileClassifier(str(f))
assert len(fc.video_files) == 0
def test_init_custom_extensions(self, tmp_path):
(tmp_path / "video.mp4").write_bytes(b"\x00" * 100)
(tmp_path / "video.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path), video_extensions=[".mp4"])
assert len(fc.video_files) == 1
assert fc.video_files[0].name == "video.mp4"
def test_init_fallback_ratio(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=25.0)
assert fc.fallback_ratio == 25.0
def test_init_filters_macos_resource_fork(self, tmp_path):
(tmp_path / "._hidden.mkv").write_bytes(b"\x00" * 100)
(tmp_path / ".dot.mkv").write_bytes(b"\x00" * 100)
(tmp_path / "visible.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path))
assert len(fc.video_files) == 1
assert fc.video_files[0].name == "visible.mkv"
def test_init_deduplicates(self, tmp_path):
(tmp_path / "dup.mkv").write_bytes(b"\x00" * 100)
fc = FileClassifier(str(tmp_path), video_extensions=[".mkv"])
assert len(fc.video_files) == 1
def test_init_empty_folder(self, tmp_path):
fc = FileClassifier(str(tmp_path))
assert fc.video_files == []
class TestResolveExtensions:
def test_resolve_default(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert ".mkv" in fc.video_extensions
def test_resolve_custom(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), video_extensions=["mp4", "avi"])
assert ".mp4" in fc.video_extensions
assert ".avi" in fc.video_extensions
assert ".mkv" not in fc.video_extensions
def test_resolve_env_var(self, mock_video_files, monkeypatch):
monkeypatch.setenv("VIDEO_EXTENSIONS", "webm,mp4")
fc = FileClassifier(str(mock_video_files))
assert "webm" in fc.video_extensions
assert "mp4" in fc.video_extensions
class TestResolveFallbackRatio:
def test_explicit_ratio(self, mock_video_files):
fc = FileClassifier(str(mock_video_files), fallback_duration_minutes_per_gb=30.0)
assert fc.fallback_ratio == 30.0
def test_env_ratio(self, mock_video_files, monkeypatch):
monkeypatch.setenv("FALLBACK_DURATION_RATIO", "33.0")
fc = FileClassifier(str(mock_video_files))
assert fc.fallback_ratio == 33.0
def test_env_ratio_invalid(self, mock_video_files, monkeypatch):
monkeypatch.setenv("FALLBACK_DURATION_RATIO", "abc")
fc = FileClassifier(str(mock_video_files))
assert fc.fallback_ratio == 45.0
class TestFileAnalysis:
def test_get_file_size(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
size = fc._get_file_size(mock_video_files / "Disc 1_t01.mkv")
assert size == int(2 * 1024)
def test_analyze_files_structure(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
assert len(fc.file_info) > 0
info = fc.file_info[0]
assert "path" in info
assert "name" in info
assert "size_bytes" in info
assert "size_mb" in info
assert "size_gb" in info
assert "duration_minutes" in info
def test_calculate_stats(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
stats = fc._calculate_stats()
assert stats["count"] == len(fc.file_info)
assert "avg_size_bytes" in stats
assert "avg_duration_minutes" in stats
assert "median_size_bytes" in stats
def test_calculate_stats_empty(self, tmp_path):
fc = FileClassifier(str(tmp_path))
stats = fc._calculate_stats()
assert stats == {}
class TestClassifyFiles:
def test_classify_no_files(self, tmp_path):
fc = FileClassifier(str(tmp_path))
episodes, extras = fc.classify_files()
assert episodes == []
assert extras == []
def test_classify_with_tvdb(self, mock_video_files, mock_tvdb_episodes):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(
expected_episode_count=6, tvdb_episodes=mock_tvdb_episodes
)
assert len(episodes) <= 6
assert len(episodes) + len(extras) == len(fc.file_info)
def test_classify_by_stats(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(expected_episode_count=6)
assert len(episodes) <= 6
assert len(episodes) > 0
def test_classify_extra_small_file(self, mock_video_files):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files()
extra_names = [e["name"] for e in extras]
assert "extra_short.mkv" in extra_names
def test_classify_expected_count(self, mock_video_files, mock_tvdb_episodes):
fc = FileClassifier(str(mock_video_files))
episodes, extras = fc.classify_files(
expected_episode_count=3, tvdb_episodes=mock_tvdb_episodes
)
assert len(episodes) <= 3
class TestVideoDuration:
@patch("src.Matcher.file_classifier.MediaInfo", None)
def test_duration_fallback_no_mediainfo(self, mock_video_files):
fc = FileClassifier(
str(mock_video_files), fallback_duration_minutes_per_gb=45.0
)
target = mock_video_files / "Disc 1_t01.mkv"
dur = fc._get_video_duration(target)
expected = (target.stat().st_size / (1024**3)) * 45.0
assert abs(dur - expected) < 0.001
assert dur > 0
def test_duration_fallback_custom_ratio(self, mock_video_files):
with patch("src.Matcher.file_classifier.MediaInfo", None):
fc = FileClassifier(
str(mock_video_files), fallback_duration_minutes_per_gb=25.0
)
target = mock_video_files / "Disc 1_t01.mkv"
dur = fc._get_video_duration(target)
expected = (target.stat().st_size / (1024**3)) * 25.0
assert abs(dur - expected) < 0.001
assert dur > 0
class TestPrintAnalysis:
def test_print_analysis_has_files(self, mock_video_files, capsys):
fc = FileClassifier(str(mock_video_files))
fc.print_analysis()
out = capsys.readouterr().out
assert "File Analysis" in out
assert "Total files" in out
def test_print_analysis_no_files(self, tmp_path, capsys):
fc = FileClassifier(str(tmp_path))
fc.print_analysis()
out = capsys.readouterr().out
assert "No video files found" in out

View File

@ -1,236 +0,0 @@
"""Tests for src/TVDBProvider/tvdb_cache.py - TVDBCache."""
import sys
import json
import time
from pathlib import Path
from datetime import datetime, timedelta
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.TVDBProvider.tvdb_cache import TVDBCache
class TestTVDBCacheInit:
def test_init_creates_dir(self, tmp_path):
cache = TVDBCache(str(tmp_path / "cache"))
assert (tmp_path / "cache").exists()
def test_init_default_dir(self):
cache = TVDBCache()
assert cache.cache_dir.exists()
assert cache.cache_duration == timedelta(days=7)
def test_init_absolute_path(self):
cache = TVDBCache("/tmp/absolute_cache_test")
try:
assert cache.cache_dir.is_absolute()
finally:
import shutil
if cache.cache_dir.exists():
shutil.rmtree(cache.cache_dir, ignore_errors=True)
class TestSanitizeName:
def test_basic_sanitization(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("Game of Thrones")
assert result == "game_of_thrones"
def test_special_chars_removed(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("The Simpsons (1989)")
assert "(" not in result
assert ")" not in result
def test_dots_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("../../etc/passwd")
assert ".." not in result
assert "/" not in result
def test_slashes_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("a/b")
assert result == "ab"
def test_backslash_stripped(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("a\\b")
assert result == "ab"
def test_numbers_preserved(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("Show 123")
assert "123" in result
def test_underscores_preserved(self):
cache = TVDBCache.__new__(TVDBCache)
result = cache._sanitize_name("my_show")
assert result == "my_show"
class TestCacheKey:
def test_cache_key_format(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
key = cache._get_cache_key("Test Show", 3)
assert key == "test_show_s03"
def test_cache_key_special_chars(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
key = cache._get_cache_key("Test (Show)!", 1)
assert "(" not in key
assert ")" not in key
class TestCacheFile:
def test_cache_file_path(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = cache._get_cache_file("test_s01")
assert f.name == "test_s01.json"
def test_cache_file_escape_detected(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
with pytest.raises(ValueError, match="escape detected"):
cache._get_cache_file("../evil")
class TestCacheValidity:
def test_valid_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "valid.json"
f.write_text("data")
assert cache._is_cache_valid(f) is True
def test_expired_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_duration = timedelta(seconds=0)
f = tmp_path / "c" / "old.json"
f.write_text("data")
time.sleep(0.1)
assert cache._is_cache_valid(f) is False
def test_nonexistent_cache(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
assert cache._is_cache_valid(tmp_path / "c" / "nope.json") is False
class TestCacheEpisodes:
def test_cache_and_retrieve(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
episodes = [{"episode_number": 1, "runtime": 45}]
result = cache.cache_episodes("Test Show", 1, episodes)
assert result is True
retrieved = cache.get_cached_episodes("Test Show", 1)
assert len(retrieved) == 1
assert retrieved[0]["episode_number"] == 1
def test_cache_write_failure(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_dir = Path("/nonexistent_dir")
result = cache.cache_episodes("Test", 1, [])
assert result is False
def test_cache_data_structure(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
episodes = [{"episode_number": 1, "runtime": 45}]
cache.cache_episodes("Test", 1, episodes)
cache_file = tmp_path / "c" / "test_s01.json"
data = json.loads(cache_file.read_text())
assert data["series_name"] == "Test"
assert data["season_number"] == 1
assert "cached_at" in data
assert len(data["episodes"]) == 1
def test_cache_empty_episodes(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
result = cache.cache_episodes("Empty", 1, [])
assert result is True
retrieved = cache.get_cached_episodes("Empty", 1)
assert retrieved == []
class TestGetCachedEpisodes:
def test_cache_miss(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
result = cache.get_cached_episodes("NoShow", 1)
assert result is None
def test_corrupted_json(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "broken.json"
f.write_text("{invalid")
result = cache.get_cached_episodes("Broken", 1)
assert result is None
def test_missing_fields(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
f = tmp_path / "c" / "partial_s01.json"
f.write_text(json.dumps({"only": "this"}))
result = cache.get_cached_episodes("Partial", 1)
assert result is None
class TestClearCache:
def test_clear_removes_files(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Show1", 1, [])
cache.cache_episodes("Show2", 1, [])
result = cache.clear_cache()
assert result is True
files = list((tmp_path / "c").glob("*.json"))
assert len(files) == 0
def test_clear_empty(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
assert cache.clear_cache() is True
class TestListCache:
def test_list_entries(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Show1", 1, [{"ep": 1}])
cache.cache_episodes("Show2", 2, [{"ep": 1}, {"ep": 2}])
entries = cache.list_cache()
assert len(entries) == 2
for e in entries:
assert "series_name" in e
assert "season_number" in e
assert "episode_count" in e
assert "is_valid" in e
def test_list_empty(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
entries = cache.list_cache()
assert entries == []
def test_list_entry_fields(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Test", 1, [{"e": 1}])
entries = cache.list_cache()
e = entries[0]
assert e["series_name"] == "Test"
assert e["season_number"] == 1
assert e["episode_count"] == 1
assert "cached_at" in e
assert "age_days" in e
assert "cache_key" in e
def test_list_sorted_by_date(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("First", 1, [])
time.sleep(0.01)
cache.cache_episodes("Second", 1, [])
entries = cache.list_cache()
assert entries[0]["series_name"] == "Second"
def test_list_corrupted_file(self, tmp_path):
cache = TVDBCache(str(tmp_path / "c"))
cache.cache_episodes("Good", 1, [{"e": 1}])
(tmp_path / "c" / "bad_s01.json").write_text("{bad}")
entries = cache.list_cache()
assert len(entries) == 1

View File

@ -1,224 +0,0 @@
"""Tests for src/TVDBProvider/tvdb_client.py."""
import sys
from pathlib import Path
from unittest.mock import patch, MagicMock, PropertyMock
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from src.TVDBProvider.tvdb_client import TVDBClient, MockTVDBClient
class TestTVDBClientInit:
def test_init(self):
client = TVDBClient()
assert client.base_url == "https://api4.thetvdb.com/v4"
assert client.token is None
assert "Content-Type" in client.headers
def test_init_has_cache(self):
client = TVDBClient()
assert client.cache is not None
class TestTVDBClientAuthenticate:
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_success(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {"data": {"token": "abc123"}}
client = TVDBClient()
assert client.authenticate("my_api_key") is True
assert client.token == "abc123"
assert "Bearer abc123" in client.headers["Authorization"]
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_failure_status(self, mock_post):
mock_post.return_value.status_code = 401
client = TVDBClient()
assert client.authenticate("bad_key") is False
assert client.token is None
@patch("src.TVDBProvider.tvdb_client.requests.post")
def test_auth_exception(self, mock_post):
mock_post.side_effect = Exception("network error")
client = TVDBClient()
assert client.authenticate("key") is False
class TestTVDBClientSearch:
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_success(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"data": [{"tvdb_id": 123, "name": "Test Show"}]
}
client = TVDBClient()
client.token = "tok"
result = client.search_series("Test Show")
assert result["tvdb_id"] == 123
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_no_results(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"data": []}
client = TVDBClient()
client.token = "tok"
result = client.search_series("NoMatch")
assert result is None
def test_search_not_authenticated(self):
client = TVDBClient()
result = client.search_series("Show")
assert result is None
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_search_error(self, mock_get):
mock_get.side_effect = Exception("timeout")
client = TVDBClient()
client.token = "tok"
result = client.search_series("Show")
assert result is None
class TestTVDBClientSeasonEpisodes:
@patch("src.TVDBProvider.tvdb_client.requests.get")
def test_get_season_episodes(self, mock_get):
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {
"data": {"episodes": [{"number": 1, "name": "E1"}]}
}
client = TVDBClient()
client.token = "tok"
result = client.get_season_episodes(123, 1)
assert len(result) == 1
def test_episodes_not_authenticated(self):
client = TVDBClient()
result = client.get_season_episodes(123, 1)
assert result is None
class TestTVDBClientSeriesInfo:
@patch.object(TVDBClient, "search_series")
def test_get_series_info(self, mock_search):
mock_search.return_value = {"tvdb_id": 123, "name": "Test", "slug": "test", "year": 2020}
client = TVDBClient()
client.token = "tok"
result = client.get_series_info("Test")
assert result["id"] == 123
assert result["name"] == "Test"
@patch.object(TVDBClient, "search_series")
def test_get_series_info_no_match(self, mock_search):
mock_search.return_value = None
client = TVDBClient()
client.token = "tok"
result = client.get_series_info("NoMatch")
assert result is None
class TestTVDBClientEpisodeDurations:
def test_get_episode_durations(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.base_url = "https://api4.thetvdb.com/v4"
client.token = "tok"
client.headers = {"Content-Type": "application/json"}
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = None
client.cache = mock_cache
client.get_series_info = MagicMock(return_value={"id": 123})
client.get_season_episodes = MagicMock(return_value=[
{"number": 1, "name": "E1", "runtime": 45, "aired": "2020-01-01"},
{"number": 2, "name": "E2", "runtime": 50, "aired": "2020-01-08"},
])
result = client.get_episode_durations("Test", 1)
assert len(result) == 2
assert result[0]["episode_number"] == 1
assert result[0]["runtime"] == 45
mock_cache.cache_episodes.assert_called_once()
def test_get_episode_durations_cached(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.token = "tok"
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = [
{"episode_number": 1, "runtime": 45}
]
client.cache = mock_cache
result = client.get_episode_durations("Test", 1)
assert len(result) == 1
def test_get_episode_durations_no_series(self, tmp_path):
with patch("src.TVDBProvider.tvdb_client.TVDBClient.__init__") as mock_init:
mock_init.return_value = None
client = TVDBClient()
client.token = "tok"
mock_cache = MagicMock()
mock_cache.get_cached_episodes.return_value = None
client.cache = mock_cache
client.get_series_info = MagicMock(return_value=None)
result = client.get_episode_durations("NoShow", 1)
assert result is None
class TestMockTVDBClient:
def test_mock_auth(self):
client = MockTVDBClient()
assert client.authenticate() is True
assert client.authenticated is True
def test_mock_auth_with_key(self):
client = MockTVDBClient()
assert client.authenticate("some_key") is True
def test_mock_episodes_unauthenticated(self):
client = MockTVDBClient()
result = client.get_episode_durations("Show", 1)
assert result is None
def test_mock_episodes_drama(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Game of Thrones", 1)
assert len(result) == 10
def test_mock_episodes_sitcom(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Friends", 1)
assert len(result) == 24
def test_mock_episodes_default(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Unknown Show", 1)
assert len(result) == 22
def test_mock_episode_structure(self):
client = MockTVDBClient()
client.authenticate()
result = client.get_episode_durations("Test", 1)
ep = result[0]
assert ep["episode_number"] == 1
assert ep["runtime"] == 45
assert "name" in ep
assert "aired" in ep
def test_mock_cached_episodes(self):
client = MockTVDBClient()
client.authenticate()
r1 = client.get_episode_durations("Test", 1)
r2 = client.get_episode_durations("Test", 1)
assert r1 is not None and r2 is not None