""" 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 import tempfile import importlib.util from pathlib import Path 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 from modules.file_manager import FileManager from modules.downloader import ContentDownloader from modules.utils import get_ffmpeg_executable, get_twitch_downloader_executable, detect_hardware_acceleration, resolve_hwaccel_type def load_twitch_archive_module(): """Load the main script module for targeted regression tests.""" module_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'twitch-archive.py') spec = importlib.util.spec_from_file_location('twitch_archive_main', module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module 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_rclone_smoke_test_option(self): """Test --rclone-smoke-test option parsing.""" argv = ['--rclone-smoke-test'] 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", "healthcheck", "rclone-smoke-test", "use-chat-downloader-primary", "no-chat-downloader-fallback"] ) self.assertEqual(len(opts), 1) self.assertEqual(opts[0], ('--rclone-smoke-test', '')) def test_healthcheck_option(self): """Test --healthcheck option parsing.""" argv = ['--healthcheck'] 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", "healthcheck", "rclone-smoke-test", "use-chat-downloader-primary", "no-chat-downloader-fallback"] ) self.assertEqual(len(opts), 1) self.assertEqual(opts[0], ('--healthcheck', '')) 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) class TestFileManagerUploadPaths(unittest.TestCase): """Test rclone upload path preparation.""" def test_build_upload_relative_paths_uses_forward_slashes(self): """Rclone --files-from entries must use POSIX separators on Windows.""" with tempfile.TemporaryDirectory() as temp_dir: manager = FileManager( root_path=temp_dir, username='testuser', config={ 'uploadCloud': True, 'uploadPreMergeVideo': True, 'uploadMergedVideo': True, 'uploadChatVideo': True } ) relative_paths = manager._build_upload_relative_paths('20260424_12h00m00s') self.assertTrue(relative_paths) self.assertTrue(all('\\' not in path for path in relative_paths)) self.assertIn('testuser/metadata/METADA_20260424_12h00m00s.json', relative_paths) self.assertIn('testuser/chat/json/CHAT_20260424_12h00m00s.json', relative_paths) class TestDownloaderConfiguration(unittest.TestCase): """Regression tests for downloader config wiring.""" def test_download_vod_method_not_shadowed_by_boolean_flag(self): """Config booleans must not overwrite callable downloader methods.""" downloader = ContentDownloader( twitch_downloader_path='TwitchDownloaderCLI', ffmpeg_path='ffmpeg', config={ 'downloadVOD': True, 'downloadCHAT': True, 'downloadLiveCHAT': True } ) self.assertTrue(callable(downloader.download_vod)) self.assertTrue(downloader.download_vod_enabled) class TestLinuxToolResolution(unittest.TestCase): """Ensure Linux containers prefer their own installed toolchain.""" @patch('modules.utils.shutil.which') def test_linux_prefers_system_ffmpeg(self, mock_which): mock_which.return_value = '/usr/bin/ffmpeg' self.assertEqual(get_ffmpeg_executable('linux'), '/usr/bin/ffmpeg') @patch('modules.utils.shutil.which') def test_linux_prefers_system_twitch_downloader(self, mock_which): mock_which.return_value = '/usr/local/bin/TwitchDownloaderCLI' self.assertEqual(get_twitch_downloader_executable('linux'), '/usr/local/bin/TwitchDownloaderCLI') @patch('modules.utils.os.path.exists', return_value=False) @patch('modules.utils.shutil.which', return_value=None) def test_linux_auto_hwaccel_falls_back_to_software_without_runtime(self, _mock_which, _mock_exists): detected = detect_hardware_acceleration('auto', 'linux') self.assertEqual(detected, 'none') self.assertEqual(resolve_hwaccel_type(detected, 'linux'), 'none') @patch('modules.utils.shutil.which', return_value='/usr/bin/nvidia-smi') def test_linux_auto_hwaccel_uses_nvenc_when_nvidia_runtime_visible(self, _mock_which): detected = detect_hardware_acceleration('auto', 'linux') self.assertEqual(detected, 'nvenc') class TestChatRenderBehavior(unittest.TestCase): """Regression tests for chat rendering defaults and retry behavior.""" @patch('modules.downloader.sys.platform', 'linux') @patch('modules.downloader.subprocess.Popen') def test_linux_chat_render_uses_container_safe_font(self, mock_popen): with tempfile.TemporaryDirectory() as temp_dir: json_path = os.path.join(temp_dir, 'chat.json') video_path = os.path.join(temp_dir, 'chat.mp4') with open(json_path, 'w', encoding='utf-8') as handle: json.dump( { 'comments': [ { 'message': {'body': 'hello world ' * 20}, 'commenter': {'display_name': 'tester'} } ] }, handle ) class FakeProcess: def __init__(self, output_path: str): self.stdout = iter(['rendering']) self.output_path = output_path def wait(self): with open(self.output_path, 'wb') as handle: handle.write(b'x' * 2048) return 0 captured = {} def build_process(cmd, **_kwargs): captured['cmd'] = cmd return FakeProcess(video_path) mock_popen.side_effect = build_process downloader = ContentDownloader('TwitchDownloaderCLI', 'ffmpeg', {'downloadCHAT': True}) result = downloader.render_chat(json_path, video_path, '-c:v libx264 "{save_path}"') self.assertTrue(result) self.assertIn('DejaVu Sans', captured['cmd']) class TestMultiStreamerCleanupRegression(unittest.TestCase): """Regression tests for multi-streamer conversion and cleanup behavior.""" def setUp(self): self.module = load_twitch_archive_module() def _build_archiver(self, temp_dir: str, upload_cloud: bool = True): archiver = MagicMock() archiver.username = 'maddoscientist0' archiver.os_type = 'linux' archiver.quality = 'best' archiver.downloadLiveCHAT = False archiver.downloadCHAT = False archiver.downloadVOD = False archiver.downloadMETADATA = False archiver.mergeVideoChat = False archiver.mergeChatLayout = 'side-by-side' archiver.onlyRaw = False archiver.vodTimeout = 0 archiver.shutdown_requested = False archiver.deleteFiles = True archiver.uploadCloud = upload_cloud archiver.notification_manager = MagicMock() archiver.recorder = MagicMock() archiver.processor = MagicMock() archiver.downloader = MagicMock() archiver.stream_monitor = MagicMock() archiver.file_manager = MagicMock() archiver.file_manager.raw_path = Path(temp_dir) / 'raw' archiver.file_manager.video_path = Path(temp_dir) / 'video' archiver.file_manager.chat_json_path = Path(temp_dir) / 'chat_json' archiver.file_manager.chat_mp4_path = Path(temp_dir) / 'chat' os.makedirs(archiver.file_manager.raw_path, exist_ok=True) os.makedirs(archiver.file_manager.video_path, exist_ok=True) os.makedirs(archiver.file_manager.chat_json_path, exist_ok=True) os.makedirs(archiver.file_manager.chat_mp4_path, exist_ok=True) return archiver def test_process_stream_keeps_raw_when_conversion_fails(self): manager = self.module.TwitchArchiveManager(specific_streamer='maddoscientist0') with tempfile.TemporaryDirectory() as temp_dir: archiver = self._build_archiver(temp_dir) archiver.processor.process_raw_stream.return_value = False archiver.file_manager.upload_to_cloud.return_value = True def write_raw_file(_stream_info, raw_path): with open(raw_path, 'wb') as handle: handle.write(b'x' * 4096) return True archiver.recorder.record.side_effect = write_raw_file stream_info = { 'title': 'Test', 'createdAt': '2026-04-25T09:14:01Z' } manager._process_stream(archiver, stream_info, 'stream-id') archiver.file_manager.clean_raw_file.assert_not_called() archiver.file_manager.delete_local_files.assert_called_once() def test_process_stream_only_deletes_rendered_files_after_real_upload(self): manager = self.module.TwitchArchiveManager(specific_streamer='maddoscientist0') with tempfile.TemporaryDirectory() as temp_dir: archiver = self._build_archiver(temp_dir, upload_cloud=False) archiver.processor.process_raw_stream.return_value = True archiver.file_manager.upload_to_cloud.return_value = True def write_raw_file(_stream_info, raw_path): with open(raw_path, 'wb') as handle: handle.write(b'x' * 4096) return True archiver.recorder.record.side_effect = write_raw_file stream_info = { 'title': 'Test', 'createdAt': '2026-04-25T09:14:01Z' } manager._process_stream(archiver, stream_info, 'stream-id') archiver.file_manager.clean_raw_file.assert_called_once() archiver.file_manager.delete_local_files.assert_not_called() upload_filename_base = archiver.file_manager.upload_to_cloud.call_args.args[0] self.assertFalse(upload_filename_base.startswith('LIVE_')) def test_process_stream_preserves_files_when_chat_render_fails(self): manager = self.module.TwitchArchiveManager(specific_streamer='maddoscientist0') with tempfile.TemporaryDirectory() as temp_dir: archiver = self._build_archiver(temp_dir) archiver.downloadCHAT = True archiver.downloadVOD = True archiver.vodTimeout = 30 archiver.processor.process_raw_stream.return_value = True archiver.file_manager.upload_to_cloud.return_value = True archiver.downloader.last_chat_render_attempted = False archiver.downloader.last_chat_render_succeeded = False def write_raw_file(_stream_info, raw_path): with open(raw_path, 'wb') as handle: handle.write(b'x' * 4096) return True def failed_chat_render(*_args, **_kwargs): archiver.downloader.last_chat_render_attempted = True archiver.downloader.last_chat_render_succeeded = False return False archiver.recorder.record.side_effect = write_raw_file archiver.downloader.download_and_render_chat.side_effect = failed_chat_render archiver.stream_monitor.get_latest_vod.return_value = { 'data': { 'user': { 'videos': { 'edges': [ { 'node': { 'id': '2756589076', 'title': 'Test', 'recordedAt': '2026-04-25T09:14:01Z' } } ] } } } } stream_info = { 'title': 'Test', 'createdAt': '2026-04-25T09:14:01Z' } manager._process_stream(archiver, stream_info, 'stream-id') archiver.file_manager.clean_raw_file.assert_not_called() archiver.file_manager.delete_local_files.assert_not_called() class TestEnvironmentLoadingRegression(unittest.TestCase): """Regression tests for process environment startup in containers.""" def setUp(self): self.module = load_twitch_archive_module() @patch.dict(os.environ, {'CLIENT-ID': 'portainer-client', 'CLIENT-SECRET': 'portainer-secret'}, clear=True) @patch('dotenv.main.find_dotenv', return_value='') @patch('dotenv.main.load_dotenv', return_value=False) def test_load_environment_variables_accepts_process_environment_without_dotenv(self, _mock_load_dotenv, _mock_find_dotenv): archive = self.module.TwitchArchive.__new__(self.module.TwitchArchive) self.module.TwitchArchive._load_environment_variables(archive) class TestHealthcheckRegression(unittest.TestCase): """Regression tests for Docker healthcheck behavior.""" def setUp(self): self.module = load_twitch_archive_module() @patch.object(os, 'getenv', side_effect=lambda name, default=None: default) def test_run_healthcheck_fails_when_heartbeat_is_missing(self, _mock_getenv): config_manager = MagicMock() config_manager.get_all_enabled_streamers.return_value = ['maddoscientist0'] config_manager.load_streamer_config.return_value = {'downloadVOD': False, 'downloadCHAT': False, 'uploadCloud': False} archive = MagicMock() archive.os_type = 'linux' archive.downloadVOD = False archive.downloadCHAT = False archive.uploadCloud = False with patch.object(self.module, 'ConfigManager', return_value=config_manager), \ patch.object(self.module, 'TwitchArchive', return_value=archive), \ patch.object(self.module, 'verify_streamlink', return_value=True), \ patch.object(self.module, 'verify_ffmpeg', return_value=True), \ patch.object(self.module, 'has_fresh_healthcheck_heartbeat', return_value=False): result = self.module.run_healthcheck('maddoscientist0') self.assertEqual(result, 1) @patch.object(os, 'getenv', side_effect=lambda name, default=None: default) def test_run_healthcheck_succeeds_with_fresh_heartbeat(self, _mock_getenv): config_manager = MagicMock() config_manager.get_all_enabled_streamers.return_value = ['maddoscientist0'] config_manager.load_streamer_config.return_value = {'downloadVOD': False, 'downloadCHAT': False, 'uploadCloud': False} archive = MagicMock() archive.os_type = 'linux' archive.downloadVOD = False archive.downloadCHAT = False archive.uploadCloud = False with patch.object(self.module, 'ConfigManager', return_value=config_manager), \ patch.object(self.module, 'TwitchArchive', return_value=archive), \ patch.object(self.module, 'verify_streamlink', return_value=True), \ patch.object(self.module, 'verify_ffmpeg', return_value=True), \ patch.object(self.module, 'has_fresh_healthcheck_heartbeat', return_value=True): result = self.module.run_healthcheck('maddoscientist0') self.assertEqual(result, 0) 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)