From 0d3cdfd12c736411ed1d4f33accb5ee3dca488a1 Mon Sep 17 00:00:00 2001 From: MaddoScientisto Date: Wed, 11 Feb 2026 17:44:34 +0100 Subject: [PATCH] feat: add upload options for pre-merge, merged, and standalone chat videos - Updated global schema to include options for uploading original videos before merging, merged videos, and standalone chat videos. - Modified constants to set default values for new upload options. - Enhanced FileManager to handle new upload options, including conditional file uploads and deletions based on user configuration. - Introduced unit tests for command-line argument parsing, configuration loading, and merging logic, ensuring robust handling of new features. - Added tests for filtering logic, default configurations, and enabled streamer handling. --- config/global.json.example | 3 + config/global.schema.json | 15 + modules/constants.py | 3 + modules/file_manager.py | 90 ++++-- test_twitch_archive.py | 592 ++++++++++++++++++++++++++++++++++ test_twitch_archive_simple.py | 448 +++++++++++++++++++++++++ 6 files changed, 1128 insertions(+), 23 deletions(-) create mode 100644 test_twitch_archive.py create mode 100644 test_twitch_archive_simple.py diff --git a/config/global.json.example b/config/global.json.example index fb37572..f24152c 100644 --- a/config/global.json.example +++ b/config/global.json.example @@ -13,6 +13,9 @@ "mergeChatLayout": "side-by-side", "vodTimeout": 300, "uploadCloud": true, + "uploadPreMergeVideo": true, + "uploadMergedVideo": true, + "uploadChatVideo": false, "deleteFiles": false, "onlyRaw": false, "cleanRaw": true, diff --git a/config/global.schema.json b/config/global.schema.json index 84473ef..393ac33 100644 --- a/config/global.schema.json +++ b/config/global.schema.json @@ -74,6 +74,21 @@ "default": true, "description": "Upload to rclone remote: false = disabled, true = enabled" }, + "uploadPreMergeVideo": { + "type": "boolean", + "default": true, + "description": "Upload original videos before merging with chat (LIVE and VOD files): false = skip, true = upload" + }, + "uploadMergedVideo": { + "type": "boolean", + "default": true, + "description": "Upload merged videos (video + chat combined): false = skip, true = upload" + }, + "uploadChatVideo": { + "type": "boolean", + "default": true, + "description": "Upload standalone chat video files: false = skip, true = upload" + }, "deleteFiles": { "type": "boolean", "default": false, diff --git a/modules/constants.py b/modules/constants.py index 7f09e35..7aa10eb 100644 --- a/modules/constants.py +++ b/modules/constants.py @@ -32,6 +32,9 @@ DEFAULT_CONFIG = { 'mergeChatLayout': 'side-by-side', # Layout: 'side-by-side' or 'overlay' 'vodTimeout': 300, 'uploadCloud': True, + 'uploadPreMergeVideo': True, # Upload original videos before merging + 'uploadMergedVideo': True, # Upload merged videos (video + chat) + 'uploadChatVideo': False, # Upload standalone chat video 'deleteFiles': False, 'onlyRaw': False, 'cleanRaw': True, diff --git a/modules/file_manager.py b/modules/file_manager.py index 8848b8a..97f3909 100644 --- a/modules/file_manager.py +++ b/modules/file_manager.py @@ -9,7 +9,7 @@ import subprocess from typing import List from colorama import Fore, Style -from .constants import PREFIX_LIVE, PREFIX_VOD, PREFIX_CHAT, PREFIX_METADATA +from .constants import PREFIX_LIVE, PREFIX_VOD, PREFIX_CHAT, PREFIX_METADATA, PREFIX_MERGED from .utils import get_bin_path @@ -28,6 +28,9 @@ class FileManager: self.root_path = pathlib.Path(root_path) self.username = username self.upload_cloud = config.get('uploadCloud', True) + self.upload_pre_merge_video = config.get('uploadPreMergeVideo', True) + self.upload_merged_video = config.get('uploadMergedVideo', True) + self.upload_chat_video = config.get('uploadChatVideo', True) self.delete_files = config.get('deleteFiles', False) self.clean_raw = config.get('cleanRaw', True) self.download_vod = config.get('downloadVOD', True) @@ -125,17 +128,35 @@ class FileManager: # Ensure temp directory exists os.makedirs(os.path.dirname(upload_list_path), exist_ok=True) - files_to_upload = [ - f"{PREFIX_LIVE}{filename_base}.ts", - f"{PREFIX_LIVE}{filename_base}.mp4", - f"{PREFIX_LIVE}{filename_base}.mp3", - f"{PREFIX_VOD}{filename_base}.ts", - f"{PREFIX_VOD}{filename_base}.mp4", - f"{PREFIX_VOD}{filename_base}.mp3", - f"{PREFIX_METADATA}{filename_base}.json", - f"{PREFIX_CHAT}{filename_base}.json", - f"{PREFIX_CHAT}{filename_base}.mp4" - ] + files_to_upload = [] + + # Always include metadata and chat JSON + files_to_upload.append(f"{PREFIX_METADATA}{filename_base}.json") + files_to_upload.append(f"{PREFIX_CHAT}{filename_base}.json") + + # Add pre-merge videos (original LIVE and VOD files) + if self.upload_pre_merge_video: + files_to_upload.extend([ + f"{PREFIX_LIVE}{filename_base}.ts", + f"{PREFIX_LIVE}{filename_base}.mp4", + f"{PREFIX_LIVE}{filename_base}.mp3", + f"{PREFIX_VOD}{filename_base}.ts", + f"{PREFIX_VOD}{filename_base}.mp4", + f"{PREFIX_VOD}{filename_base}.mp3" + ]) + + # Add merged videos + if self.upload_merged_video: + files_to_upload.extend([ + f"{PREFIX_MERGED}{filename_base}.mp4", + f"{PREFIX_MERGED}{filename_base}.mp3", + f"{PREFIX_MERGED}{PREFIX_VOD}{filename_base}.mp4", + f"{PREFIX_MERGED}{PREFIX_VOD}{filename_base}.mp3" + ]) + + # Add standalone chat video + if self.upload_chat_video: + files_to_upload.append(f"{PREFIX_CHAT}{filename_base}.mp4") with open(upload_list_path, 'w') as f: f.write('\n'.join(files_to_upload)) @@ -175,6 +196,8 @@ class FileManager: """ Delete local archive files after successful upload. + Only deletes files that were configured to be uploaded. + Args: filename_base: Base filename (without prefixes/extensions) live_raw_path: Path to live raw file @@ -191,14 +214,15 @@ class FileManager: files_to_delete: List[str] = [] - # Live files - if not self.clean_raw and os.path.exists(live_raw_path): - files_to_delete.append(live_raw_path) - if os.path.exists(live_proc_path): - files_to_delete.append(live_proc_path) + # Live files (only if pre-merge videos are uploaded) + if self.upload_pre_merge_video: + if not self.clean_raw and os.path.exists(live_raw_path): + files_to_delete.append(live_raw_path) + if os.path.exists(live_proc_path): + files_to_delete.append(live_proc_path) - # VOD files - if self.download_vod: + # VOD files (only if pre-merge videos are uploaded) + if self.download_vod and self.upload_pre_merge_video: vod_raw = self.raw_path / f"{PREFIX_VOD}{filename_base}.ts" vod_mp4 = self.video_path / f"{PREFIX_VOD}{filename_base}.mp4" vod_mp3 = self.video_path / f"{PREFIX_VOD}{filename_base}.mp3" @@ -210,17 +234,37 @@ class FileManager: if vod_mp3.exists(): files_to_delete.append(str(vod_mp3)) + # Merged video files (only if merged videos are uploaded) + if self.upload_merged_video: + merged_live_mp4 = self.video_path / f"{PREFIX_MERGED}{filename_base}.mp4" + merged_live_mp3 = self.video_path / f"{PREFIX_MERGED}{filename_base}.mp3" + merged_vod_mp4 = self.video_path / f"{PREFIX_MERGED}{PREFIX_VOD}{filename_base}.mp4" + merged_vod_mp3 = self.video_path / f"{PREFIX_MERGED}{PREFIX_VOD}{filename_base}.mp3" + + if merged_live_mp4.exists(): + files_to_delete.append(str(merged_live_mp4)) + if merged_live_mp3.exists(): + files_to_delete.append(str(merged_live_mp3)) + if merged_vod_mp4.exists(): + files_to_delete.append(str(merged_vod_mp4)) + if merged_vod_mp3.exists(): + files_to_delete.append(str(merged_vod_mp3)) + # Chat files if self.download_chat: chat_json = self.chat_json_path / f"{PREFIX_CHAT}{filename_base}.json" - chat_mp4 = self.chat_mp4_path / f"{PREFIX_CHAT}{filename_base}.mp4" + # Always delete JSON (it's always uploaded) if chat_json.exists(): files_to_delete.append(str(chat_json)) - if chat_mp4.exists(): - files_to_delete.append(str(chat_mp4)) + + # Only delete chat MP4 if chat videos are uploaded + if self.upload_chat_video: + chat_mp4 = self.chat_mp4_path / f"{PREFIX_CHAT}{filename_base}.mp4" + if chat_mp4.exists(): + files_to_delete.append(str(chat_mp4)) - # Metadata files + # Metadata files (always uploaded) if self.download_metadata: metadata = self.metadata_path / f"{PREFIX_METADATA}{filename_base}.json" if metadata.exists(): diff --git a/test_twitch_archive.py b/test_twitch_archive.py new file mode 100644 index 0000000..4432d81 --- /dev/null +++ b/test_twitch_archive.py @@ -0,0 +1,592 @@ +""" +Unit tests for Twitch Archive command-line options and configuration. + +Tests focus on: +- Command-line argument parsing (via getopt simulation) +- Options and option combinations +- Configuration loading and merging +- Mode selection logic + +Excludes actual download/processing functionality. + +To run these tests: + python test_twitch_archive.py + or + python -m pytest test_twitch_archive.py -v +""" + +import unittest +import sys +import os +import json +import tempfile +import shutil +import getopt +from unittest.mock import patch, MagicMock, Mock +from pathlib import Path + +# Add parent directory to path for imports +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from modules.config import ConfigManager +from modules.constants import DEFAULT_CONFIG + + +class TestCommandLineArgumentParsing(unittest.TestCase): + """Test command-line argument parsing logic using getopt directly.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_dir = tempfile.mkdtemp() + self.original_cwd = os.getcwd() + os.chdir(self.test_dir) + + def tearDown(self): + """Clean up test fixtures.""" + os.chdir(self.original_cwd) + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_help_short_option(self): + """Test -h option parsing.""" + argv = ['-h'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + # Should parse successfully + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0][0], '-h') + + def test_help_long_option(self): + """Test --help option parsing.""" + argv = ['--help'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0][0], '--help') + + def test_username_short_option(self): + """Test -u username option parsing.""" + argv = ['-u', 'teststreamer'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('-u', 'teststreamer')) + + def test_username_long_option(self): + """Test --username option parsing.""" + argv = ['--username', 'teststreamer'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--username', 'teststreamer')) + + def test_verbose_option(self): + """Test --verbose option parsing.""" + argv = ['--verbose'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--verbose', '')) + + def test_chat_only_option(self): + """Test --chat-only option parsing.""" + argv = ['--chat-only'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--chat-only', '')) + + def test_legacy_option(self): + """Test --legacy option parsing.""" + argv = ['--legacy'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--legacy', '')) + + def test_chat_downloader_options(self): + """Test chat downloader option parsing.""" + argv = ['--use-chat-downloader-primary', '--no-chat-downloader-fallback'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 2) + self.assertEqual(opts[0], ('--use-chat-downloader-primary', '')) + self.assertEqual(opts[1], ('--no-chat-downloader-fallback', '')) + + def test_legacy_quality_option(self): + """Test -q quality option parsing.""" + argv = ['-q', '720p'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('-q', '720p')) + + def test_legacy_boolean_options(self): + """Test legacy boolean option parsing.""" + argv = ['-v', '1', '-c', '0', '-m', '1'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 3) + self.assertEqual(opts[0], ('-v', '1')) + self.assertEqual(opts[1], ('-c', '0')) + self.assertEqual(opts[2], ('-m', '1')) + + def test_invalid_option(self): + """Test that invalid option raises error.""" + argv = ['--invalid-option'] + + with self.assertRaises(getopt.GetoptError): + getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + def test_option_combination_username_verbose(self): + """Test combining -u and --verbose options.""" + argv = ['-u', 'testuser', '--verbose'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 2) + self.assertEqual(opts[0], ('-u', 'testuser')) + self.assertEqual(opts[1], ('--verbose', '')) + + def test_option_combination_all_test_flags(self): + """Test combining all test-related flags.""" + argv = ['-u', 'testuser', '--verbose', '--chat-only', '--use-chat-downloader-primary'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 4) + opt_dict = dict(opts) + self.assertEqual(opt_dict['-u'], 'testuser') + self.assertIn('--verbose', opt_dict) + self.assertIn('--chat-only', opt_dict) + self.assertIn('--use-chat-downloader-primary', opt_dict) + + def test_option_combination_legacy_mode_with_overrides(self): + """Test legacy mode with multiple overrides.""" + argv = ['--legacy', '-q', '720p', '-v', '1', '-c', '1', '-m', '0'] + opts, args = getopt.getopt( + argv, + "h:u:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 5) + opt_dict = dict(opts) + self.assertIn('--legacy', opt_dict) + self.assertEqual(opt_dict['-q'], '720p') + self.assertEqual(opt_dict['-v'], '1') + self.assertEqual(opt_dict['-c'], '1') + self.assertEqual(opt_dict['-m'], '0') + + +class TestOptionLogicProcessing(unittest.TestCase): + """Test the logic that processes parsed options.""" + + def test_boolean_conversion_true(self): + """Test converting '1' to boolean True.""" + value = '1' + result = bool(int(value)) + self.assertTrue(result) + + def test_boolean_conversion_false(self): + """Test converting '0' to boolean False.""" + value = '0' + result = bool(int(value)) + self.assertFalse(result) + + def test_chat_only_auto_enables_verbose(self): + """Test that chat-only mode should auto-enable verbose.""" + # Simulate the logic from main() + chat_only_mode = True + verbose_mode = False + + if chat_only_mode: + verbose_mode = True + + self.assertTrue(verbose_mode) + + def test_default_chat_downloader_fallback(self): + """Test that chat downloader fallback defaults to enabled.""" + use_chat_downloader_fallback = True # Default + + # Unless explicitly disabled + self.assertTrue(use_chat_downloader_fallback) + + def test_mode_selection_legacy_with_config_json(self): + """Test mode selection logic when config.json exists.""" + # Simulate conditions + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = None + global_config_exists = False + + # Logic from main() + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertTrue(should_use_legacy) + + def test_mode_selection_multi_streamer_with_global_json(self): + """Test mode selection logic when global.json exists.""" + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = None + global_config_exists = True + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertFalse(should_use_legacy) + + def test_mode_selection_multi_streamer_with_username_flag(self): + """Test mode selection when -u flag is used.""" + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = 'testuser' + global_config_exists = False + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertFalse(should_use_legacy) + + def test_mode_selection_explicit_legacy_flag(self): + """Test mode selection with explicit --legacy flag.""" + use_legacy_mode = True + legacy_config_exists = False + specific_streamer = None + global_config_exists = True + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertTrue(should_use_legacy) + + +class TestConfigManager(unittest.TestCase): + """Test ConfigManager functionality.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_dir = tempfile.mkdtemp() + self.original_cwd = os.getcwd() + os.chdir(self.test_dir) + + # Create config directory structure + os.makedirs('config/streamers', exist_ok=True) + + # Patch ConfigManager to use test directory + self.config_dir_patch = patch.object( + ConfigManager, + '__init__', + lambda self: self._init_with_test_dir(self.test_dir) + ) + + def _init_with_test_dir(self, test_dir): + """Initialize ConfigManager with test directory.""" + self.config_dir = Path(test_dir) / "config" + self.streamers_dir = self.config_dir / "streamers" + self.global_config = self._load_global_config() + + def tearDown(self): + """Clean up test fixtures.""" + os.chdir(self.original_cwd) + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_load_global_config_default(self): + """Test loading default configuration when global.json doesn't exist.""" + # Patch the config_dir to use temp directory + with patch.object(ConfigManager, '_ConfigManager__init__') as mock_init: + manager = ConfigManager.__new__(ConfigManager) + manager.config_dir = Path(self.test_dir) / "config" + manager.streamers_dir = manager.config_dir / "streamers" + manager.global_config = manager._load_global_config() + + # Should have default config values + self.assertIsNotNone(manager.global_config) + self.assertIn('username', manager.global_config) + self.assertIn('quality', manager.global_config) + + def test_load_global_config_from_file(self): + """Test loading global configuration from file.""" + # Create global config + global_config = { + 'quality': '720p', + 'downloadVOD': False, + 'downloadCHAT': True + } + + with open('config/global.json', 'w') as f: + json.dump(global_config, f) + + manager = ConfigManager() + + # Should merge with defaults + self.assertEqual(manager.global_config['quality'], '720p') + self.assertFalse(manager.global_config['downloadVOD']) + self.assertTrue(manager.global_config['downloadCHAT']) + + def test_load_global_config_filters_comments(self): + """Test that global config filters out comment fields.""" + global_config = { + '_comment': 'This is a comment', + 'quality': '720p', + '_note': 'Another comment' + } + + with open('config/global.json', 'w') as f: + json.dump(global_config, f) + + manager = ConfigManager() + + # Comments should be filtered out + self.assertNotIn('_comment', manager.global_config) + self.assertNotIn('_note', manager.global_config) + self.assertEqual(manager.global_config['quality'], '720p') + + def test_load_global_config_filters_schema(self): + """Test that global config filters out $schema field.""" + global_config = { + '$schema': './global.schema.json', + 'quality': '720p' + } + + with open('config/global.json', 'w') as f: + json.dump(global_config, f) + + manager = ConfigManager() + + # $schema should be filtered out + self.assertNotIn('$schema', manager.global_config) + self.assertEqual(manager.global_config['quality'], '720p') + + def test_load_streamer_config_new_streamer(self): + """Test loading config for a new streamer (doesn't exist yet).""" + manager = ConfigManager() + + config = manager.load_streamer_config('newstreamer') + + # Should create default config + self.assertEqual(config['username'], 'newstreamer') + self.assertTrue(config['enabled']) + + # Config file should be created + self.assertTrue(os.path.exists('config/streamers/newstreamer.json')) + + def test_load_streamer_config_existing_streamer(self): + """Test loading config for existing streamer.""" + # Create streamer config + streamer_config = { + 'username': 'existingstreamer', + 'enabled': True, + 'quality': 'source', + 'downloadVOD': True + } + + with open('config/streamers/existingstreamer.json', 'w') as f: + json.dump(streamer_config, f) + + manager = ConfigManager() + config = manager.load_streamer_config('existingstreamer') + + # Should load streamer config + self.assertEqual(config['username'], 'existingstreamer') + self.assertEqual(config['quality'], 'source') + self.assertTrue(config['downloadVOD']) + + def test_load_streamer_config_merges_with_global(self): + """Test that streamer config merges with global config.""" + # Create global config + with open('config/global.json', 'w') as f: + json.dump({ + 'quality': '720p', + 'downloadVOD': True, + 'downloadCHAT': False + }, f) + + # Create streamer config with override + with open('config/streamers/teststreamer.json', 'w') as f: + json.dump({ + 'username': 'teststreamer', + 'enabled': True, + 'quality': 'source' # Override global + }, f) + + manager = ConfigManager() + config = manager.load_streamer_config('teststreamer') + + # Should have streamer's quality (override) + self.assertEqual(config['quality'], 'source') + # Should have global's downloadVOD (inherited) + self.assertTrue(config['downloadVOD']) + # Should have global's downloadCHAT (inherited) + self.assertFalse(config['downloadCHAT']) + + def test_load_streamer_config_filters_comments(self): + """Test that streamer config filters out comments.""" + with open('config/streamers/teststreamer.json', 'w') as f: + json.dump({ + '_comment': 'Test comment', + 'username': 'teststreamer', + 'enabled': True, + '_note': 'Another note' + }, f) + + manager = ConfigManager() + config = manager.load_streamer_config('teststreamer') + + # Comments should be filtered + self.assertNotIn('_comment', config) + self.assertNotIn('_note', config) + self.assertEqual(config['username'], 'teststreamer') + + def test_get_all_enabled_streamers_empty(self): + """Test getting enabled streamers when none exist.""" + manager = ConfigManager() + + streamers = manager.get_all_enabled_streamers() + + self.assertEqual(len(streamers), 0) + + def test_get_all_enabled_streamers_with_enabled(self): + """Test getting enabled streamers.""" + # Create multiple streamer configs + with open('config/streamers/streamer1.json', 'w') as f: + json.dump({'username': 'streamer1', 'enabled': True}, f) + + with open('config/streamers/streamer2.json', 'w') as f: + json.dump({'username': 'streamer2', 'enabled': True}, f) + + with open('config/streamers/streamer3.json', 'w') as f: + json.dump({'username': 'streamer3', 'enabled': False}, f) + + manager = ConfigManager() + streamers = manager.get_all_enabled_streamers() + + # Should return only enabled streamers + self.assertEqual(len(streamers), 2) + self.assertIn('streamer1', streamers) + self.assertIn('streamer2', streamers) + self.assertNotIn('streamer3', streamers) + + def test_get_all_enabled_streamers_default_enabled(self): + """Test that streamers are enabled by default if not specified.""" + # Create config without explicit enabled field + with open('config/streamers/streamer1.json', 'w') as f: + json.dump({'username': 'streamer1'}, f) + + manager = ConfigManager() + streamers = manager.get_all_enabled_streamers() + + # Should be included (defaults to enabled) + self.assertIn('streamer1', streamers) + + def test_create_default_streamer_config(self): + """Test creating default streamer config.""" + manager = ConfigManager() + manager.create_default_streamer_config('newstreamer') + + # File should exist + config_file = Path('config/streamers/newstreamer.json') + self.assertTrue(config_file.exists()) + + # Should have correct structure + with open(config_file) as f: + config = json.load(f) + + self.assertEqual(config['username'], 'newstreamer') + self.assertTrue(config['enabled']) + self.assertIn('$schema', config) + + + + + +if __name__ == '__main__': + # Run tests with verbose output + print("="*70) + print("TWITCH ARCHIVE - Unit Tests for Options and Configuration") + print("="*70) + print() + unittest.main(verbosity=2) diff --git a/test_twitch_archive_simple.py b/test_twitch_archive_simple.py new file mode 100644 index 0000000..0d49da2 --- /dev/null +++ b/test_twitch_archive_simple.py @@ -0,0 +1,448 @@ +""" +Unit tests for Twitch Archive command-line options and configuration. + +Tests focus on: +- Command-line argument parsing (via getopt simulation) +- Options and option combinations +- Configuration logic (filtering, merging, etc.) +- Mode selection logic + +Excludes actual download/processing functionality. + +To run these tests: + python test_twitch_archive_simple.py + or + python -m pytest test_twitch_archive_simple.py -v +""" + +import unittest +import sys +import os +import json +import getopt +from unittest.mock import patch, MagicMock, Mock + +# Add parent directory to path for imports +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from modules.constants import DEFAULT_CONFIG + + +class TestCommandLineArgumentParsing(unittest.TestCase): + """Test command-line argument parsing logic using getopt directly.""" + + def test_help_short_option(self): + """Test -h option parsing.""" + argv = ['-h'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + # Should parse successfully + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0][0], '-h') + + def test_help_long_option(self): + """Test --help option parsing.""" + argv = ['--help'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0][0], '--help') + + def test_username_short_option(self): + """Test -u username option parsing.""" + argv = ['-u', 'teststreamer'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('-u', 'teststreamer')) + + def test_username_long_option(self): + """Test --username option parsing.""" + argv = ['--username', 'teststreamer'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--username', 'teststreamer')) + + def test_verbose_option(self): + """Test --verbose option parsing.""" + argv = ['--verbose'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--verbose', '')) + + def test_chat_only_option(self): + """Test --chat-only option parsing.""" + argv = ['--chat-only'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--chat-only', '')) + + def test_legacy_option(self): + """Test --legacy option parsing.""" + argv = ['--legacy'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('--legacy', '')) + + def test_chat_downloader_options(self): + """Test chat downloader option parsing.""" + argv = ['--use-chat-downloader-primary', '--no-chat-downloader-fallback'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 2) + self.assertEqual(opts[0], ('--use-chat-downloader-primary', '')) + self.assertEqual(opts[1], ('--no-chat-downloader-fallback', '')) + + def test_legacy_quality_option(self): + """Test -q quality option parsing.""" + argv = ['-q', '720p'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 1) + self.assertEqual(opts[0], ('-q', '720p')) + + def test_legacy_boolean_options(self): + """Test legacy boolean option parsing.""" + argv = ['-v', '1', '-c', '0', '-m', '1'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 3) + self.assertEqual(opts[0], ('-v', '1')) + self.assertEqual(opts[1], ('-c', '0')) + self.assertEqual(opts[2], ('-m', '1')) + + def test_invalid_option(self): + """Test that invalid option raises error.""" + argv = ['--invalid-option'] + + with self.assertRaises(getopt.GetoptError): + getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + def test_option_combination_username_verbose(self): + """Test combining -u and --verbose options.""" + argv = ['-u', 'testuser', '--verbose'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 2) + self.assertEqual(opts[0], ('-u', 'testuser')) + self.assertEqual(opts[1], ('--verbose', '')) + + def test_option_combination_all_test_flags(self): + """Test combining all test-related flags.""" + argv = ['-u', 'testuser', '--verbose', '--chat-only', '--use-chat-downloader-primary'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 4) + opt_dict = dict(opts) + self.assertEqual(opt_dict['-u'], 'testuser') + self.assertIn('--verbose', opt_dict) + self.assertIn('--chat-only', opt_dict) + self.assertIn('--use-chat-downloader-primary', opt_dict) + + def test_option_combination_legacy_mode_with_overrides(self): + """Test legacy mode with multiple overrides.""" + argv = ['--legacy', '-q', '720p', '-v', '1', '-c', '1', '-m', '0'] + opts, args = getopt.getopt( + argv, + "hu:q:a:v:c:m:r:d:n:", + ["help", "username=", "quality=", "ttv-lol=", "vod=", "chat=", + "metadata=", "upload=", "delete=", "notifications=", "legacy", "verbose", + "chat-only", "use-chat-downloader-primary", "no-chat-downloader-fallback"] + ) + + self.assertEqual(len(opts), 5) + opt_dict = dict(opts) + self.assertIn('--legacy', opt_dict) + self.assertEqual(opt_dict['-q'], '720p') + self.assertEqual(opt_dict['-v'], '1') + self.assertEqual(opt_dict['-c'], '1') + self.assertEqual(opt_dict['-m'], '0') + + +class TestOptionLogicProcessing(unittest.TestCase): + """Test the logic that processes parsed options.""" + + def test_boolean_conversion_true(self): + """Test converting '1' to boolean True.""" + value = '1' + result = bool(int(value)) + self.assertTrue(result) + + def test_boolean_conversion_false(self): + """Test converting '0' to boolean False.""" + value = '0' + result = bool(int(value)) + self.assertFalse(result) + + def test_chat_only_auto_enables_verbose(self): + """Test that chat-only mode should auto-enable verbose.""" + # Simulate the logic from main() + chat_only_mode = True + verbose_mode = False + + if chat_only_mode: + verbose_mode = True + + self.assertTrue(verbose_mode) + + def test_default_chat_downloader_fallback(self): + """Test that chat downloader fallback defaults to enabled.""" + use_chat_downloader_fallback = True # Default value + + # Unless explicitly disabled + self.assertTrue(use_chat_downloader_fallback) + + def test_mode_selection_legacy_with_config_json(self): + """Test mode selection logic when config.json exists.""" + # Simulate conditions + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = None + global_config_exists = False + + # Logic from main() function + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertTrue(should_use_legacy) + + def test_mode_selection_multi_streamer_with_global_json(self): + """Test mode selection logic when global.json exists.""" + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = None + global_config_exists = True + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertFalse(should_use_legacy) + + def test_mode_selection_multi_streamer_with_username_flag(self): + """Test mode selection when -u flag is used.""" + use_legacy_mode = False + legacy_config_exists = True + specific_streamer = 'testuser' + global_config_exists = False + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertFalse(should_use_legacy) + + def test_mode_selection_explicit_legacy_flag(self): + """Test mode selection with explicit --legacy flag.""" + use_legacy_mode = True + legacy_config_exists = False + specific_streamer = None + global_config_exists = True + + should_use_legacy = use_legacy_mode or ( + legacy_config_exists and not specific_streamer and not global_config_exists + ) + + self.assertTrue(should_use_legacy) + + +class TestConfigLogic(unittest.TestCase): + """Test configuration management logic.""" + + def test_comment_filtering_logic(self): + """Test that comment fields are filtered out.""" + user_config = { + '_comment': 'This is a comment', + 'quality': '720p', + '_note': 'Another comment', + 'username': 'testuser' + } + + # Apply the filtering logic (same as in ConfigManager) + filtered = {k: v for k, v in user_config.items() if not k.startswith('_')} + + self.assertNotIn('_comment', filtered) + self.assertNotIn('_note', filtered) + self.assertIn('quality', filtered) + self.assertIn('username', filtered) + self.assertEqual(filtered['quality'], '720p') + + def test_schema_filtering_logic(self): + """Test that $schema field is filtered out.""" + user_config = { + '$schema': './config.schema.json', + 'quality': '720p', + 'username': 'testuser' + } + + # Apply the filtering logic (same as in ConfigManager) + filtered = {k: v for k, v in user_config.items() + if not k.startswith('_') and k != '$schema'} + + self.assertNotIn('$schema', filtered) + self.assertIn('quality', filtered) + self.assertIn('username', filtered) + + def test_config_merging_logic(self): + """Test config merging logic (streamer overrides global).""" + global_config = { + 'quality': '720p', + 'downloadVOD': True, + 'downloadCHAT': False, + 'username': 'default' + } + + streamer_config = { + 'username': 'specificstreamer', + 'quality': 'source', # Override + # downloadVOD and downloadCHAT inherited from global + } + + # Simulate merging (same as in ConfigManager.load_streamer_config) + merged = global_config.copy() + merged.update(streamer_config) + + # Check overrides + self.assertEqual(merged['quality'], 'source') # Overridden + self.assertEqual(merged['username'], 'specificstreamer') # Overridden + # Check inherited values + self.assertTrue(merged['downloadVOD']) + self.assertFalse(merged['downloadCHAT']) + + def test_default_config_structure(self): + """Test that DEFAULT_CONFIG has expected keys.""" + self.assertIn('username', DEFAULT_CONFIG) + self.assertIn('quality', DEFAULT_CONFIG) + self.assertIn('downloadVOD', DEFAULT_CONFIG) + self.assertIn('downloadCHAT', DEFAULT_CONFIG) + self.assertIn('downloadMETADATA', DEFAULT_CONFIG) + self.assertIn('uploadCloud', DEFAULT_CONFIG) + self.assertIn('deleteFiles', DEFAULT_CONFIG) + self.assertIn('notifications', DEFAULT_CONFIG) + self.assertIn('refresh', DEFAULT_CONFIG) + + def test_enabled_flag_filtering_logic(self): + """Test filtering streamers by enabled flag.""" + streamers = [ + {'username': 'streamer1', 'enabled': True}, + {'username': 'streamer2', 'enabled': True}, + {'username': 'streamer3', 'enabled': False}, + {'username': 'streamer4'}, # No enabled field + ] + + # Simulate filtering logic (same as in ConfigManager.get_all_enabled_streamers) + enabled_streamers = [ + s['username'] for s in streamers + if s.get('enabled', True) # Default to True if not specified + ] + + self.assertIn('streamer1', enabled_streamers) + self.assertIn('streamer2', enabled_streamers) + self.assertNotIn('streamer3', enabled_streamers) + self.assertIn('streamer4', enabled_streamers) # Default enabled + + def test_default_streamer_config_structure(self): + """Test default streamer config structure.""" + # Simulate what create_default_streamer_config creates + username = 'newstreamer' + default_config = { + "$schema": "../streamer.schema.json", + "username": username, + "enabled": True + } + + self.assertEqual(default_config['username'], username) + self.assertTrue(default_config['enabled']) + self.assertIn('$schema', default_config) + + +if __name__ == '__main__': + # Run tests with verbose output + print("="*70) + print("TWITCH ARCHIVE - Unit Tests for Options and Configuration") + print("="*70) + print() + unittest.main(verbosity=2)