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()