137 lines
7.8 KiB
Python
137 lines
7.8 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
# Import without contacting the user's services or starting hardware probes.
|
|
with patch('threading.Thread.start'):
|
|
import app as dashboard
|
|
|
|
|
|
class LanguageTests(unittest.TestCase):
|
|
def probe(self, streams):
|
|
result = SimpleNamespace(returncode=0, stdout=json.dumps({'streams': streams, 'format': {'size': '1024'}}))
|
|
with patch.object(dashboard.subprocess, 'run', return_value=result), patch.object(dashboard, 'external_subtitle_files', return_value=[]):
|
|
return dashboard.run_ffprobe('/media/example.mkv')
|
|
|
|
def test_codes_names_and_regions(self):
|
|
for value, expected in [('GER', 'de'), ('de-DE', 'de'), ('eng_US', 'en'), ('German', 'de'), ('Portuguese', 'pt'), ('dut', 'nl'), ('pol', 'pl'), ('zho-Hans', 'zh'), ('unknown', 'und'), (None, 'und')]:
|
|
with self.subTest(value=value):
|
|
self.assertEqual(dashboard.norm_lang(value)['code'], expected)
|
|
|
|
def test_audio_title_fallback(self):
|
|
info = self.probe([{'codec_type': 'audio', 'tags': {'language': 'und', 'title': 'Deutsch / DTS-HD MA 5.1'}}])
|
|
self.assertEqual(info['audio_languages'][0]['name'], 'Deutsch')
|
|
self.assertEqual(info['audio_details'][0]['language_source'], 'title')
|
|
|
|
def test_uppercase_tags_and_subtitle_titles(self):
|
|
info = self.probe([{'codec_type': 'audio', 'tags': {'LANGUAGE': 'eng'}}, {'codec_type': 'subtitle', 'tags': {'TITLE': 'Português (forced)'}}])
|
|
self.assertEqual(info['audio_languages'][0]['code'], 'en')
|
|
self.assertEqual(info['subtitle_languages'][0]['code'], 'pt')
|
|
|
|
def test_explicit_tag_wins_over_title(self):
|
|
info = self.probe([{'codec_type': 'audio', 'tags': {'language': 'jpn', 'title': 'German commentary'}}])
|
|
self.assertEqual(info['audio_languages'][0]['code'], 'ja')
|
|
|
|
def test_ambiguous_titles_stay_unknown(self):
|
|
for title in ['DTS 5.1', 'German / English', 'Original', 'Commentary']:
|
|
with self.subTest(title=title):
|
|
info = self.probe([{'codec_type': 'audio', 'tags': {'title': title}}])
|
|
self.assertEqual(info['audio_languages'][0]['code'], 'und')
|
|
|
|
def test_languages_are_deduplicated(self):
|
|
info = self.probe([{'codec_type': 'audio', 'tags': {'language': code}} for code in ['de', 'deu', 'ger', 'eng']])
|
|
self.assertEqual([lang['code'] for lang in info['audio_languages']], ['de', 'en'])
|
|
|
|
def test_external_subtitles_only_use_suffix(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
media = Path(tmp) / 'English [2024].mkv'
|
|
media.touch()
|
|
for suffix in ['.pt-BR.forced.srt', '.ger.ass', '.forced.srt']:
|
|
media.with_name(media.stem + suffix).touch()
|
|
self.assertEqual({lang['code'] for lang in dashboard.external_subtitle_languages(media)}, {'pt', 'de'})
|
|
|
|
def test_file_metadata_fallback(self):
|
|
file = {'path': '/remote/movie.mkv', 'mediaInfo': {'audioLanguages': 'German / English'}}
|
|
with patch.object(dashboard, 'cached', return_value={'audio_languages': [dashboard.norm_lang('und')]}), patch.object(dashboard, 'external_subtitle_languages', return_value=[]):
|
|
row = dashboard.media_row('Movie', 2024, file, '/remote', '/local')
|
|
self.assertEqual({lang['code'] for lang in row['audio_languages']}, {'de', 'en'})
|
|
self.assertTrue(all(lang['source'] == 'arr' for lang in row['audio_languages']))
|
|
self.assertEqual(row['state']['label'], 'DE + EN')
|
|
|
|
def test_partial_known_tracks_not_replaced(self):
|
|
file = {'path': '/movie.mkv', 'languages': [{'name': 'German'}]}
|
|
tracks = [dashboard.norm_lang('en'), dashboard.norm_lang('und')]
|
|
with patch.object(dashboard, 'cached', return_value={'audio_languages': tracks}), patch.object(dashboard, 'external_subtitle_languages', return_value=[]):
|
|
row = dashboard.media_row('Movie', 2024, file, '', '')
|
|
self.assertEqual(row['audio_languages'], tracks)
|
|
self.assertEqual(row['state']['label'], 'Sprache prüfen')
|
|
|
|
def test_missing_german_ignores_failed_and_pending_scans(self):
|
|
base = {'missing': False, 'audio_languages': [dashboard.norm_lang('en')]}
|
|
self.assertTrue(dashboard.needs_german_review(base))
|
|
for extra in [{'pending': True}, {'error': 'ffprobe failed'}, {'missing': True}, {'subtitle_languages': [dashboard.norm_lang('de')]}]:
|
|
self.assertFalse(dashboard.needs_german_review({**base, **extra}))
|
|
|
|
def test_english_only_is_not_multilingual(self):
|
|
self.assertEqual(dashboard.language_state([dashboard.norm_lang('en'), dashboard.norm_lang('fr')])['label'], 'Andere')
|
|
|
|
def test_release_separators_and_dl_not_guessed(self):
|
|
self.assertEqual({lang['code'] for lang in dashboard.release_languages('Movie_[GER]_ENG_1080p')}, {'de', 'en'})
|
|
self.assertEqual([lang['code'] for lang in dashboard.release_languages('Movie.German.DL.1080p')], ['de'])
|
|
|
|
def test_original_movie_language_is_not_file_metadata(self):
|
|
self.assertEqual(dashboard.file_metadata_languages({'originalLanguage': {'name': 'German'}}), [])
|
|
|
|
def test_path_mapping_respects_directory_boundary(self):
|
|
self.assertEqual(dashboard.map_path('/data/films/a.mkv', '/data/films', '/local'), '/local/a.mkv')
|
|
self.assertEqual(dashboard.map_path('/data/films-other/a.mkv', '/data/films', '/local'), '/data/films-other/a.mkv')
|
|
|
|
|
|
class CacheTests(unittest.TestCase):
|
|
def test_old_scanner_and_transient_errors_are_retried(self):
|
|
with tempfile.TemporaryDirectory() as tmp, patch.object(dashboard, 'DB_PATH', str(Path(tmp) / 'cache.db')):
|
|
media = Path(tmp) / 'film.mkv'
|
|
media.touch()
|
|
con = dashboard.db()
|
|
def store(version, error=None, age=0):
|
|
data = {'scanner_version': version, 'external_subtitle_files': [], 'audio_languages': []}
|
|
if error:
|
|
data['error'] = error
|
|
con.execute('INSERT OR REPLACE INTO media_cache VALUES(?,?,?,?)', (str(media), media.stat().st_mtime, json.dumps(data), (datetime.now(timezone.utc) - timedelta(seconds=age)).isoformat()))
|
|
con.commit()
|
|
store(dashboard.SCANNER_VERSION - 1)
|
|
self.assertIsNone(dashboard.cached(media))
|
|
store(dashboard.SCANNER_VERSION)
|
|
self.assertIsNotNone(dashboard.cached(media))
|
|
store(dashboard.SCANNER_VERSION, error='timeout')
|
|
self.assertIsNotNone(dashboard.cached(media))
|
|
store(dashboard.SCANNER_VERSION, error='timeout', age=301)
|
|
self.assertIsNone(dashboard.cached(media))
|
|
con.close()
|
|
|
|
def test_missing_file_does_not_loop_scans(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = dashboard.cached(Path(tmp) / 'missing.mkv')
|
|
self.assertFalse(result['pending'])
|
|
self.assertIn('Datei nicht gefunden', result['error'])
|
|
|
|
|
|
class PageTests(unittest.TestCase):
|
|
def test_empty_and_enabled_views_render(self):
|
|
for enabled in [False, True]:
|
|
with self.subTest(enabled=enabled), patch.object(dashboard, 'SONARR_URL', 'http://sonarr' if enabled else ''), patch.object(dashboard, 'SONARR_API_KEY', 'test' if enabled else ''), patch.object(dashboard, 'SAB_URL', 'http://sab' if enabled else ''), patch.object(dashboard, 'SAB_API_KEY', 'test' if enabled else ''):
|
|
response = dashboard.app.test_client().get('/')
|
|
self.assertEqual(response.status_code, 200)
|
|
page = response.get_data(as_text=True)
|
|
self.assertIn('id="sidebar"', page)
|
|
self.assertEqual('id="sonarr"' in page, enabled)
|
|
self.assertEqual('id="downloads"' in page, enabled)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|