From 393cc6729c259f2bf65e17d8e7f28f009dc726d5 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 12 Sep 2025 22:18:25 +0300 Subject: [PATCH] More tests --- app/library/Download.py | 37 +- app/library/YTDLPOpts.py | 2 +- app/tests/test_notifications.py | 1101 +++++++++++++++++++++++++++++++ app/tests/test_tasks.py | 864 ++++++++++++++++++++++++ app/tests/test_ytdlpopts.py | 508 ++++++++++++++ 5 files changed, 2493 insertions(+), 19 deletions(-) create mode 100644 app/tests/test_notifications.py create mode 100644 app/tests/test_tasks.py create mode 100644 app/tests/test_ytdlpopts.py diff --git a/app/library/Download.py b/app/library/Download.py index aa9557ac..8acbdcb8 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -10,6 +10,7 @@ from copy import deepcopy from datetime import UTC, datetime from email.utils import formatdate from pathlib import Path +from typing import Any import yt_dlp.utils @@ -128,10 +129,10 @@ class Download: "Time when the download started." self.queue_time: datetime = datetime.now(tz=UTC) "Time when the download was queued." - self.logs = logs if logs else [] + self.logs: list[str] = logs if logs else [] "Logs from yt-dlp." - def _progress_hook(self, data: dict): + def _progress_hook(self, data: dict) -> None: if self.debug: d_copy: dict = deepcopy(data) for k in ["formats", "thumbnails", "description", "tags", "_format_sort_fields"]: @@ -147,7 +148,7 @@ class Download: } ) - def _postprocessor_hook(self, data: dict): + def _postprocessor_hook(self, data: dict) -> None: if self.debug: d_copy: dict = deepcopy(data) for k in ["formats", "thumbnails", "description", "tags", "_format_sort_fields"]: @@ -159,7 +160,7 @@ class Download: if "__finaldir" in data.get("info_dict", {}) and "filepath" in data.get("info_dict", {}): filename = str(Path(data["info_dict"]["__finaldir"]) / Path(data["info_dict"]["filepath"]).name) else: - filename = data.get("info_dict", {}).get("filepath", data.get("filename")) + filename: str | None = data.get("info_dict", {}).get("filepath", data.get("filename")) self.logger.debug(f"Final filename: '{filename}'.") self.status_queue.put({"id": self.id, "action": "moved", "status": "finished", "final_name": filename}) @@ -168,13 +169,13 @@ class Download: dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields} self.status_queue.put({"id": self.id, "action": "postprocessing", **dataDict, "status": "postprocessing"}) - def post_hooks(self, filename: str | None = None): + def post_hooks(self, filename: str | None = None) -> None: if not filename: return self.status_queue.put({"id": self.id, "filename": filename}) - def _download(self): + def _download(self) -> None: if not self._notify: self._notify = EventBus.get_instance() @@ -227,11 +228,11 @@ class Download: load_cookies(cookie_file) except Exception as e: - err_msg = f"Failed to create cookie file for '{self.info.id}: {self.info.title}'. '{e!s}'." + err_msg: str = f"Failed to create cookie file for '{self.info.id}: {self.info.title}'. '{e!s}'." self.logger.error(err_msg) raise ValueError(err_msg) from e - # Safe-guard incase downloading take too long and the info expires. + # Safe-guard in-case downloading take too long and the info expires. if self.info_dict and isinstance(self.info_dict, dict) and self.download_info_expires > 0: _ts: int | None = self.info_dict.get("epoch", self.info_dict.get("timestamp", None)) _ts = datetime.fromtimestamp(_ts, tz=UTC) if _ts else None @@ -242,14 +243,14 @@ class Download: if not self.info_dict or not isinstance(self.info_dict, dict): self.logger.info(f"Extracting info for '{self.info.url}'.") self.logs = [] - ie_params = params.copy() + ie_params: dict = params.copy() ie_params["callback"] = { "func": lambda _, msg: self.logs.append(msg), "level": logging.WARNING, "name": "callback-logger", } - info = extract_info( + info: dict = extract_info( config=params, url=self.info.url, debug=self.debug, @@ -337,8 +338,8 @@ class Download: f'Task {self.info.name()} preset="{self.info.preset}" cookies="{bool(params.get("cookiefile"))}" completed.' ) - async def start(self): - self.status_queue = Config.get_manager().Queue() + async def start(self) -> None: + self.status_queue: multiprocessing.Queue[Any] = Config.get_manager().Queue() # Create temp dir for each download. if not self.temp_disabled: @@ -479,7 +480,7 @@ class Download: return False - def delete_temp(self, by_pass: bool = False): + def delete_temp(self, by_pass: bool = False) -> None: if self.temp_disabled or self.temp_keep is True or not self.temp_path: return @@ -508,7 +509,7 @@ class Download: else: self.logger.info(f"Temp folder '{self.temp_path}' deletion is {'success' if status else 'failed'}.") - async def _process_status_update(self, status): + async def _process_status_update(self, status) -> None: if status.get("id") != self.id or len(status) < 2: self.logger.warning(f"Received invalid status update. {status}") return @@ -587,7 +588,7 @@ class Download: if not self.final_update or fl: self._notify.emit(Events.ITEM_UPDATED, data=self.info) - async def progress_update(self): + async def progress_update(self) -> None: """ Update status of download task and notify the client. """ @@ -638,11 +639,11 @@ class Download: return False - def __getstate__(self): - state = self.__dict__.copy() + def __getstate__(self) -> dict[str, Any]: + state: dict[str, Any] = self.__dict__.copy() # Exclude (unpickleable) keys during pickling, this issue arise mostly on Windows. - excluded_keys = ("_notify",) + excluded_keys: tuple[str] = ("_notify",) for key in excluded_keys: if key in state: state[key] = None diff --git a/app/library/YTDLPOpts.py b/app/library/YTDLPOpts.py index 4996cbe0..727c5c78 100644 --- a/app/library/YTDLPOpts.py +++ b/app/library/YTDLPOpts.py @@ -44,7 +44,7 @@ class YTDLPOpts: YTDLPOpts: The instance of the class """ - if not args or len(args) < 2 or not isinstance(args, str): + if not args or not isinstance(args, str) or len(args) < 2: return self try: diff --git a/app/tests/test_notifications.py b/app/tests/test_notifications.py new file mode 100644 index 00000000..b67e4103 --- /dev/null +++ b/app/tests/test_notifications.py @@ -0,0 +1,1101 @@ +import json +import tempfile +import uuid +from pathlib import Path +from unittest.mock import AsyncMock, Mock, patch + +import httpx +import pytest +from aiohttp import web + +from app.library.BackgroundWorker import BackgroundWorker +from app.library.encoder import Encoder +from app.library.Events import Event +from app.library.ItemDTO import ItemDTO +from app.library.Notifications import ( + Notification, + NotificationEvents, + Target, + TargetRequest, + TargetRequestHeader, +) + + +class TestTargetRequestHeader: + """Test the TargetRequestHeader dataclass.""" + + def test_target_request_header_creation(self): + """Test creating a TargetRequestHeader.""" + header = TargetRequestHeader(key="Authorization", value="Bearer token123") + + assert header.key == "Authorization" + assert header.value == "Bearer token123" + + def test_target_request_header_serialize(self): + """Test serializing a TargetRequestHeader.""" + header = TargetRequestHeader(key="Content-Type", value="application/json") + serialized = header.serialize() + + expected = {"key": "Content-Type", "value": "application/json"} + assert serialized == expected + + def test_target_request_header_json(self): + """Test JSON serialization of TargetRequestHeader.""" + header = TargetRequestHeader(key="X-API-Key", value="secret123") + json_str = header.json() + + # Should be valid JSON + parsed = json.loads(json_str) + assert parsed["key"] == "X-API-Key" + assert parsed["value"] == "secret123" + + def test_target_request_header_get(self): + """Test get method of TargetRequestHeader.""" + header = TargetRequestHeader(key="Authorization", value="Bearer token") + + assert header.get("key") == "Authorization" + assert header.get("value") == "Bearer token" + assert header.get("nonexistent", "default") == "default" + + +class TestTargetRequest: + """Test the TargetRequest dataclass.""" + + def test_target_request_creation_defaults(self): + """Test creating a TargetRequest with defaults.""" + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + + assert request.type == "json" + assert request.method == "POST" + assert request.url == "https://example.com/webhook" + assert request.headers == [] + assert request.data_key == "data" + + def test_target_request_creation_with_headers(self): + """Test creating a TargetRequest with headers.""" + headers = [ + TargetRequestHeader(key="Authorization", value="Bearer token"), + TargetRequestHeader(key="Content-Type", value="application/json") + ] + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook", + headers=headers, + data_key="payload" + ) + + assert len(request.headers) == 2 + assert request.headers[0].key == "Authorization" + assert request.data_key == "payload" + + def test_target_request_serialize(self): + """Test serializing a TargetRequest.""" + headers = [TargetRequestHeader(key="X-Token", value="abc123")] + request = TargetRequest( + type="json", + method="PUT", + url="https://api.example.com/notify", + headers=headers, + data_key="content" + ) + + serialized = request.serialize() + expected = { + "type": "json", + "method": "PUT", + "url": "https://api.example.com/notify", + "data_key": "content", + "headers": [{"key": "X-Token", "value": "abc123"}] + } + assert serialized == expected + + def test_target_request_json(self): + """Test JSON serialization of TargetRequest.""" + request = TargetRequest( + type="form", + method="POST", + url="https://webhook.site/test" + ) + json_str = request.json() + + parsed = json.loads(json_str) + assert parsed["type"] == "form" + assert parsed["method"] == "POST" + assert parsed["url"] == "https://webhook.site/test" + + def test_target_request_get(self): + """Test get method of TargetRequest.""" + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + + assert request.get("type") == "json" + assert request.get("method") == "POST" + assert request.get("nonexistent", "default") == "default" + + +class TestTarget: + """Test the Target dataclass.""" + + def test_target_creation_minimal(self): + """Test creating a Target with minimal required fields.""" + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Webhook", + request=request + ) + + assert target.name == "Test Webhook" + assert target.on == [] + assert target.presets == [] + assert target.request == request + + def test_target_creation_full(self): + """Test creating a Target with all fields.""" + target_id = str(uuid.uuid4()) + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=target_id, + name="Full Test Webhook", + on=["item_completed", "item_failed"], + presets=["default", "audio_only"], + request=request + ) + + assert target.id == target_id + assert target.name == "Full Test Webhook" + assert target.on == ["item_completed", "item_failed"] + assert target.presets == ["default", "audio_only"] + + def test_target_serialize(self): + """Test serializing a Target.""" + target_id = str(uuid.uuid4()) + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=target_id, + name="Test Target", + on=["item_completed"], + presets=["default"], + request=request + ) + + serialized = target.serialize() + assert serialized["id"] == target_id + assert serialized["name"] == "Test Target" + assert serialized["on"] == ["item_completed"] + assert serialized["presets"] == ["default"] + assert isinstance(serialized["request"], dict) + + def test_target_json(self): + """Test JSON serialization of Target.""" + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="JSON Test", + request=request + ) + + json_str = target.json() + parsed = json.loads(json_str) + assert parsed["name"] == "JSON Test" + + def test_target_get(self): + """Test get method of Target.""" + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Get Test", + request=request + ) + + assert target.get("name") == "Get Test" + assert target.get("nonexistent", "default") == "default" + + +class TestNotificationEvents: + """Test the NotificationEvents class.""" + + def test_notification_events_constants(self): + """Test that NotificationEvents has expected constants.""" + assert hasattr(NotificationEvents, "TEST") + assert hasattr(NotificationEvents, "ITEM_ADDED") + assert hasattr(NotificationEvents, "ITEM_COMPLETED") + assert hasattr(NotificationEvents, "ITEM_CANCELLED") + assert hasattr(NotificationEvents, "ITEM_DELETED") + assert hasattr(NotificationEvents, "LOG_INFO") + assert hasattr(NotificationEvents, "LOG_ERROR") + + def test_get_events(self): + """Test get_events static method.""" + events = NotificationEvents.get_events() + + assert isinstance(events, dict) + assert "TEST" in events + assert "ITEM_ADDED" in events + assert "ITEM_COMPLETED" in events + + def test_events_function(self): + """Test events function.""" + events_list = NotificationEvents.events() + + assert isinstance(events_list, list) + assert len(events_list) > 0 + + def test_is_valid(self): + """Test is_valid static method.""" + # Valid events + assert NotificationEvents.is_valid("test") + assert NotificationEvents.is_valid("item_added") + assert NotificationEvents.is_valid("item_completed") + + # Invalid events + assert not NotificationEvents.is_valid("invalid_event") + assert not NotificationEvents.is_valid("") + assert not NotificationEvents.is_valid(None) + + +class TestNotification: + """Test the Notification singleton class.""" + + def setup_method(self): + """Set up test environment before each test.""" + # Reset singleton instance to ensure clean state + Notification._reset_singleton() + + def teardown_method(self): + """Clean up after each test.""" + # Reset singleton instance to prevent test pollution + Notification._reset_singleton() + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + def test_notification_singleton(self, mock_background_worker, mock_config): + """Test that Notification follows singleton pattern.""" + mock_config.get_instance.return_value = Mock( + debug=False, + config_path="/tmp", + app_version="1.0.0" + ) + mock_background_worker.get_instance.return_value = Mock() + + instance1 = Notification.get_instance() + instance2 = Notification.get_instance() + + assert instance1 is instance2 + assert isinstance(instance1, Notification) + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + def test_notification_init_default_params(self, mock_background_worker, mock_config): + """Test Notification initialization with default parameters.""" + mock_config_instance = Mock( + debug=False, + config_path="/tmp/test", + app_version="1.0.0" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_background_worker.get_instance.return_value = Mock() + + with patch("pathlib.Path.exists", return_value=False): + notification = Notification.get_instance() + + assert notification._debug is False + assert notification._version == "1.0.0" + assert isinstance(notification._targets, list) + assert len(notification._targets) == 0 + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + def test_notification_init_custom_params(self, mock_background_worker, mock_config): + """Test Notification initialization with custom parameters.""" + _ = mock_background_worker, mock_config # Suppress unused variable warnings + mock_config_instance = Mock( + debug=True, + config_path="/custom/path", + app_version="2.0.0" + ) + mock_client = Mock(spec=httpx.AsyncClient) + mock_encoder = Mock(spec=Encoder) + mock_worker = Mock(spec=BackgroundWorker) + + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_path = temp_file.name + + notification = Notification.get_instance( + file=temp_path, + client=mock_client, + encoder=mock_encoder, + config=mock_config_instance, + background_worker=mock_worker + ) + + assert notification._debug is True + assert notification._version == "2.0.0" + assert notification._client == mock_client + assert notification._encoder == mock_encoder + assert notification._offload == mock_worker + + def test_get_targets_empty(self): + """Test get_targets when no targets are loaded.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + targets = notification.get_targets() + + assert isinstance(targets, list) + assert len(targets) == 0 + + def test_clear_targets(self): + """Test clearing notification targets.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + # Add a dummy target + notification._targets = ["dummy_target"] + assert len(notification._targets) == 1 + + # Clear targets + result = notification.clear() + + assert result == notification # Should return self + assert len(notification._targets) == 0 + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + def test_save_targets(self, mock_worker, mock_config): + """Test saving targets to file.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: + temp_path = temp_file.name + + # Create a test target + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Target", + request=request + ) + + notification = Notification.get_instance(file=temp_path) + result = notification.save([target]) + + assert result == notification # Should return self + + # Verify file was written + with open(temp_path) as f: + saved_data = json.load(f) + + assert isinstance(saved_data, list) + assert len(saved_data) == 1 + assert saved_data[0]["name"] == "Test Target" + + # Clean up + Path(temp_path).unlink() + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + def test_load_targets_empty_file(self, mock_worker, mock_config): + """Test loading targets from empty file.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: + temp_file.write("") # Empty file + temp_path = temp_file.name + + notification = Notification.get_instance(file=temp_path) + result = notification.load() + + assert result == notification # Should return self + assert len(notification._targets) == 0 + + # Clean up + Path(temp_path).unlink() + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + @patch("app.library.Notifications.Presets") + def test_load_targets_valid_file(self, mock_presets, mock_worker, mock_config): + """Test loading targets from valid file.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + # Mock preset with name attribute + mock_preset = Mock() + mock_preset.name = "default" + mock_presets.get_instance.return_value.get_all.return_value = [mock_preset] + + target_data = [{ + "id": str(uuid.uuid4()), + "name": "Test Webhook", + "on": ["item_completed"], + "presets": ["default"], + "request": { + "type": "json", + "method": "POST", + "url": "https://example.com/webhook", + "data_key": "data", + "headers": [] + } + }] + + with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file: + json.dump(target_data, temp_file) + temp_path = temp_file.name + + notification = Notification.get_instance(file=temp_path) + result = notification.load() + + assert result == notification # Should return self + assert len(notification._targets) == 1 + assert notification._targets[0].name == "Test Webhook" + + # Clean up + Path(temp_path).unlink() + + def test_make_target(self): + """Test make_target method.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + target_dict = { + "id": str(uuid.uuid4()), + "name": "Test Target", + "on": ["item_completed"], + "presets": ["default"], + "request": { + "type": "json", + "method": "POST", + "url": "https://example.com/webhook", + "data_key": "payload", + "headers": [{"key": "Authorization", "value": "Bearer token"}] + } + } + + target = notification.make_target(target_dict) + + assert isinstance(target, Target) + assert target.name == "Test Target" + assert target.on == ["item_completed"] + assert target.presets == ["default"] + assert target.request.url == "https://example.com/webhook" + assert target.request.data_key == "payload" + assert len(target.request.headers) == 1 + assert target.request.headers[0].key == "Authorization" + + def test_validate_target_valid(self): + """Test validate method with valid target.""" + target_dict = { + "id": str(uuid.uuid4()), + "name": "Valid Target", + "request": { + "url": "https://example.com/webhook" + } + } + + with patch("app.library.Notifications.NotificationEvents.get_events") as mock_events, \ + patch("app.library.Notifications.Presets") as mock_presets: + + mock_events.return_value.values.return_value = ["item_completed"] + mock_presets.get_instance.return_value.get_all.return_value = [] + + result = Notification.validate(target_dict) + assert result is True + + def test_validate_target_missing_id(self): + """Test validate method with missing ID.""" + target_dict = { + "name": "Missing ID Target", + "request": { + "url": "https://example.com/webhook" + } + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."): + Notification.validate(target_dict) + + def test_validate_target_invalid_id(self): + """Test validate method with invalid UUID.""" + target_dict = { + "id": "invalid-uuid", + "name": "Invalid ID Target", + "request": { + "url": "https://example.com/webhook" + } + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. No ID found\."): + Notification.validate(target_dict) + + def test_validate_target_missing_name(self): + """Test validate method with missing name.""" + target_dict = { + "id": str(uuid.uuid4()), + "request": { + "url": "https://example.com/webhook" + } + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. No name found\."): + Notification.validate(target_dict) + + def test_validate_target_missing_request(self): + """Test validate method with missing request.""" + target_dict = { + "id": str(uuid.uuid4()), + "name": "Missing Request Target" + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. No request details found\."): + Notification.validate(target_dict) + + def test_validate_target_missing_url(self): + """Test validate method with missing URL.""" + target_dict = { + "id": str(uuid.uuid4()), + "name": "Missing URL Target", + "request": {} + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. No URL found\."): + Notification.validate(target_dict) + + def test_validate_target_invalid_method(self): + """Test validate method with invalid HTTP method.""" + target_dict = { + "id": str(uuid.uuid4()), + "name": "Invalid Method Target", + "request": { + "url": "https://example.com/webhook", + "method": "GET" # Only POST and PUT are allowed + } + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid method found\."): + Notification.validate(target_dict) + + def test_validate_target_invalid_type(self): + """Test validate method with invalid request type.""" + target_dict = { + "id": str(uuid.uuid4()), + "name": "Invalid Type Target", + "request": { + "url": "https://example.com/webhook", + "type": "xml" # Only json and form are allowed + } + } + + with pytest.raises(ValueError, match=r"Invalid notification target\. Invalid type found\."): + Notification.validate(target_dict) + + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + @patch("app.library.Notifications.EventBus") + def test_attach(self, mock_eventbus, mock_worker, mock_config): + """Test attach method.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + mock_eventbus_instance = Mock() + mock_eventbus.get_instance.return_value = mock_eventbus_instance + + notification = Notification.get_instance() + app = Mock(spec=web.Application) + + with patch.object(notification, "load") as mock_load: + notification.attach(app) + + mock_load.assert_called_once() + mock_eventbus_instance.subscribe.assert_called_once() + + @pytest.mark.asyncio + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + async def test_send_no_targets(self, mock_worker, mock_config): + """Test send method with no targets.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + event = Event( + event="test", + data={"test": "data"} + ) + + result = await notification.send(event) + assert result == [] + + @pytest.mark.asyncio + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + async def test_send_invalid_event_data(self, mock_worker, mock_config): + """Test send method with invalid event data.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + # Add a target + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Target", + request=request + ) + notification._targets = [target] + + event = Event( + event="test", + data="invalid_string_data" # Should be ItemDTO or dict + ) + + result = await notification.send(event) + assert result == [] + + @pytest.mark.asyncio + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + async def test_send_with_http_target(self, mock_worker, mock_config): + """Test send method with HTTP target.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + # Mock HTTP client + mock_response = Mock() + mock_response.status_code = 200 + mock_response.text = "OK" + mock_client = AsyncMock() + mock_client.request = AsyncMock(return_value=mock_response) + + notification = Notification.get_instance(client=mock_client) + + # Add HTTP target + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test HTTP Target", + request=request + ) + notification._targets = [target] + + # Create test event + item_dto = ItemDTO( + id="test_id", + url="https://youtube.com/watch?v=test", + title="Test Video", + folder="/downloads" + ) + event = Event( + event="item_completed", + data=item_dto + ) + + result = await notification.send(event) + + assert len(result) == 1 + assert result[0]["status"] == 200 + assert result[0]["text"] == "OK" + mock_client.request.assert_called_once() + + @pytest.mark.asyncio + @patch("app.library.Notifications.Config") + @patch("app.library.Notifications.BackgroundWorker") + async def test_send_with_apprise_target(self, mock_worker, mock_config): + """Test send method with Apprise target.""" + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0", + apprise_config="/tmp/apprise.conf" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + # Add Apprise target (non-HTTP URL) + request = TargetRequest( + type="json", + method="POST", + url="discord://webhook_id/webhook_token" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Discord Target", + request=request + ) + notification._targets = [target] + + # Create test event + event = Event( + event="item_completed", + data={"test": "data"} + ) + + with patch("app.library.Notifications.Path") as mock_path, \ + patch("builtins.__import__") as mock_import: + + # Mock apprise config file doesn't exist + mock_path.return_value.exists.return_value = False + + # Mock apprise import + mock_apprise = Mock() + mock_notify = Mock() + mock_apprise.Apprise.return_value = mock_notify + mock_import.return_value = mock_apprise + + result = await notification.send(event) + + # Should return empty dict from _apprise method + assert len(result) == 1 + assert result[0] == {} + + def test_check_preset_no_presets(self): + """Test _check_preset method with target having no preset filters.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="No Preset Filter", + presets=[], # No preset filter + request=request + ) + + event = Event( + event="item_completed", + data={"preset": "default"} + ) + + result = notification._check_preset(target, event) + assert result is True # Should pass when no preset filter + + def test_check_preset_with_matching_preset(self): + """Test _check_preset method with matching preset.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Preset Filter", + presets=["default", "audio_only"], + request=request + ) + + # Test with ItemDTO + item_dto = ItemDTO( + id="test_id", + url="https://youtube.com/test", + title="Test Video", + folder="/downloads", + preset="default" + ) + event = Event( + event="item_completed", + data=item_dto + ) + + result = notification._check_preset(target, event) + assert result is True + + def test_check_preset_with_non_matching_preset(self): + """Test _check_preset method with non-matching preset.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Preset Filter", + presets=["audio_only"], + request=request + ) + + event = Event( + event="item_completed", + data={"preset": "video_only"} # Different preset + ) + + result = notification._check_preset(target, event) + assert result is False + + def test_emit_invalid_event(self): + """Test emit method with invalid event.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker_instance = Mock() + mock_worker.get_instance.return_value = mock_worker_instance + + notification = Notification.get_instance() + + # Add a target + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Target", + request=request + ) + notification._targets = [target] + + # Create invalid event + event = Event( + event="invalid_event", + data={"test": "data"} + ) + + with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False): + result = notification.emit(event, None) + + # Should return None and not submit to background worker + assert result is None + mock_worker_instance.submit.assert_not_called() + + def test_emit_valid_event(self): + """Test emit method with valid event.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker_instance = Mock() + mock_worker.get_instance.return_value = mock_worker_instance + + notification = Notification.get_instance() + + # Add a target + request = TargetRequest( + type="json", + method="POST", + url="https://example.com/webhook" + ) + target = Target( + id=str(uuid.uuid4()), + name="Test Target", + request=request + ) + notification._targets = [target] + + # Create valid event + event = Event( + event="item_completed", + data={"test": "data"} + ) + + with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True): + result = notification.emit(event, None) + + # Should return None but submit to background worker + assert result is None + mock_worker_instance.submit.assert_called_once_with(notification.send, event) + + def test_deep_unpack_simple_dict(self): + """Test _deep_unpack method with simple dictionary.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + data = {"key1": "value1", "key2": 123} + result = notification._deep_unpack(data) + + assert result == {"key1": "value1", "key2": 123} + + def test_deep_unpack_nested_dict(self): + """Test _deep_unpack method with nested dictionary.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + data = { + "level1": { + "level2": { + "value": "nested" + } + } + } + result = notification._deep_unpack(data) + + assert result["level1"]["level2"]["value"] == "nested" + + def test_deep_unpack_with_datetime(self): + """Test _deep_unpack method with datetime objects.""" + from datetime import UTC, datetime + + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + test_datetime = datetime.now(tz=UTC) + data = {"timestamp": test_datetime} + result = notification._deep_unpack(data) + + assert result["timestamp"] == test_datetime.isoformat() + + def test_deep_unpack_with_serializable_object(self): + """Test _deep_unpack method with object having serialize method.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + + # Mock object with serialize method + mock_obj = Mock() + mock_obj.serialize.return_value = {"serialized": "data"} + + data = {"object": mock_obj} + result = notification._deep_unpack(data) + + assert result["object"] == {"serialized": "data"} + + @pytest.mark.asyncio + async def test_noop(self): + """Test noop method.""" + with patch("app.library.Notifications.Config") as mock_config, \ + patch("app.library.Notifications.BackgroundWorker") as mock_worker: + + mock_config.get_instance.return_value = Mock( + debug=False, config_path="/tmp", app_version="1.0.0" + ) + mock_worker.get_instance.return_value = Mock() + + notification = Notification.get_instance() + result = await notification.noop() + + assert result is None diff --git a/app/tests/test_tasks.py b/app/tests/test_tasks.py new file mode 100644 index 00000000..71ceed25 --- /dev/null +++ b/app/tests/test_tasks.py @@ -0,0 +1,864 @@ +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from app.library.Tasks import Task, Tasks + + +class TestTask: + """Test the Task dataclass.""" + + def test_task_creation_with_required_fields(self): + """Test creating a task with only required fields.""" + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + assert task.id == "test_id" + assert task.name == "test_task" + assert task.url == "https://example.com/video" + assert task.folder == "" + assert task.preset == "" + assert task.timer == "" + assert task.template == "" + assert task.cli == "" + assert task.auto_start is True + assert task.handler_enabled is True + + def test_task_creation_with_all_fields(self): + """Test creating a task with all fields specified.""" + task = Task( + id="test_id", + name="test_task", + url="https://example.com/video", + folder="test_folder", + preset="test_preset", + timer="0 */6 * * *", + template="%(title)s.%(ext)s", + cli="--extract-flat", + auto_start=False, + handler_enabled=False, + ) + + assert task.id == "test_id" + assert task.name == "test_task" + assert task.url == "https://example.com/video" + assert task.folder == "test_folder" + assert task.preset == "test_preset" + assert task.timer == "0 */6 * * *" + assert task.template == "%(title)s.%(ext)s" + assert task.cli == "--extract-flat" + assert task.auto_start is False + assert task.handler_enabled is False + + def test_task_serialize(self): + """Test task serialization to dictionary.""" + task = Task( + id="test_id", + name="test_task", + url="https://example.com/video", + preset="test_preset", + ) + + serialized = task.serialize() + + assert isinstance(serialized, dict) + assert serialized["id"] == "test_id" + assert serialized["name"] == "test_task" + assert serialized["url"] == "https://example.com/video" + assert serialized["preset"] == "test_preset" + assert serialized["folder"] == "" + assert serialized["auto_start"] is True + + @patch("app.library.Tasks.Encoder") + def test_task_json(self, mock_encoder): + """Test task JSON serialization.""" + mock_encoder_instance = Mock() + mock_encoder_instance.encode.return_value = '{"id": "test_id"}' + mock_encoder.return_value = mock_encoder_instance + + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + json_str = task.json() + + assert json_str == '{"id": "test_id"}' + mock_encoder_instance.encode.assert_called_once() + + def test_task_get_method(self): + """Test task get method for accessing fields.""" + task = Task( + id="test_id", + name="test_task", + url="https://example.com/video", + preset="test_preset", + ) + + assert task.get("id") == "test_id" + assert task.get("name") == "test_task" + assert task.get("preset") == "test_preset" + assert task.get("nonexistent") is None + assert task.get("nonexistent", "default_value") == "default_value" + + @patch("app.library.Tasks.YTDLPOpts") + def test_task_get_ytdlp_opts_without_preset_or_cli(self, mock_ytdlp_opts): + """Test getting YTDLPOpts without preset or CLI options.""" + mock_opts_instance = Mock() + mock_ytdlp_opts.get_instance.return_value = mock_opts_instance + + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + result = task.get_ytdlp_opts() + + assert result == mock_opts_instance + mock_ytdlp_opts.get_instance.assert_called_once() + + @patch("app.library.Tasks.YTDLPOpts") + def test_task_get_ytdlp_opts_with_preset_and_cli(self, mock_ytdlp_opts): + """Test getting YTDLPOpts with preset and CLI options.""" + mock_opts_instance = Mock() + mock_preset_opts = Mock() + mock_final_opts = Mock() + + mock_opts_instance.preset.return_value = mock_preset_opts + mock_preset_opts.add_cli.return_value = mock_final_opts + mock_ytdlp_opts.get_instance.return_value = mock_opts_instance + + task = Task( + id="test_id", + name="test_task", + url="https://example.com/video", + preset="test_preset", + cli="--extract-flat", + ) + + result = task.get_ytdlp_opts() + + assert result == mock_final_opts + mock_opts_instance.preset.assert_called_once_with(name="test_preset") + mock_preset_opts.add_cli.assert_called_once_with("--extract-flat", from_user=True) + + @patch("app.library.Tasks.archive_add") + @patch("app.library.Tasks.extract_info") + def test_task_mark_success(self, mock_extract_info, mock_archive_add): + """Test successful task marking.""" + # Mock extract_info to return playlist data + mock_extract_info.return_value = { + "_type": "playlist", + "entries": [ + { + "_type": "video", + "id": "video1", + "ie_key": "Youtube", + }, + { + "_type": "video", + "id": "video2", + "ie_key": "Youtube", + }, + ], + } + mock_archive_add.return_value = True + + with patch.object(Task, "get_ytdlp_opts") as mock_get_opts: + mock_get_opts.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"} + + task = Task(id="test_id", name="test_task", url="https://example.com/playlist") + + success, message = task.mark() + + assert success is True + assert "marked as downloaded" in message + mock_archive_add.assert_called_once() + # Check that archive_add was called with the correct items + call_args = mock_archive_add.call_args[0] + archive_file = call_args[0] + items = call_args[1] + assert str(archive_file) == "/tmp/archive.txt" + assert "youtube video1" in items + assert "youtube video2" in items + + def test_task_mark_no_url(self): + """Test task marking with no URL.""" + task = Task(id="test_id", name="test_task", url="") + + success, message = task.mark() + + assert success is False + assert "No URL found" in message + + def test_task_mark_no_archive_file(self): + """Test task marking with no archive file configured.""" + with patch.object(Task, "get_ytdlp_opts") as mock_get_opts: + mock_get_opts.return_value.get_all.return_value = {} + + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + success, message = task.mark() + + assert success is False + assert "No archive file found" in message + + @patch("app.library.Tasks.archive_delete") + @patch("app.library.Tasks.extract_info") + def test_task_unmark_success(self, mock_extract_info, mock_archive_delete): + """Test successful task unmarking.""" + # Mock extract_info to return playlist data + mock_extract_info.return_value = { + "_type": "playlist", + "entries": [ + { + "_type": "video", + "id": "video1", + "ie_key": "Youtube", + }, + ], + } + mock_archive_delete.return_value = True + + with patch.object(Task, "get_ytdlp_opts") as mock_get_opts: + mock_get_opts.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"} + + task = Task(id="test_id", name="test_task", url="https://example.com/playlist") + + success, message = task.unmark() + + assert success is True + assert "Removed" in message + assert "items from archive file" in message + mock_archive_delete.assert_called_once() + + @patch("app.library.Tasks.extract_info") + def test_task_mark_logic_invalid_extract_info(self, mock_extract_info): + """Test _mark_logic with invalid extract_info response.""" + mock_extract_info.return_value = None + + with patch.object(Task, "get_ytdlp_opts") as mock_get_opts: + mock_get_opts.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"} + + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + result = task._mark_logic() + + assert isinstance(result, tuple) + success, message = result + assert success is False + assert "Failed to extract information" in message + + @patch("app.library.Tasks.extract_info") + def test_task_mark_logic_not_playlist(self, mock_extract_info): + """Test _mark_logic with non-playlist type.""" + mock_extract_info.return_value = { + "_type": "video", + "id": "video1", + } + + with patch.object(Task, "get_ytdlp_opts") as mock_get_opts: + mock_get_opts.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"} + + task = Task(id="test_id", name="test_task", url="https://example.com/video") + + result = task._mark_logic() + + assert isinstance(result, tuple) + success, message = result + assert success is False + assert "Expected a playlist type" in message + + +class TestTasks: + """Test the Tasks singleton manager.""" + + def setup_method(self): + """Set up test environment before each test.""" + # Reset singleton instance to ensure clean state + Tasks._reset_singleton() + + def teardown_method(self): + """Clean up after each test.""" + # Reset singleton instance to prevent test pollution + Tasks._reset_singleton() + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_singleton(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test that Tasks follows singleton pattern.""" + mock_config.get_instance.return_value = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + instance1 = Tasks.get_instance() + instance2 = Tasks.get_instance() + + assert instance1 is instance2 + assert isinstance(instance1, Tasks) + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_initialization(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test Tasks initialization with proper dependencies.""" + mock_config_instance = Mock( + debug=True, default_preset="test_preset", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus_instance = Mock() + mock_eventbus.get_instance.return_value = mock_eventbus_instance + mock_scheduler_instance = Mock() + mock_scheduler.get_instance.return_value = mock_scheduler_instance + mock_download_queue_instance = Mock() + mock_download_queue.get_instance.return_value = mock_download_queue_instance + + with tempfile.TemporaryDirectory() as temp_dir: + tasks_file = Path(temp_dir) / "tasks.json" + tasks = Tasks(file=tasks_file, config=mock_config_instance) + + # Check initialization + assert tasks._debug is True + assert tasks._default_preset == "test_preset" + assert tasks._file == tasks_file + assert tasks._tasks == [] + assert tasks._scheduler == mock_scheduler_instance + assert tasks._notify == mock_eventbus_instance + assert tasks._downloadQueue == mock_download_queue_instance + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_get_all_empty(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test getting all tasks when no tasks exist.""" + mock_config.get_instance.return_value = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks.get_instance() + + result = tasks.get_all() + + assert result == [] + assert isinstance(result, list) + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_get_by_id_not_found(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test getting task by ID when task doesn't exist.""" + mock_config.get_instance.return_value = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks.get_instance() + + result = tasks.get("nonexistent_id") + + assert result is None + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_load_empty_file(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test loading tasks from empty or non-existent file.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + with tempfile.TemporaryDirectory() as temp_dir: + tasks_file = Path(temp_dir) / "nonexistent.json" + tasks = Tasks(file=tasks_file, config=mock_config_instance) + + result = tasks.load() + + assert result == tasks # Should return self + assert tasks.get_all() == [] + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_load_valid_file(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test loading tasks from valid JSON file.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler_instance = Mock() + mock_scheduler.get_instance.return_value = mock_scheduler_instance + mock_download_queue.get_instance.return_value = Mock() + + with tempfile.TemporaryDirectory() as temp_dir: + tasks_file = Path(temp_dir) / "tasks.json" + + # Create valid JSON file with task data + task_data = [ + { + "id": "task1", + "name": "Test Task 1", + "url": "https://example.com/video1", + "preset": "default", + "timer": "0 6 * * *", + "folder": "", + "template": "", + "cli": "", + "auto_start": True, + "handler_enabled": True, + } + ] + tasks_file.write_text("[\n" + "\n".join(f" {line}" for line in str(task_data).split("\n")) + "\n]") + + # Mock json.loads to return our test data + with patch("json.loads", return_value=task_data): + tasks = Tasks(file=tasks_file, config=mock_config_instance) + + # Mock the scheduler.add method + mock_scheduler_instance.add = Mock() + + result = tasks.load() + + assert result == tasks + loaded_tasks = tasks.get_all() + assert len(loaded_tasks) == 1 + assert loaded_tasks[0].id == "task1" + assert loaded_tasks[0].name == "Test Task 1" + assert loaded_tasks[0].url == "https://example.com/video1" + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_load_invalid_json(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test loading tasks from file with invalid JSON.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + with tempfile.TemporaryDirectory() as temp_dir: + tasks_file = Path(temp_dir) / "tasks.json" + tasks_file.write_text("invalid json {") + + tasks = Tasks(file=tasks_file, config=mock_config_instance) + + result = tasks.load() + + assert result == tasks + assert tasks.get_all() == [] # Should be empty due to invalid JSON + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_save_valid_tasks(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test saving valid tasks to file.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + with tempfile.TemporaryDirectory() as temp_dir: + tasks_file = Path(temp_dir) / "tasks.json" + tasks = Tasks(file=tasks_file, config=mock_config_instance) + + task_data = [ + { + "id": "task1", + "name": "Test Task", + "url": "https://example.com/video", + "preset": "default", + "folder": "", + "template": "", + "cli": "", + "timer": "", + "auto_start": True, + "handler_enabled": True, + } + ] + + result = tasks.save(task_data) + + assert result == tasks + assert tasks_file.exists() + # Verify file content was written + assert tasks_file.stat().st_size > 0 + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_get_by_id_found(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test getting task by ID when task exists.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks(file=None, config=mock_config_instance) + + # Manually add a task to the internal list + test_task = Task(id="test_id", name="Test Task", url="https://example.com/video") + tasks._tasks.append(test_task) + + result = tasks.get("test_id") + + assert result is not None + assert result.id == "test_id" + assert result.name == "Test Task" + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_get_all_with_tasks(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test getting all tasks when tasks exist.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks(file=None, config=mock_config_instance) + + # Manually add tasks to the internal list + task1 = Task(id="task1", name="Task 1", url="https://example.com/video1") + task2 = Task(id="task2", name="Task 2", url="https://example.com/video2") + tasks._tasks.extend([task1, task2]) + + result = tasks.get_all() + + assert len(result) == 2 + assert result[0].id == "task1" + assert result[1].id == "task2" + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_clear_with_scheduler_cleanup(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test clearing tasks with scheduler cleanup.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler_instance = Mock() + mock_scheduler_instance.has.return_value = True + mock_scheduler_instance.remove = Mock() + mock_scheduler.get_instance.return_value = mock_scheduler_instance + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks(file=None, config=mock_config_instance) + + # Add a task with timer to the internal list + task_with_timer = Task(id="task1", name="Task 1", url="https://example.com/video1", timer="0 6 * * *") + tasks._tasks.append(task_with_timer) + + result = tasks.clear() + + assert result == tasks + assert len(tasks.get_all()) == 0 + mock_scheduler_instance.has.assert_called_with("task1") + mock_scheduler_instance.remove.assert_called_with("task1") + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_clear_no_tasks(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test clearing when no tasks exist.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler_instance = Mock() + mock_scheduler.get_instance.return_value = mock_scheduler_instance + mock_download_queue.get_instance.return_value = Mock() + + tasks = Tasks(file=None, config=mock_config_instance) + + result = tasks.clear() + + assert result == tasks + assert len(tasks.get_all()) == 0 + # Should not call scheduler methods when no tasks exist + mock_scheduler_instance.has.assert_not_called() + mock_scheduler_instance.remove.assert_not_called() + + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + def test_tasks_validate_valid_task_dict(self, mock_download_queue, mock_scheduler, mock_eventbus, mock_config): + """Test task validation with valid task dictionary.""" + mock_config.get_instance.return_value = Mock() + mock_eventbus.get_instance.return_value = Mock() + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue.get_instance.return_value = Mock() + + valid_task = { + "name": "Test Task", + "url": "https://example.com/video", + "timer": "0 6 * * *", + "cli": "--write-info-json", + } + + with patch("app.library.Tasks.validate_url") as mock_validate_url: + mock_validate_url.return_value = True + + result = Tasks.validate(valid_task) + + assert result is True + mock_validate_url.assert_called_once_with("https://example.com/video", allow_internal=True) + + def test_tasks_validate_invalid_task_type(self): + """Test task validation with invalid task type.""" + invalid_task = "not a dict or Task" + + with pytest.raises(ValueError, match="Invalid task type"): + Tasks.validate(invalid_task) + + def test_tasks_validate_missing_name(self): + """Test task validation with missing name.""" + invalid_task = { + "url": "https://example.com/video", + } + + with pytest.raises(ValueError, match="No name found"): + Tasks.validate(invalid_task) + + def test_tasks_validate_missing_url(self): + """Test task validation with missing URL.""" + invalid_task = { + "name": "Test Task", + } + + with pytest.raises(ValueError, match="No URL found"): + Tasks.validate(invalid_task) + + def test_tasks_validate_invalid_url(self): + """Test task validation with invalid URL.""" + invalid_task = { + "name": "Test Task", + "url": "invalid-url", + } + + with patch("app.library.Tasks.validate_url") as mock_validate_url: + mock_validate_url.side_effect = ValueError("Invalid URL") + + with pytest.raises(ValueError, match="Invalid URL format"): + Tasks.validate(invalid_task) + + def test_tasks_validate_invalid_timer(self): + """Test task validation with invalid timer format.""" + invalid_task = { + "name": "Test Task", + "url": "https://example.com/video", + "timer": "invalid timer", + } + + with patch("app.library.Tasks.validate_url") as mock_validate_url: + mock_validate_url.return_value = True + + # Import error will be raised by cronsim for invalid format + with patch("cronsim.CronSim") as mock_cronsim: + mock_cronsim.side_effect = Exception("Invalid cron format") + + with pytest.raises(ValueError, match="Invalid timer format"): + Tasks.validate(invalid_task) + + def test_tasks_validate_invalid_cli(self): + """Test task validation with invalid CLI options.""" + invalid_task = { + "name": "Test Task", + "url": "https://example.com/video", + "cli": "invalid --cli options", + } + + with patch("app.library.Tasks.validate_url") as mock_validate_url: + mock_validate_url.return_value = True + + with patch("app.library.Utils.arg_converter") as mock_arg_converter: + mock_arg_converter.side_effect = Exception("Invalid CLI args") + + with pytest.raises(ValueError, match="Invalid command options for yt-dlp"): + Tasks.validate(invalid_task) + + @pytest.mark.asyncio + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + @patch("app.library.Tasks.Item") + @patch("app.library.Tasks.datetime") + @patch("app.library.Tasks.time") + async def test_tasks_runner_success( + self, mock_time, mock_datetime, mock_item, mock_download_queue, mock_scheduler, mock_eventbus, mock_config + ): + """Test successful task runner execution.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus_instance = Mock() + mock_eventbus.get_instance.return_value = mock_eventbus_instance + mock_scheduler.get_instance.return_value = Mock() + + # Mock download queue + mock_download_queue_instance = Mock() + mock_download_queue_instance.add = AsyncMock(return_value={"success": True, "id": "download123"}) + mock_download_queue.get_instance.return_value = mock_download_queue_instance + + # Mock time and datetime + mock_time.time.side_effect = [1000.0, 1005.5] # start and end times + mock_datetime_instance = Mock() + mock_datetime_instance.isoformat.return_value = "2024-01-01T12:00:00Z" + mock_datetime.now.return_value = mock_datetime_instance + + # Mock Item.format + mock_item.format.return_value = {"formatted": "item"} + + tasks = Tasks(file=None, config=mock_config_instance) + + test_task = Task( + id="task1", + name="Test Task", + url="https://example.com/video", + preset="test_preset", + folder="test_folder", + template="test_template", + cli="--write-info-json", + auto_start=True, + ) + + await tasks._runner(test_task) + + # Verify download queue was called + mock_download_queue_instance.add.assert_called_once() + + # Verify Item.format was called with correct parameters + mock_item.format.assert_called_once_with( + { + "url": "https://example.com/video", + "preset": "test_preset", + "folder": "test_folder", + "template": "test_template", + "cli": "--write-info-json", + "auto_start": True, + } + ) + + # Verify events were emitted + assert mock_eventbus_instance.emit.call_count == 2 + + @pytest.mark.asyncio + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + @patch("app.library.Tasks.datetime") + async def test_tasks_runner_no_url( + self, mock_datetime, mock_download_queue, mock_scheduler, mock_eventbus, mock_config + ): + """Test task runner with no URL.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus_instance = Mock() + mock_eventbus.get_instance.return_value = mock_eventbus_instance + mock_scheduler.get_instance.return_value = Mock() + mock_download_queue_instance = Mock() + mock_download_queue.get_instance.return_value = mock_download_queue_instance + + mock_datetime_instance = Mock() + mock_datetime_instance.isoformat.return_value = "2024-01-01T12:00:00Z" + mock_datetime.now.return_value = mock_datetime_instance + + tasks = Tasks(file=None, config=mock_config_instance) + + test_task = Task( + id="task1", + name="Test Task", + url="", # Empty URL + ) + + await tasks._runner(test_task) + + # Verify download queue was NOT called + mock_download_queue_instance.add.assert_not_called() + + @pytest.mark.asyncio + @patch("app.library.Tasks.Config") + @patch("app.library.Tasks.EventBus") + @patch("app.library.Tasks.Scheduler") + @patch("app.library.Tasks.DownloadQueue") + @patch("app.library.Tasks.Item") + @patch("app.library.Tasks.datetime") + async def test_tasks_runner_download_queue_failure( + self, mock_datetime, mock_item, mock_download_queue, mock_scheduler, mock_eventbus, mock_config + ): + """Test task runner when download queue add fails.""" + mock_config_instance = Mock( + debug=False, default_preset="default", config_path="/tmp", tasks_handler_timer="15 */1 * * *" + ) + mock_config.get_instance.return_value = mock_config_instance + mock_eventbus_instance = Mock() + mock_eventbus.get_instance.return_value = mock_eventbus_instance + mock_scheduler.get_instance.return_value = Mock() + + # Mock download queue to raise exception + mock_download_queue_instance = Mock() + mock_download_queue_instance.add = AsyncMock(side_effect=Exception("Queue error")) + mock_download_queue.get_instance.return_value = mock_download_queue_instance + + mock_datetime_instance = Mock() + mock_datetime_instance.isoformat.return_value = "2024-01-01T12:00:00Z" + mock_datetime.now.return_value = mock_datetime_instance + + mock_item.format.return_value = {"formatted": "item"} + + tasks = Tasks(file=None, config=mock_config_instance) + + test_task = Task( + id="task1", + name="Test Task", + url="https://example.com/video", + ) + + await tasks._runner(test_task) + + # Verify error event was emitted (check last call) + calls = mock_eventbus_instance.emit.call_args_list + assert len(calls) > 0 + # Verify the last call was an error event + last_call = calls[-1] + assert "Task failed" in str(last_call) diff --git a/app/tests/test_ytdlpopts.py b/app/tests/test_ytdlpopts.py new file mode 100644 index 00000000..736f1a04 --- /dev/null +++ b/app/tests/test_ytdlpopts.py @@ -0,0 +1,508 @@ +"""Tests for YTDLPOpts class.""" + +from unittest.mock import Mock, patch + +import pytest + +from app.library.Presets import Preset +from app.library.YTDLPOpts import YTDLPOpts + + +class TestYTDLPOpts: + """Test the YTDLPOpts class.""" + + def test_constructor_initializes_correctly(self): + """Test that YTDLPOpts constructor initializes all attributes.""" + with patch("app.library.YTDLPOpts.Config") as mock_config: + mock_config_instance = Mock() + mock_config.get_instance.return_value = mock_config_instance + + opts = YTDLPOpts() + + assert opts._config is mock_config_instance + assert opts._item_opts == {} + assert opts._preset_opts == {} + assert opts._item_cli == [] + assert opts._preset_cli == "" + + def test_get_instance_returns_reset_instance(self): + """Test that get_instance returns a reset YTDLPOpts instance.""" + with patch("app.library.YTDLPOpts.Config"): + opts = YTDLPOpts.get_instance() + + assert isinstance(opts, YTDLPOpts) + assert opts._item_opts == {} + assert opts._preset_opts == {} + assert opts._item_cli == [] + assert opts._preset_cli == "" + + def test_add_cli_with_valid_args(self): + """Test adding valid CLI arguments.""" + with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.arg_converter") as mock_converter: + mock_converter.return_value = {"format": "best"} + opts = YTDLPOpts() + + result = opts.add_cli("--format best", from_user=False) + + assert result is opts # Returns self for chaining + assert "--format best" in opts._item_cli + mock_converter.assert_called_once_with(args="--format best", level=False) + + def test_add_cli_with_invalid_args_raises_error(self): + """Test that invalid CLI arguments raise ValueError.""" + with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.arg_converter") as mock_converter: + mock_converter.side_effect = Exception("Invalid argument") + opts = YTDLPOpts() + + with pytest.raises(ValueError, match="Invalid command options for yt-dlp were given"): + opts.add_cli("--invalid-arg", from_user=True) + + def test_add_cli_with_empty_args_returns_self(self): + """Test that empty or invalid args return self without processing.""" + with patch("app.library.YTDLPOpts.Config"): + opts = YTDLPOpts() + + # Test empty string + result1 = opts.add_cli("", from_user=False) + assert result1 is opts + assert len(opts._item_cli) == 0 + + # Test short string + result2 = opts.add_cli("a", from_user=False) + assert result2 is opts + assert len(opts._item_cli) == 0 + + # Test non-string (this should be handled gracefully) + result3 = opts.add_cli(123, from_user=False) # type: ignore + assert result3 is opts + assert len(opts._item_cli) == 0 + + def test_add_with_valid_config(self): + """Test adding configuration options.""" + with patch("app.library.YTDLPOpts.Config"): + opts = YTDLPOpts() + + config = {"format": "best", "quality": "720p"} + result = opts.add(config, from_user=False) + + assert result is opts + assert opts._item_opts["format"] == "best" + assert opts._item_opts["quality"] == "720p" + + def test_add_with_user_config_filters_bad_options(self): + """Test that user config filters out dangerous options.""" + with ( + patch("app.library.YTDLPOpts.Config"), + patch("app.library.YTDLPOpts.REMOVE_KEYS", [{"paths": "-P, --paths", "outtmpl": "-o, --output"}]), + ): + opts = YTDLPOpts() + + config = { + "format": "best", + "paths": "/dangerous/path", # Should be filtered + "outtmpl": "dangerous_template", # Should be filtered + "quality": "720p", + } + + result = opts.add(config, from_user=True) + + assert result is opts + assert opts._item_opts["format"] == "best" + assert opts._item_opts["quality"] == "720p" + assert "paths" not in opts._item_opts + assert "outtmpl" not in opts._item_opts + + def test_preset_with_valid_preset(self): + """Test applying a valid preset.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.Presets") as mock_presets, + patch("app.library.YTDLPOpts.arg_converter") as mock_converter, + ): + # Mock config + mock_config_instance = Mock() + mock_config_instance.config_path = "/test/config" + mock_config_instance.download_path = "/test/downloads" + mock_config_instance.temp_path = "/test/temp" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config.get_instance.return_value = mock_config_instance + + # Mock preset + mock_preset = Mock(spec=Preset) + mock_preset.id = "test_preset" + mock_preset.name = "Test Preset" + mock_preset.cli = "--format best" + mock_preset.cookies = None + mock_preset.template = "custom_template" + mock_preset.folder = "custom_folder" + + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = mock_preset + mock_presets.get_instance.return_value = mock_presets_instance + + mock_converter.return_value = {"format": "best"} + + with patch("app.library.YTDLPOpts.calc_download_path") as mock_calc_path: + mock_calc_path.return_value = "/test/downloads/custom_folder" + + opts = YTDLPOpts() + result = opts.preset("test_preset") + + assert result is opts + assert opts._preset_cli == "--format best" + assert opts._preset_opts["outtmpl"]["default"] == "custom_template" + assert opts._preset_opts["paths"]["home"] == "/test/downloads/custom_folder" + + def test_preset_with_nonexistent_preset(self): + """Test applying a nonexistent preset returns self.""" + with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.Presets") as mock_presets: + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = None + mock_presets.get_instance.return_value = mock_presets_instance + + opts = YTDLPOpts() + result = opts.preset("nonexistent") + + assert result is opts + assert opts._preset_cli == "" + assert opts._preset_opts == {} + + def test_preset_with_cookies_creates_file(self): + """Test that preset with cookies creates cookie file.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.Presets") as mock_presets, + patch("app.library.YTDLPOpts.load_cookies") as mock_load_cookies, + ): + # Mock config + mock_config_instance = Mock() + mock_config_instance.config_path = "/test/config" + mock_config.get_instance.return_value = mock_config_instance + + # Mock preset with cookies + mock_preset = Mock(spec=Preset) + mock_preset.id = "cookie_preset" + mock_preset.name = "Cookie Preset" + mock_preset.cli = None + mock_preset.cookies = "cookie_data" + mock_preset.template = None + mock_preset.folder = None + + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = mock_preset + mock_presets.get_instance.return_value = mock_presets_instance + + # Use real Path but mock file operations + with ( + patch("pathlib.Path.exists", return_value=False), + patch("pathlib.Path.mkdir"), + patch("pathlib.Path.write_text"), + ): + opts = YTDLPOpts() + opts.preset("cookie_preset") + + # Check that the cookie file would be created at the right path + expected_path = "/test/config/cookies/cookie_preset.txt" + assert opts._preset_opts["cookiefile"] == expected_path + mock_load_cookies.assert_called_once() + + def test_preset_with_invalid_cli_raises_error(self): + """Test that preset with invalid CLI raises ValueError.""" + with ( + patch("app.library.YTDLPOpts.Config"), + patch("app.library.YTDLPOpts.Presets") as mock_presets, + patch("app.library.YTDLPOpts.arg_converter") as mock_converter, + ): + mock_preset = Mock(spec=Preset) + mock_preset.id = "bad_preset" + mock_preset.name = "Bad Preset" + mock_preset.cli = "--invalid-option" + mock_preset.cookies = None + mock_preset.template = None + mock_preset.folder = None + + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = mock_preset + mock_presets.get_instance.return_value = mock_presets_instance + + mock_converter.side_effect = Exception("Invalid CLI") + + opts = YTDLPOpts() + + with pytest.raises(ValueError, match="Invalid preset 'Bad Preset' command options for yt-dlp"): + opts.preset("bad_preset") + + def test_get_all_with_default_options(self): + """Test get_all returns correct default options.""" + with patch("app.library.YTDLPOpts.Config") as mock_config: + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config.get_instance.return_value = mock_config_instance + + with patch("app.library.YTDLPOpts.merge_dict") as mock_merge: + mock_merge.return_value = { + "paths": {"home": "/downloads", "temp": "/temp"}, + "outtmpl": {"default": "default_template", "chapter": "chapter_template"}, + } + + opts = YTDLPOpts() + result = opts.get_all(keep=True) + + assert result["paths"]["home"] == "/downloads" + assert result["outtmpl"]["default"] == "default_template" + + def test_get_all_processes_cli_arguments(self): + """Test get_all processes CLI arguments correctly.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.arg_converter") as mock_converter, + patch("app.library.YTDLPOpts.merge_dict") as mock_merge, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config_instance.get_replacers.return_value = {"home": "/test"} + mock_config.get_instance.return_value = mock_config_instance + + mock_converter.return_value = {"format": "best"} + mock_merge.return_value = {"format": "best"} + + opts = YTDLPOpts() + opts._item_cli = ["--format best"] + opts._preset_cli = "--quality 720p" + + opts.get_all(keep=True) + + # Should join CLI args and process replacers + expected_cli = "--quality 720p\n--format best" + mock_converter.assert_called_once_with(args=expected_cli, level=True) + + def test_get_all_handles_format_special_cases(self): + """Test get_all handles special format values correctly.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.merge_dict") as mock_merge, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config.get_instance.return_value = mock_config_instance + + opts = YTDLPOpts() + + # Test "not_set" format + mock_merge.return_value = {"format": "not_set"} + result = opts.get_all(keep=True) + assert result["format"] is None + + # Test "default" format + mock_merge.return_value = {"format": "default"} + result = opts.get_all(keep=True) + assert result["format"] is None + + # Test "best" format + mock_merge.return_value = {"format": "best"} + result = opts.get_all(keep=True) + assert result["format"] is None + + # Test "-best" format (should remove leading dash) + mock_merge.return_value = {"format": "-best"} + result = opts.get_all(keep=True) + assert result["format"] == "best" + + def test_get_all_with_invalid_cli_raises_error(self): + """Test get_all raises error for invalid CLI arguments.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.arg_converter") as mock_converter, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config_instance.get_replacers.return_value = {} + mock_config.get_instance.return_value = mock_config_instance + + mock_converter.side_effect = Exception("Invalid CLI") + + opts = YTDLPOpts() + opts._item_cli = ["--invalid-arg"] + + with pytest.raises(ValueError, match="Invalid command options for yt-dlp were given"): + opts.get_all() + + def test_get_all_resets_unless_keep_true(self): + """Test get_all resets instance unless keep=True.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.merge_dict") as mock_merge, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config_instance.get_replacers.return_value = {} # Return empty dict + mock_config.get_instance.return_value = mock_config_instance + + mock_merge.return_value = {} + + opts = YTDLPOpts() + opts._item_opts = {"test": "value"} + opts._item_cli = ["--test"] + + # Test with keep=False (default) + opts.get_all(keep=False) + assert opts._item_opts == {} + assert opts._item_cli == [] + + # Reset test data + opts._item_opts = {"test": "value"} + opts._item_cli = ["--test"] + + # Test with keep=True + opts.get_all(keep=True) + assert opts._item_opts == {"test": "value"} + assert opts._item_cli == ["--test"] + + def test_reset_clears_all_options(self): + """Test reset clears all internal state.""" + with patch("app.library.YTDLPOpts.Config"): + opts = YTDLPOpts() + + # Set some state + opts._item_opts = {"format": "best"} + opts._preset_opts = {"quality": "720p"} + opts._item_cli = ["--format best"] + opts._preset_cli = "--quality 720p" + + result = opts.reset() + + assert result is opts + assert opts._item_opts == {} + assert opts._preset_opts == {} + assert opts._item_cli == [] + assert opts._preset_cli == "" + + def test_method_chaining(self): + """Test that methods support chaining.""" + with ( + patch("app.library.YTDLPOpts.Config"), + patch("app.library.YTDLPOpts.arg_converter"), + patch("app.library.YTDLPOpts.Presets") as mock_presets, + ): + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = None + mock_presets.get_instance.return_value = mock_presets_instance + + opts = YTDLPOpts() + + # Test chaining + result = opts.add_cli("--format best").add({"quality": "720p"}).preset("nonexistent").reset() + + assert result is opts + + def test_debug_logging(self): + """Test debug logging when enabled.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.merge_dict") as mock_merge, + patch("app.library.YTDLPOpts.LOG") as mock_log, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = True # Enable debug + mock_config.get_instance.return_value = mock_config_instance + + test_data = {"format": "best"} + mock_merge.return_value = test_data + + opts = YTDLPOpts() + opts.get_all(keep=True) + + mock_log.debug.assert_called_once_with(f"Final yt-dlp options: '{test_data!s}'.") + + def test_cookie_loading_error_handling(self): + """Test error handling when cookie loading fails.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.Presets") as mock_presets, + patch("app.library.YTDLPOpts.load_cookies") as mock_load_cookies, + patch("app.library.YTDLPOpts.LOG") as mock_log, + ): + # Mock config + mock_config_instance = Mock() + mock_config_instance.config_path = "/test/config" + mock_config.get_instance.return_value = mock_config_instance + + # Mock preset with cookies + mock_preset = Mock(spec=Preset) + mock_preset.id = "cookie_preset" + mock_preset.name = "Cookie Preset" + mock_preset.cli = None + mock_preset.cookies = "invalid_cookie_data" + mock_preset.template = None + mock_preset.folder = None + + mock_presets_instance = Mock() + mock_presets_instance.get.return_value = mock_preset + mock_presets.get_instance.return_value = mock_presets_instance + + # Mock cookie loading failure + mock_load_cookies.side_effect = ValueError("Invalid cookies") + + with patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.write_text"): + opts = YTDLPOpts() + opts.preset("cookie_preset") + + # Should log error but not raise + mock_log.error.assert_called_once() + error_args = mock_log.error.call_args[0][0] + assert "Failed to load 'Cookie Preset' cookies" in error_args + + # cookiefile should not be set + assert "cookiefile" not in opts._preset_opts + + def test_replacer_substitution_in_cli(self): + """Test that CLI arguments get replacer substitution.""" + with ( + patch("app.library.YTDLPOpts.Config") as mock_config, + patch("app.library.YTDLPOpts.arg_converter") as mock_converter, + patch("app.library.YTDLPOpts.merge_dict") as mock_merge, + ): + mock_config_instance = Mock() + mock_config_instance.download_path = "/downloads" + mock_config_instance.temp_path = "/temp" + mock_config_instance.output_template = "default_template" + mock_config_instance.output_template_chapter = "chapter_template" + mock_config_instance.debug = False + mock_config_instance.get_replacers.return_value = {"home": "/actual/home", "user": "testuser"} + mock_config.get_instance.return_value = mock_config_instance + + mock_converter.return_value = {"output": "/actual/home/testuser"} + mock_merge.return_value = {} + + opts = YTDLPOpts() + opts._item_cli = ["--output %(home)s/%(user)s"] + + opts.get_all(keep=True) + + # Should replace %(home)s and %(user)s + expected_cli = "--output /actual/home/testuser" + mock_converter.assert_called_once_with(args=expected_cli, level=True)