diff --git a/.vscode/settings.json b/.vscode/settings.json index de180aac..dc24bfef 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -160,6 +160,7 @@ "rejecttitle", "remux", "reqs", + "restrictfilenames", "RPAREN", "rtime", "rvfc", diff --git a/API.md b/API.md index 9c6486df..9fac609e 100644 --- a/API.md +++ b/API.md @@ -1805,6 +1805,31 @@ or an error: --- +### POST /api/system/check-updates +**Purpose**: Manually trigger a check for application updates from GitHub releases. + +**Response**: +```json +{ + "status": "update_available", + "current_version": "v1.0.14", + "new_version": "v1.0.15" +} +``` + +**Status values**: +- `disabled`: Update checking is disabled in configuration +- `up_to_date`: No updates available +- `update_available`: New version is available +- `error`: Error occurred during check (HTTP error, network issue, etc.) + +- `400 Bad Request` if update checking is disabled in configuration (`check_for_updates: false`). +- `new_version` will be `null` when status is `up_to_date`, `disabled`, or `error`. +- The check runs synchronously and returns the result immediately. +- Results are also stored in config and pushed to frontend via WebSocket. + +--- + ### GET /api/dev/loop **Purpose**: Development-only. Show event loop details and running tasks. diff --git a/FAQ.md b/FAQ.md index c3bed594..5e2ae000 100644 --- a/FAQ.md +++ b/FAQ.md @@ -56,6 +56,7 @@ or the `environment:` section in `compose.yaml` file. | YTP_DEFAULT_PAGINATION | The default number of items per page for history. | `50` | | YTP_TASK_HANDLER_RANDOM_DELAY | The maximum random delay in seconds before starting a task handler. | `60` | | YTP_IGNORE_ARCHIVED_ITEMS | Don't report archived items in the download history. | `false` | +| YTP_CHECK_FOR_UPDATES | Whether to check for application updates. | `true` | > [!NOTE] > To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_`. diff --git a/app/library/Download.py b/app/library/Download.py index 9c5dc2e0..9308b466 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -306,7 +306,7 @@ class Download: raise ValueError(msg) # noqa: TRY301 self.logger.info( - f'Task {self.info.name()}, preset="{self.info.preset}", cookies="{bool(params.get("cookiefile"))}" started.' + f'Download {self.info.name()}, preset="{self.info.preset}", cookies="{bool(params.get("cookiefile"))}" started.' ) if self.debug: @@ -336,7 +336,13 @@ class Download: self.logger.debug(f"Downloading '{self.info.url}' using pre-info.") _dct: dict = self.info_dict.copy() if isinstance(self.info.extras, dict) and len(self.info.extras) > 0: - _dct.update({k: v for k, v in self.info.extras.items() if k not in _dct or not _dct.get(k)}) + _dct.update( + { + k: v + for k, v in self.info.extras.items() + if k not in _dct and v is not None and k not in ("is_live",) + } + ) cls.process_ie_result(ie_result=_dct, download=True) ret: int = cls._download_retcode diff --git a/app/library/UpdateChecker.py b/app/library/UpdateChecker.py new file mode 100644 index 00000000..a6c939e7 --- /dev/null +++ b/app/library/UpdateChecker.py @@ -0,0 +1,217 @@ +import asyncio +import logging +import re +from typing import TYPE_CHECKING + +from aiohttp import web + +from .cache import Cache +from .config import Config +from .Events import EventBus, Events +from .httpx_client import async_client +from .Scheduler import Scheduler +from .Singleton import Singleton +from .version import APP_VERSION + +if TYPE_CHECKING: + from app.library.dl_fields import Any + +LOG: logging.Logger = logging.getLogger("update_checker") + + +class UpdateChecker(metaclass=Singleton): + """ + Checks for application updates from GitHub releases. + """ + + GITHUB_API_URL: str = "https://api.github.com/repos/arabcoders/ytptube/releases/latest" + "GitHub API endpoint for latest release" + + CACHE_DURATION: int = 300 + "Cache duration in seconds (5 minutes)" + + CACHE_KEY: str = "update_checker:result" + "Cache key for storing check results" + + def __init__( + self, config: Config | None = None, scheduler: Scheduler | None = None, notify: EventBus | None = None + ) -> None: + self._config: Config = config or Config.get_instance() + "Instance of Config to use." + + self._scheduler: Scheduler = scheduler or Scheduler.get_instance() + "Instance of Scheduler to use." + + self._notify: EventBus = notify or EventBus.get_instance() + "Instance of EventBus for notifications." + + self._cache: Cache = Cache() + "Instance of Cache for caching check results." + + self._job_id: str | None = None + "Scheduler job ID for update checks." + + @staticmethod + def get_instance( + config: Config | None = None, scheduler: Scheduler | None = None, notify: EventBus | None = None + ) -> "UpdateChecker": + """ + Get the singleton instance of UpdateChecker. + + Args: + config (Config | None): Optional Config instance to use. + scheduler (Scheduler | None): Optional Scheduler instance to use. + notify (EventBus | None): Optional EventBus instance to use. + + Returns: + UpdateChecker: The singleton instance of UpdateChecker. + + """ + return UpdateChecker(config=config, scheduler=scheduler, notify=notify) + + def attach(self, _: web.Application) -> None: + """ + Attach the UpdateChecker to the application. + + Args: + _ (web.Application): The aiohttp web application instance. + + """ + from .Services import Services + + Services.get_instance().add("update_checker", self) + + if not self._config.check_for_updates: + LOG.info("Update checking is disabled.") + return + + if re.search(r"^v\d+", APP_VERSION) is None: + LOG.warning("Not on a release version, skipping update checker setup.") + return + + async def event_handler(_, __): + await self.check_for_updates() + + self._notify.subscribe( + event=Events.STARTED, + callback=event_handler, + name=f"{__class__.__name__}.{__class__.attach.__name__}", + ) + + self._schedule_check() + + async def on_shutdown(self, _: web.Application) -> None: + """ + Handle application shutdown event. + + Args: + _ (web.Application): The aiohttp web application instance. + + """ + if not self._job_id: + return + + self._scheduler.remove(self._job_id) + self._job_id = None + LOG.debug("Stopped update check scheduled task.") + + def _schedule_check(self) -> None: + """Schedule the update check task to run daily at 3 AM.""" + if not self._config.check_for_updates: + LOG.debug("Update checking is disabled, skipping scheduling.") + return + + # Run daily at 3 AM + timer: str = "0 3 * * *" + + self._job_id = self._scheduler.add( + timer=timer, + func=lambda: asyncio.create_task(self.check_for_updates()), + id="update_checker", + ) + + LOG.info(f"Scheduled update check to run daily at 3 AM (cron: {timer}).") + + async def check_for_updates(self) -> tuple[str, str | None]: + """ + Check for updates from GitHub releases. + Updates config.new_version if a newer version is available. + Stops the scheduled task if an update is found. + + Returns: + tuple[str, str | None]: (status, new_version) + status: "disabled", "error", "up_to_date", or "update_available" + new_version: The new version tag if available, None otherwise + + """ + if not self._config.check_for_updates: + LOG.debug("Update checking is disabled, skipping check.") + return ("disabled", None) + + # Check cache + cached = await self._cache.aget(self.CACHE_KEY) + if cached: + ttl = await self._cache.attl(self.CACHE_KEY) + LOG.debug(f"Returning cached result (TTL: {ttl:.0f}s)") + return cached + + try: + LOG.info("Checking for application updates...") + + current_version: str = APP_VERSION.lstrip("v") + + async with async_client(timeout=10.0) as client: + response = await client.get( + self.GITHUB_API_URL, + headers={"Accept": "application/vnd.github+json"}, + ) + + if 200 != response.status_code: + LOG.warning(f"Failed to check for updates: HTTP {response.status_code}") + return ("error", None) + + data: dict[str, Any] = response.json() + + latest_tag: str = data.get("tag_name", "").lstrip("v") + if not latest_tag: + LOG.warning("No tag_name found in GitHub release data.") + return ("error", None) + + if self._compare_versions(current_version, latest_tag): + LOG.warning(f"Update available: {current_version} → {latest_tag}") + new_version_tag = data.get("tag_name", "") + self._config.new_version = new_version_tag + await self.on_shutdown(None) + result = ("update_available", new_version_tag) + await self._cache.aset(self.CACHE_KEY, result, self.CACHE_DURATION) + return result + + LOG.info("No updates available.") + self._config.new_version = "" + result = ("up_to_date", None) + await self._cache.aset(self.CACHE_KEY, result, self.CACHE_DURATION) + return result + except Exception as e: + LOG.exception(e) + LOG.error(f"Error checking for updates: {e!s}") + return ("error", None) + + def _compare_versions(self, current: str, latest: str) -> bool: + """ + Compare version strings to determine if an update is available. + + Args: + current (str): Current version string + latest (str): Latest version string + + Returns: + bool: True if latest > current, False otherwise + + """ + try: + from packaging.version import parse as parse_version + + return parse_version(latest) > parse_version(current) + except Exception as e: + LOG.warning(f"Error comparing versions '{current}' vs '{latest}': {e}") + return False diff --git a/app/library/Utils.py b/app/library/Utils.py index ead7069b..37f541f3 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -96,29 +96,34 @@ class FileLogFormatter(logging.Formatter): return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds") -def get_static_ytdlp(reload: bool = False) -> YTDLP: +def get_ytdlp(params: dict | None = None) -> YTDLP: """ Get a static YTDLP instance for info extraction. Args: - reload (bool): If True, forces re-creation of the instance. + params (dict|None): YTDLP parameters. Returns: YTDLP: A static YTDLP instance. """ global YTDLP_INFO_CLS # noqa: PLW0603 - if YTDLP_INFO_CLS is None or reload: - YTDLP_INFO_CLS = YTDLP( - params={ - "color": "no_color", - "extract_flat": True, - "skip_download": True, - "ignoreerrors": True, - "ignore_no_formats_error": True, - "quiet": True, - } - ) + + default_params: dict[str, Any] = { + "color": "no_color", + "extract_flat": True, + "skip_download": True, + "ignoreerrors": True, + "ignore_no_formats_error": True, + "quiet": True, + } + + if params: + return YTDLP(params=merge_dict(params, default_params)) + + if YTDLP_INFO_CLS is None: + YTDLP_INFO_CLS = YTDLP(params=default_params) + return YTDLP_INFO_CLS @@ -1438,7 +1443,7 @@ def get_archive_id(url: str) -> dict[str, str | None]: "archive_id": None, } - for key, _ie in get_static_ytdlp()._ies.items(): + for key, _ie in get_ytdlp()._ies.items(): try: if not _ie.suitable(url): continue @@ -1937,16 +1942,17 @@ def get_extras(entry: dict, kind: str = "video") -> dict: return extras -def parse_outtmpl(output_template: str, info_dict: dict) -> str: +def parse_outtmpl(output_template: str, info_dict: dict, params: dict | None = None) -> str: """ Parse yt-dlp output template with given info_dict. Args: output_template (str): The output template string. info_dict (dict): The info dictionary from yt-dlp. + params (dict|None): Additional parameters for yt-dlp. Returns: str: The parsed output string. """ - return get_static_ytdlp().prepare_filename(info_dict=info_dict, outtmpl=output_template) + return get_ytdlp(params=params).prepare_filename(info_dict=info_dict, outtmpl=output_template) diff --git a/app/library/config.py b/app/library/config.py index c936416c..13b44051 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -229,6 +229,12 @@ class Config(metaclass=Singleton): static_ui_path: str = "" "The path to the static UI files." + check_for_updates: bool = True + "Check for application updates." + + new_version: str = "" + "The new version available." + _manual_vars: tuple = ( "temp_path", "config_path", @@ -245,6 +251,7 @@ class Config(metaclass=Singleton): "app_commit_sha", "app_build_date", "app_branch", + "new_version", ) "The variables that are immutable." @@ -284,6 +291,7 @@ class Config(metaclass=Singleton): "allow_internal_urls", "simple_mode", "ignore_archived_items", + "check_for_updates", ) "The variables that are booleans." @@ -314,6 +322,8 @@ class Config(metaclass=Singleton): "app_build_date", "app_branch", "default_pagination", + "check_for_updates", + "new_version", ) "The variables that are relevant to the frontend." diff --git a/app/main.py b/app/main.py index 822e9cbe..6fb6130e 100644 --- a/app/main.py +++ b/app/main.py @@ -29,6 +29,7 @@ from app.library.Services import Services from app.library.sqlite_store import SqliteStore from app.library.TaskDefinitions import TaskDefinitions from app.library.Tasks import Tasks +from app.library.UpdateChecker import UpdateChecker LOG = logging.getLogger("app") MIME = magic.Magic(mime=True) @@ -120,6 +121,7 @@ class Main: DLFields.get_instance().attach(self._app) TaskDefinitions.get_instance().attach(self._app) DownloadQueue.get_instance().attach(self._app) + UpdateChecker.get_instance().attach(self._app) EventBus.get_instance().emit( Events.LOADED, diff --git a/app/routes/api/system.py b/app/routes/api/system.py index 79a538b8..c0a5b3a6 100644 --- a/app/routes/api/system.py +++ b/app/routes/api/system.py @@ -11,6 +11,7 @@ from app.library.DownloadQueue import DownloadQueue from app.library.encoder import Encoder from app.library.Events import EventBus, Events from app.library.router import route +from app.library.UpdateChecker import UpdateChecker LOG: logging.Logger = logging.getLogger(__name__) @@ -130,3 +131,50 @@ async def shutdown_system(request: Request, config: Config, encoder: Encoder, no status=web.HTTPOk.status_code, dumps=encoder.encode, ) + + +@route("POST", "api/system/check-updates", "system.check_updates") +async def check_updates(config: Config, encoder: Encoder, update_checker: UpdateChecker) -> Response: + """ + Manually trigger update check. + + Args: + config (Config): The config instance. + encoder (Encoder): The encoder instance. + update_checker (UpdateChecker): The update checker instance. + + Returns: + Response: The response object. + + """ + if not config.check_for_updates: + return web.json_response( + {"error": "Update checking is disabled in configuration."}, + status=web.HTTPBadRequest.status_code, + dumps=encoder.encode, + ) + + # If update already found, return cached result + if config.new_version: + return web.json_response( + data={ + "status": "update_available", + "current_version": config.app_version, + "new_version": config.new_version, + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + # Run check and await result + status, new_version = await update_checker.check_for_updates() + + return web.json_response( + data={ + "status": status, + "current_version": config.app_version, + "new_version": new_version, + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) diff --git a/app/routes/api/tasks.py b/app/routes/api/tasks.py index 983d4d0c..4ed97dca 100644 --- a/app/routes/api/tasks.py +++ b/app/routes/api/tasks.py @@ -2,6 +2,7 @@ import asyncio import functools import logging import uuid +from pathlib import Path from typing import TYPE_CHECKING, Any from aiohttp import web @@ -266,14 +267,16 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R if not task.folder: try: - outtmpl = parse_outtmpl( - output_template=task.get_ytdlp_opts().get_all().get("outtmpl", {}).get("default", "{title} [{id}]"), + ytdlp_opts: dict = task.get_ytdlp_opts().get_all() + outtmpl: str = parse_outtmpl( + output_template=ytdlp_opts.get("outtmpl", {}).get("default", "{title} [{id}]"), info_dict=metadata, + params=ytdlp_opts, ) if outtmpl: - _path = save_path / outtmpl + _path: Path = save_path / outtmpl if not _path.is_dir(): - _path = _path.parent + _path: Path = _path.parent (save_path, _) = get_file(config.download_path, _path.relative_to(config.download_path)) if not str(save_path or "").startswith(str(config.download_path)): @@ -302,6 +305,8 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R data={"error": "Failed to get title from metadata."}, status=web.HTTPBadRequest.status_code ) + LOG.info(f"Generating metadata for task '{task.name}' in '{save_path!s}'") + from yt_dlp.utils import sanitize_filename from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP diff --git a/app/tests/test_conditions.py b/app/tests/test_conditions.py index 94fab479..2021c5e0 100644 --- a/app/tests/test_conditions.py +++ b/app/tests/test_conditions.py @@ -30,16 +30,14 @@ class TestCondition: """Test creating a condition with default values.""" condition = Condition(name="test", filter="duration > 60") - # Check that ID is generated - assert condition.id + assert condition.id, "Check that ID is generated" assert isinstance(condition.id, str) # Check required fields assert condition.name == "test" assert condition.filter == "duration > 60" - # Check defaults - assert condition.cli == "" + assert condition.cli == "", "Check defaults" assert condition.extras == {} assert condition.enabled is True @@ -235,8 +233,7 @@ class TestConditions: assert result is conditions assert len(conditions._items) == 2 - # Check first condition - assert conditions._items[0].name == "short_videos" + assert conditions._items[0].name == "short_videos", "Check first condition" assert conditions._items[0].filter == "duration < 300" assert conditions._items[0].cli == "--format worst" assert conditions._items[0].extras == {"category": "short"} @@ -244,8 +241,7 @@ class TestConditions: assert conditions._items[0].priority == 0 assert conditions._items[0].description == "Download short videos" - # Check second condition - assert conditions._items[1].name == "music_videos" + assert conditions._items[1].name == "music_videos", "Check second condition" assert conditions._items[1].filter == "title ~= 'music'" assert conditions._items[1].enabled is False assert conditions._items[1].priority == 5 @@ -265,8 +261,7 @@ class TestConditions: conditions = Conditions(file=file_path) conditions.load() - # Should have generated ID - assert len(conditions._items) == 1 + assert len(conditions._items) == 1, "Should have generated ID" assert conditions._items[0].id assert conditions._items[0].name == "no_id_test" @@ -286,8 +281,7 @@ class TestConditions: conditions = Conditions(file=file_path) conditions.load() - # Should have generated empty extras - assert len(conditions._items) == 1 + assert len(conditions._items) == 1, "Should have generated empty extras" assert conditions._items[0].extras == {} # Should call save due to changes @@ -306,8 +300,7 @@ class TestConditions: conditions = Conditions(file=file_path) conditions.load() - # Should have generated enabled=True - assert len(conditions._items) == 1 + assert len(conditions._items) == 1, "Should have generated enabled=True" assert conditions._items[0].enabled is True # Should call save due to changes @@ -326,8 +319,7 @@ class TestConditions: conditions = Conditions(file=file_path) conditions.load() - # Should have generated priority=0 - assert len(conditions._items) == 1 + assert len(conditions._items) == 1, "Should have generated priority=0" assert conditions._items[0].priority == 0 # Should call save due to changes @@ -348,8 +340,7 @@ class TestConditions: conditions = Conditions(file=file_path) conditions.load() - # Should have generated description='' - assert len(conditions._items) == 1 + assert len(conditions._items) == 1, "Should have generated description=''" assert conditions._items[0].description == "" # Should call save due to changes @@ -383,8 +374,7 @@ class TestConditions: conditions = Conditions(file=file_path) result = conditions.load() - # Should load only valid conditions - assert result is conditions + assert result is conditions, "Should load only valid conditions" assert len(conditions._items) == 1 assert conditions._items[0].name == "valid" @@ -731,14 +721,11 @@ class TestConditions: test_condition = Condition(name="test", filter="duration > 60") conditions._items = [test_condition] - # Test with None - assert conditions.match(None) is None + assert conditions.match(None) is None, "Test with None" - # Test with empty dict - assert conditions.match({}) is None + assert conditions.match({}) is None, "Test with empty dict" - # Test with non-dict - assert conditions.match("not a dict") is None + assert conditions.match("not a dict") is None, "Test with non-dict" @patch("app.library.conditions.match_str") def test_match_filter_evaluation_error(self, mock_match_str): @@ -774,8 +761,7 @@ class TestConditions: info_dict = {"duration": 150} result = conditions.match(info_dict) - # Should skip disabled condition and match enabled one - assert result is enabled_condition + assert result is enabled_condition, "Should skip disabled condition and match enabled one" # Should only call match_str once for enabled condition mock_match_str.assert_called_once_with("duration > 120", info_dict) @@ -800,8 +786,7 @@ class TestConditions: info_dict = {"duration": 120} result = conditions.match(info_dict) - # Should match high_priority first (priority=10) - assert result is high_priority + assert result is high_priority, "Should match high_priority first (priority=10)" # Should only call match_str once for highest priority condition mock_match_str.assert_called_once_with("duration > 60", info_dict) @@ -891,8 +876,7 @@ class TestConditions: info_dict = {"duration": 120} result = conditions.single_match("disabled_single", info_dict) - # Should return None because condition is disabled - assert result is None + assert result is None, "Should return None because condition is disabled" def test_single_match_invalid_inputs(self): """Test single matching with invalid inputs.""" @@ -900,14 +884,10 @@ class TestConditions: file_path = Path(temp_dir) / "invalid_single_test.json" conditions = Conditions(file=file_path) - # Test with empty conditions - assert conditions.single_match("test", {"duration": 120}) is None + assert conditions.single_match("test", {"duration": 120}) is None, "Test with empty conditions" - # Test with None info - assert conditions.single_match("test", None) is None + assert conditions.single_match("test", None) is None, "Test with None info" - # Test with empty info dict - assert conditions.single_match("test", {}) is None + assert conditions.single_match("test", {}) is None, "Test with empty info dict" - # Test with non-dict info - assert conditions.single_match("test", "not a dict") is None + assert conditions.single_match("test", "not a dict") is None, "Test with non-dict info" diff --git a/app/tests/test_datastore.py b/app/tests/test_datastore.py index 7d25261c..8b34c3e0 100644 --- a/app/tests/test_datastore.py +++ b/app/tests/test_datastore.py @@ -523,8 +523,7 @@ class TestDataStore: assert "id1" in ids assert "id2" in ids - # Verify order is maintained (OrderedDict) - assert result[0][0] == "id1" + assert result[0][0] == "id1", "Verify order is maintained (OrderedDict)" assert result[1][0] == "id2" await db.close() diff --git a/app/tests/test_dl_fields.py b/app/tests/test_dl_fields.py index d888c310..ba7a9e6c 100644 --- a/app/tests/test_dl_fields.py +++ b/app/tests/test_dl_fields.py @@ -71,8 +71,7 @@ class TestDLField: """Test creating a DLField with default values.""" field = DLField(name="test_field", description="Test description", field="--test-option") - # Check that ID is generated - assert field.id + assert field.id, "Check that ID is generated" assert isinstance(field.id, str) # Check required fields @@ -80,8 +79,7 @@ class TestDLField: assert field.description == "Test description" assert field.field == "--test-option" - # Check defaults - assert field.kind == FieldType.TEXT + assert field.kind == FieldType.TEXT, "Check defaults" assert field.icon == "" assert field.order == 0 assert field.value == "" @@ -272,8 +270,7 @@ class TestDLFields: fields = DLFields(file=str(temp_file)) fields.load() - # Should handle error gracefully and return empty list - assert fields.get_all() == [] + assert fields.get_all() == [], "Should handle error gracefully and return empty list" @patch("app.library.dl_fields.Config") def test_dl_fields_load_missing_id_auto_generation(self, mock_config, temp_file): @@ -490,8 +487,7 @@ class TestDLFields: fields.save(test_fields) - # Verify file was written - assert temp_file.exists() + assert temp_file.exists(), "Verify file was written" # Verify content saved_data = json.loads(temp_file.read_text()) diff --git a/app/tests/test_download.py b/app/tests/test_download.py index 010faf51..6c2f24bb 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -55,8 +55,7 @@ class TestNestedLogger: assert levels.count(logging.INFO) == 1 msgs = [r.getMessage() for r in cap.records] assert "[debug]" not in msgs[0] - # [download] prefix is not stripped by NestedLogger - assert msgs[1] == "[download] progress" + assert msgs[1] == "[download] progress", "[download] prefix is not stripped by NestedLogger" assert msgs[2] == "info message" @@ -112,8 +111,7 @@ class TestDownloadHooks: ev = q.items[0] assert ev["id"] == d.id assert ev["action"] == "progress" - # ensure only whitelisted keys included - assert "other" not in ev + assert "other" not in ev, "ensure only whitelisted keys included" for k in ( "tmpfilename", "filename", diff --git a/app/tests/test_ffprobe.py b/app/tests/test_ffprobe.py index 84bf3f28..8d114970 100644 --- a/app/tests/test_ffprobe.py +++ b/app/tests/test_ffprobe.py @@ -55,8 +55,9 @@ class TestFFProbe: """Test that ffprobe results are cached with enhanced async timed_lru_cache.""" from app.library.ffprobe import ffprobe - # Test that the function has been decorated with caching - assert hasattr(ffprobe, "cache_clear"), "ffprobe should have cache_clear method from timed_lru_cache" + assert hasattr(ffprobe, "cache_clear"), ( + "Test that the function has been decorated with caching - ffprobe should have cache_clear method from timed_lru_cache" + ) assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache" # Clear cache to start fresh @@ -91,12 +92,13 @@ class TestFFProbe: assert result2 is not None assert isinstance(result2.metadata, dict) - # The subprocess should not be called again for the actual ffprobe execution - # (it may be called for the -h check, but the main execution should be cached) - assert call_count == first_call_count, "Second call should use cached result" + assert call_count == first_call_count, ( + "The subprocess should not be called again for the actual ffprobe execution (it may be called for the -h check, but the main execution should be cached) - Second call should use cached result" + ) - # Results should be equivalent (same data, may not be same object due to async nature) - assert result1.metadata == result2.metadata + assert result1.metadata == result2.metadata, ( + "Results should be equivalent (same data, may not be same object due to async nature)" + ) @pytest.mark.asyncio async def test_ffprobe_with_path_object(self): @@ -124,8 +126,7 @@ class TestFFProbe: result = FFProbeResult() - # Test empty result - assert result.video == [] + assert result.video == [], "Test empty result" assert result.audio == [] assert result.subtitle == [] assert result.attachment == [] diff --git a/app/tests/test_itemdto.py b/app/tests/test_itemdto.py index c2e7e7fa..e6fa577b 100644 --- a/app/tests/test_itemdto.py +++ b/app/tests/test_itemdto.py @@ -32,8 +32,7 @@ class TestItemFormatAndBasics: item = Item.format(data) assert isinstance(item, Item) - # URL normalized to full YouTube URL - assert item.url.startswith("https://www.youtube.com/watch?v=") + assert item.url.startswith("https://www.youtube.com/watch?v="), "URL normalized to full YouTube URL" assert item.preset == "custom" assert item.folder == "media" assert item.cookies == "abc" @@ -146,8 +145,7 @@ class TestItemDTO: def test_archive_add_and_delete_paths(self): dto = ItemDTO(id="id", title="t", url="u", folder="f") - # Precondition not met yet - assert dto.archive_add() is False + assert dto.archive_add() is False, "Precondition not met yet" # Set up to allow add dto.archive_id = "arch" diff --git a/app/tests/test_notifications.py b/app/tests/test_notifications.py index 36eb8e6b..4fd912f9 100644 --- a/app/tests/test_notifications.py +++ b/app/tests/test_notifications.py @@ -216,13 +216,11 @@ class TestNotificationEvents: def test_is_valid(self): """Test is_valid static method.""" - # Valid events - assert NotificationEvents.is_valid("test") + assert NotificationEvents.is_valid("test"), "Valid events" 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("invalid_event"), "Invalid events" assert not NotificationEvents.is_valid("") assert not NotificationEvents.is_valid(None) @@ -417,8 +415,7 @@ class TestNotification: # Verify save was called due to schema update mock_save.assert_called_once() - # Verify the target has enabled=True by default - assert len(notification._targets) == 1 + assert len(notification._targets) == 1, "Verify the target has enabled=True by default" assert notification._targets[0].enabled is True # Clean up @@ -744,8 +741,7 @@ class TestNotification: result = await notification.send(event) - # Only enabled target should be called - assert len(result) == 1 + assert len(result) == 1, "Only enabled target should be called" assert result[0]["status"] == 200 mock_client.request.assert_called_once() @@ -783,8 +779,7 @@ class TestNotification: result = await notification.send(event) - # Should return empty dict from _apprise method - assert len(result) == 1 + assert len(result) == 1, "Should return empty dict from _apprise method" assert result[0] == {} def test_check_preset_no_presets(self): @@ -881,8 +876,7 @@ class TestNotification: 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 + assert result is None, "Should return None and not submit to background worker" mock_worker_instance.submit.assert_not_called() def test_emit_valid_event(self): @@ -908,8 +902,7 @@ class TestNotification: 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 + assert result is None, "Should return None but submit to background worker" mock_worker_instance.submit.assert_called_once_with(notification.send, event) @pytest.mark.asyncio diff --git a/app/tests/test_presets.py b/app/tests/test_presets.py index 1da4c6dc..65395ff5 100644 --- a/app/tests/test_presets.py +++ b/app/tests/test_presets.py @@ -23,8 +23,7 @@ class TestPreset: assert preset.cookies == "" assert preset.cli == "" assert preset.default is False - # ID should be auto-generated UUID - assert len(preset.id) == 36 # UUID4 string length + assert len(preset.id) == 36, "ID should be auto-generated UUID" # UUID4 string length assert "-" in preset.id def test_preset_creation_with_all_fields(self): @@ -228,8 +227,7 @@ class TestPresets: assert len(presets._items) == 1 loaded_preset = presets._items[0] assert loaded_preset.name == "old_preset" - # Should have migrated format to cli - assert "best[height<=720]" in loaded_preset.cli + assert "best[height<=720]" in loaded_preset.cli, "Should have migrated format to cli" assert "--format" in loaded_preset.cli # Should have generated ID assert loaded_preset.id is not None @@ -322,8 +320,7 @@ class TestPresets: all_presets = presets.get_all() - # Should include both default and custom presets - assert len(all_presets) > 1 + assert len(all_presets) > 1, "Should include both default and custom presets" assert any(p.name == "custom" for p in all_presets) assert any(p.default is True for p in all_presets) @@ -462,7 +459,7 @@ class TestPresets: # Note: The actual chmod might fail in test environment, # but we're testing that the code attempts it - assert presets is not None + assert presets is not None, "but we're testing that the code attempts it" def test_default_presets_validation(self): """Test that all default presets are valid.""" @@ -519,8 +516,7 @@ class TestPresets: with patch.object(presets, "get", return_value=None): # The actual config instance stored in presets should be checked presets.attach(mock_app) - # Should have reset default_preset to "default" - assert presets._config.default_preset == "default" + assert presets._config.default_preset == "default", "Should have reset default_preset to 'default'" @pytest.mark.asyncio @patch("app.library.Presets.Config") @@ -567,8 +563,7 @@ class TestPresets: presets = Presets(file=presets_file, config=mock_config_instance) presets.load() - # Should only load the valid preset - assert len(presets._items) == 1 + assert len(presets._items) == 1, "Should only load the valid preset" assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'" # Should have logged an error for the invalid preset @@ -658,8 +653,7 @@ class TestPresets: all_presets = presets.get_all() non_default = [p for p in all_presets if not p.default] - # Should be sorted by priority descending - assert non_default[0].name == "high" + assert non_default[0].name == "high", "Should be sorted by priority descending" assert non_default[0].priority == 10 assert non_default[1].name == "medium" assert non_default[1].priority == 5 diff --git a/app/tests/test_scheduler.py b/app/tests/test_scheduler.py index 2f38e451..08c47352 100644 --- a/app/tests/test_scheduler.py +++ b/app/tests/test_scheduler.py @@ -130,8 +130,7 @@ class TestScheduler: assert sched.has("job1") is True new_job = sched.get("job1") assert new_job is not old - # Old job should have been stopped via remove() - assert old.stopped is True + assert old.stopped is True, "Old job should have been stopped via remove()" @patch("app.library.Scheduler.Cron", new=DummyCron) def test_remove_single_job_success(self) -> None: @@ -154,8 +153,7 @@ class TestScheduler: result = sched.remove("jobB") assert result is False - # Job should remain since stop failed - assert sched.has("jobB") is True + assert sched.has("jobB") is True, "Job should remain since stop failed" @patch("app.library.Scheduler.Cron", new=DummyCron) def test_remove_list_of_jobs(self) -> None: @@ -193,8 +191,9 @@ class TestScheduler: sched = Scheduler() sched.attach(app) - # on_shutdown handler should be registered - assert Scheduler.on_shutdown in [cb.__func__ if hasattr(cb, "__func__") else cb for cb in app.on_shutdown] + assert Scheduler.on_shutdown in [cb.__func__ if hasattr(cb, "__func__") else cb for cb in app.on_shutdown], ( + "on_shutdown handler should be registered" + ) # Patch add to verify it is called from event handler add_spy = MagicMock(wraps=sched.add) diff --git a/app/tests/test_subtitle.py b/app/tests/test_subtitle.py index fa1f0d52..de5ce860 100644 --- a/app/tests/test_subtitle.py +++ b/app/tests/test_subtitle.py @@ -13,12 +13,13 @@ class TestMsToTimestamp: assert ms_to_timestamp(9) == "0:00:00.00" assert ms_to_timestamp(10) == "0:00:00.01" assert ms_to_timestamp(12345) == "0:00:12.34" - # 1 hour, 2 minutes, 3 seconds - assert ms_to_timestamp(3600000 + 120000 + 3000) == "1:02:03.00" - # Over 10 hours (SubStation limit is < 10h, our override must exceed) - assert ms_to_timestamp(12 * 3600000 + 34 * 60000 + 56 * 1000 + 780) == "12:34:56.78" - # Well over 36 hours to ensure no clamping at 9:59:59.99 - assert ms_to_timestamp(37 * 3600000 + 12 * 60000 + 34 * 1000 + 560) == "37:12:34.56" + assert ms_to_timestamp(3600000 + 120000 + 3000) == "1:02:03.00", "1 hour, 2 minutes, 3 seconds" + assert ms_to_timestamp(12 * 3600000 + 34 * 60000 + 56 * 1000 + 780) == "12:34:56.78", ( + "Over 10 hours (SubStation limit is < 10h, our override must exceed)" + ) + assert ms_to_timestamp(37 * 3600000 + 12 * 60000 + 34 * 1000 + 560) == "37:12:34.56", ( + "Well over 36 hours to ensure no clamping at 9:59:59.99" + ) @pytest.mark.asyncio @@ -77,8 +78,7 @@ async def test_make_single_event_returns_vtt(tmp_path: Path) -> None: sub = Subtitle() out = await sub.make(srt) assert out == "OUT" - # Snapshot should contain the single event - assert d.snapshot == [1000] + assert d.snapshot == [1000], "Snapshot should contain the single event" @pytest.mark.asyncio @@ -94,8 +94,7 @@ async def test_make_two_events_pop_first_when_ends_equal(tmp_path: Path) -> None sub = Subtitle() out = await sub.make(srt) assert out == "OUT" - # Since ends are equal, first should be popped => only last remains - assert d.snapshot == [5000] + assert d.snapshot == [5000], "Since ends are equal, first should be popped => only last remains" @pytest.mark.asyncio @@ -111,5 +110,4 @@ async def test_make_two_events_no_pop_when_different(tmp_path: Path) -> None: sub = Subtitle() out = await sub.make(srt) assert out == "OUT" - # Both remain since ends differ - assert d.snapshot == [5000, 6000] + assert d.snapshot == [5000, 6000], "Both remain since ends differ" diff --git a/app/tests/test_system_routes.py b/app/tests/test_system_routes.py new file mode 100644 index 00000000..4c9087c8 --- /dev/null +++ b/app/tests/test_system_routes.py @@ -0,0 +1,117 @@ +import pytest +from unittest.mock import AsyncMock, patch + +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.UpdateChecker import UpdateChecker +from app.routes.api.system import check_updates + + +class TestCheckUpdatesEndpoint: + """Tests for the check updates endpoint.""" + + def setup_method(self): + """Reset singletons before each test.""" + Config._reset_singleton() + UpdateChecker._reset_singleton() + + @pytest.mark.asyncio + async def test_check_updates_disabled(self): + """Test check updates returns error when disabled in config.""" + config = Config.get_instance() + config.check_for_updates = False + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + response = await check_updates(config, encoder, update_checker) + + assert 400 == response.status, "Should return 400 when update checking is disabled" + body = response.body + assert b"disabled" in body.lower(), "Response should mention update checking is disabled" + + @pytest.mark.asyncio + async def test_check_updates_up_to_date(self): + """Test check updates returns up_to_date status.""" + config = Config.get_instance() + config.check_for_updates = True + config.app_version = "v1.0.0" + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + with patch.object(update_checker, "check_for_updates", new_callable=AsyncMock) as mock_check: + mock_check.return_value = ("up_to_date", None) + response = await check_updates(config, encoder, update_checker) + + assert 200 == response.status, "Should return 200" + body = response.body.decode("utf-8") + assert "up_to_date" in body, "Response should include up_to_date status" + assert mock_check.called, "Should have called check_for_updates" + + @pytest.mark.asyncio + async def test_check_updates_returns_current_version(self): + """Test check updates includes current version in response.""" + config = Config.get_instance() + config.check_for_updates = True + config.app_version = "v1.2.3" + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + with patch.object(update_checker, "check_for_updates", new_callable=AsyncMock) as mock_check: + mock_check.return_value = ("up_to_date", None) + response = await check_updates(config, encoder, update_checker) + + assert 200 == response.status, "Should return 200" + body = response.body.decode("utf-8") + assert "1.2.3" in body, "Response should include current version" + + @pytest.mark.asyncio + async def test_check_updates_update_available(self): + """Test check updates returns update_available status with new version.""" + config = Config.get_instance() + config.check_for_updates = True + config.app_version = "v1.0.0" + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + with patch.object(update_checker, "check_for_updates", new_callable=AsyncMock) as mock_check: + mock_check.return_value = ("update_available", "v1.0.5") + response = await check_updates(config, encoder, update_checker) + + assert 200 == response.status, "Should return 200" + body = response.body.decode("utf-8") + assert "v1.0.5" in body, "Response should include new version" + assert "update_available" in body, "Response should include update_available status" + + @pytest.mark.asyncio + async def test_check_updates_null_when_no_new_version(self): + """Test check updates returns null for new_version when none available.""" + config = Config.get_instance() + config.check_for_updates = True + config.app_version = "v1.0.0" + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + with patch.object(update_checker, "check_for_updates", new_callable=AsyncMock) as mock_check: + mock_check.return_value = ("up_to_date", None) + response = await check_updates(config, encoder, update_checker) + + assert 200 == response.status, "Should return 200" + body = response.body.decode("utf-8") + assert "null" in body.lower(), "Response should include null for new_version when not available" + + @pytest.mark.asyncio + async def test_check_updates_error_status(self): + """Test check updates handles error status correctly.""" + config = Config.get_instance() + config.check_for_updates = True + config.app_version = "v1.0.0" + encoder = Encoder() + update_checker = UpdateChecker.get_instance() + + with patch.object(update_checker, "check_for_updates", new_callable=AsyncMock) as mock_check: + mock_check.return_value = ("error", None) + response = await check_updates(config, encoder, update_checker) + + assert 200 == response.status, "Should return 200" + body = response.body.decode("utf-8") + assert "error" in body, "Response should include error status" diff --git a/app/tests/test_tasks.py b/app/tests/test_tasks.py index 9384bfb7..92163ca3 100644 --- a/app/tests/test_tasks.py +++ b/app/tests/test_tasks.py @@ -321,8 +321,7 @@ class TestTasks: 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._debug is True, "Check initialization" assert tasks._default_preset == "test_preset" assert tasks._file == tasks_file assert tasks._tasks == [] @@ -505,8 +504,7 @@ class TestTasks: assert result == tasks assert tasks_file.exists() - # Verify file content was written - assert tasks_file.stat().st_size > 0 + assert tasks_file.stat().st_size > 0, "Verify file content was written" @patch("app.library.Tasks.Config") @patch("app.library.Tasks.EventBus") @@ -788,8 +786,7 @@ class TestTasks: } ) - # Verify events were emitted - assert mock_eventbus_instance.emit.call_count == 2 + assert mock_eventbus_instance.emit.call_count == 2, "Verify events were emitted" @pytest.mark.asyncio @patch("app.library.Tasks.Config") @@ -963,8 +960,7 @@ class TestHandleTaskInspect: url="https://example.com/feed", preset="default", handler_name=None, static_only=True ) - # Verify result structure - assert hasattr(result, "items") + assert hasattr(result, "items"), "Verify result structure" assert hasattr(result, "metadata") assert result.items == [] assert result.metadata["matched"] is True @@ -1018,8 +1014,7 @@ class TestHandleTaskInspect: # Call inspect with static_only=False (default) result = await handler.inspect(url="https://example.com/feed", preset="default", handler_name=None) - # Verify result structure includes extracted items - assert hasattr(result, "items") + assert hasattr(result, "items"), "Verify result structure includes extracted items" assert hasattr(result, "metadata") assert len(result.items) > 0 assert result.metadata["matched"] is True diff --git a/app/tests/test_update_checker.py b/app/tests/test_update_checker.py new file mode 100644 index 00000000..18cebf34 --- /dev/null +++ b/app/tests/test_update_checker.py @@ -0,0 +1,362 @@ +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +class TestUpdateChecker: + """Test UpdateChecker functionality.""" + + def setup_method(self): + """Set up test fixtures.""" + from app.library.cache import Cache + from app.library.config import Config + from app.library.Events import EventBus + from app.library.Scheduler import Scheduler + from app.library.UpdateChecker import UpdateChecker + + Config._reset_singleton() + Scheduler._reset_singleton() + UpdateChecker._reset_singleton() + EventBus._reset_singleton() + Cache._reset_singleton() + + def test_singleton_pattern(self): + """Test that UpdateChecker follows singleton pattern.""" + from app.library.UpdateChecker import UpdateChecker + + instance1 = UpdateChecker.get_instance() + instance2 = UpdateChecker.get_instance() + + assert instance1 is instance2, "Should return same instance" + + def test_initialization_with_defaults(self): + """Test UpdateChecker initializes with default config and scheduler.""" + from app.library.UpdateChecker import UpdateChecker + + checker = UpdateChecker.get_instance() + + assert checker._config is not None, "Should have config instance" + assert checker._scheduler is not None, "Should have scheduler instance" + assert checker._notify is not None, "Should have EventBus instance" + assert checker._job_id is None, "Should have no job ID initially" + + def test_attach_schedules_check_when_enabled(self): + """Test that attach schedules update check when config.check_for_updates is True.""" + import asyncio + + from app.library.config import Config + from app.library.Events import EventBus + from app.library.Scheduler import Scheduler + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + + # Create scheduler with a fresh event loop + loop = asyncio.new_event_loop() + scheduler = Scheduler.get_instance(loop=loop) + notify = EventBus.get_instance() + + # Mock APP_VERSION to be a release version + with patch("app.library.UpdateChecker.APP_VERSION", "v1.0.0"): + checker = UpdateChecker.get_instance(config=config, scheduler=scheduler) + app_mock = MagicMock() + + try: + checker.attach(app_mock) + + assert checker._job_id is not None, "Should have scheduled a job" + assert "update_checker" == checker._job_id, "Job ID should be 'update_checker'" + finally: + # Clean up + if checker._job_id: + scheduler.remove(checker._job_id) + loop.close() + + def test_attach_skips_scheduling_when_disabled(self): + """Test that attach skips scheduling when config.check_for_updates is False.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = False + + checker = UpdateChecker.get_instance(config=config) + app_mock = MagicMock() + + checker.attach(app_mock) + + assert checker._job_id is None, "Should not have scheduled a job when disabled" + + def test_attach_skips_scheduling_for_dev_version(self): + """Test that attach skips scheduling when running dev version.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + + with patch("app.library.UpdateChecker.APP_VERSION", "dev-master"): + checker = UpdateChecker.get_instance(config=config) + app_mock = MagicMock() + + checker.attach(app_mock) + + assert checker._job_id is None, "Should not schedule job for dev version" + + @pytest.mark.asyncio + async def test_on_shutdown_removes_scheduled_job(self): + """Test that on_shutdown removes the scheduled job.""" + import asyncio + + from app.library.config import Config + from app.library.Events import EventBus + from app.library.Scheduler import Scheduler + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + + # Create scheduler with a fresh event loop + loop = asyncio.new_event_loop() + scheduler = Scheduler.get_instance(loop=loop) + notify = EventBus.get_instance() + + with patch("app.library.UpdateChecker.APP_VERSION", "v1.0.0"): + checker = UpdateChecker.get_instance(config=config, scheduler=scheduler) + app_mock = MagicMock() + + try: + checker.attach(app_mock) + initial_job_id = checker._job_id + + assert initial_job_id is not None, "Should have job ID before shutdown" + + await checker.on_shutdown(app_mock) + + assert checker._job_id is None, "Should clear job ID after shutdown" + finally: + loop.close() + + @pytest.mark.asyncio + async def test_check_for_updates_skips_when_disabled(self): + """Test that check_for_updates skips when config.check_for_updates is False.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = False + config.new_version = "" + + checker = UpdateChecker.get_instance(config=config) + + # Should return disabled status + status, new_version = await checker.check_for_updates() + + assert "disabled" == status, "Should return disabled status" + assert new_version is None, "Should return None for new_version when disabled" + assert "" == config.new_version, "Should not update new_version when disabled" + + @pytest.mark.asyncio + @patch("app.library.UpdateChecker.async_client") + async def test_check_for_updates_finds_newer_version(self, mock_client): + """Test that check_for_updates detects when a newer version is available.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + config.new_version = "" + + # Mock HTTP response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"tag_name": "v99.0.0"} + + mock_context = AsyncMock() + mock_context.__aenter__.return_value.get = AsyncMock(return_value=mock_response) + mock_client.return_value = mock_context + + # Mock current version to be older + with patch("app.library.UpdateChecker.APP_VERSION", "v1.0.0"): + checker = UpdateChecker.get_instance(config=config) + status, new_version = await checker.check_for_updates() + + assert "update_available" == status, "Should return update_available status" + assert "v99.0.0" == new_version, "Should return new version tag" + assert "v99.0.0" == config.new_version, "Should store new version tag" + assert checker._job_id is None, "Should stop scheduled task after finding update" + + @pytest.mark.asyncio + @patch("app.library.UpdateChecker.async_client") + async def test_check_for_updates_no_update_available(self, mock_client): + """Test that check_for_updates correctly handles when no update is available.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + config.new_version = "" + + # Mock HTTP response with older version (current is hardcoded as 1.0.14 in the code) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"tag_name": "v1.0.0"} + + mock_context = AsyncMock() + mock_context.__aenter__.return_value.get = AsyncMock(return_value=mock_response) + mock_client.return_value = mock_context + + checker = UpdateChecker.get_instance(config=config) + checker._job_id = "test-job" # Set a job ID to verify it's not removed + + status, new_version = await checker.check_for_updates() + + assert "up_to_date" == status, "Should return up_to_date status" + assert new_version is None, "Should return None for new_version" + assert "" == config.new_version, "Should clear new_version when no update available" + assert "test-job" == checker._job_id, "Should keep scheduled task running" + + @pytest.mark.asyncio + @patch("app.library.UpdateChecker.async_client") + async def test_check_for_updates_handles_http_error(self, mock_client): + """Test that check_for_updates handles HTTP errors gracefully.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + config.new_version = "" + + # Mock HTTP error + mock_response = MagicMock() + mock_response.status_code = 404 + + mock_context = AsyncMock() + mock_context.__aenter__.return_value.get = AsyncMock(return_value=mock_response) + mock_client.return_value = mock_context + + checker = UpdateChecker.get_instance(config=config) + + status, new_version = await checker.check_for_updates() + + assert "error" == status, "Should return error status" + assert new_version is None, "Should return None for new_version on error" + assert "" == config.new_version, "Should not set new_version on HTTP error" + + @pytest.mark.asyncio + @patch("app.library.UpdateChecker.async_client") + async def test_check_for_updates_handles_exception(self, mock_client): + """Test that check_for_updates handles exceptions gracefully.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + config.new_version = "" + + # Mock exception during HTTP request + mock_context = AsyncMock() + mock_context.__aenter__.return_value.get = AsyncMock(side_effect=Exception("Network error")) + mock_client.return_value = mock_context + + checker = UpdateChecker.get_instance(config=config) + + status, new_version = await checker.check_for_updates() + + assert "error" == status, "Should return error status on exception" + assert new_version is None, "Should return None for new_version on exception" + assert "" == config.new_version, "Should not set new_version on exception" + + def test_compare_versions_newer_available(self): + """Test version comparison detects newer version.""" + from app.library.UpdateChecker import UpdateChecker + + checker = UpdateChecker.get_instance() + + assert checker._compare_versions("1.0.0", "2.0.0") is True, "Should detect 2.0.0 > 1.0.0" + assert checker._compare_versions("1.0.0", "1.1.0") is True, "Should detect 1.1.0 > 1.0.0" + assert checker._compare_versions("1.0.0", "1.0.1") is True, "Should detect 1.0.1 > 1.0.0" + + def test_compare_versions_same_version(self): + """Test version comparison with same version.""" + from app.library.UpdateChecker import UpdateChecker + + checker = UpdateChecker.get_instance() + + assert checker._compare_versions("1.0.0", "1.0.0") is False, "Should detect versions are equal" + + def test_compare_versions_older_version(self): + """Test version comparison with older version.""" + from app.library.UpdateChecker import UpdateChecker + + checker = UpdateChecker.get_instance() + + assert checker._compare_versions("2.0.0", "1.0.0") is False, "Should detect 1.0.0 is not newer than 2.0.0" + + def test_github_api_url_constant(self): + """Test that GitHub API URL is correctly defined.""" + from app.library.UpdateChecker import UpdateChecker + + expected_url = "https://api.github.com/repos/arabcoders/ytptube/releases/latest" + assert UpdateChecker.GITHUB_API_URL == expected_url, "GitHub API URL should be correct" + + @pytest.mark.asyncio + async def test_check_for_updates_stores_tag_name(self): + """Test that check_for_updates stores the tag_name when update found.""" + from app.library.config import Config + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + config.new_version = "" + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"tag_name": "v2.0.0"} + + mock_context = AsyncMock() + mock_context.__aenter__.return_value.get = AsyncMock(return_value=mock_response) + + with patch("app.library.UpdateChecker.async_client", return_value=mock_context): + with patch("app.library.UpdateChecker.APP_VERSION", "v1.0.0"): + checker = UpdateChecker.get_instance(config=config) + await checker.check_for_updates() + + assert "v2.0.0" == config.new_version, "Should store full tag_name including 'v' prefix" + + def test_subscribe_to_started_event(self): + """Test that attach subscribes to Events.STARTED.""" + import asyncio + + from app.library.config import Config + from app.library.Events import EventBus + from app.library.Scheduler import Scheduler + from app.library.UpdateChecker import UpdateChecker + + config = Config.get_instance() + config.check_for_updates = True + + loop = asyncio.new_event_loop() + scheduler = Scheduler.get_instance(loop=loop) + notify = EventBus.get_instance() + + checker = None + try: + with patch("app.library.UpdateChecker.APP_VERSION", "v1.0.0"): + checker = UpdateChecker.get_instance(config=config, scheduler=scheduler) + app_mock = MagicMock() + + checker.attach(app_mock) + + # Verify subscription was created + subscriptions = notify._listeners.get("started", {}) + assert len(subscriptions) > 0, "Should have subscribed to STARTED event" + assert any("UpdateChecker.attach" in name for name in subscriptions.keys()), ( + "Should have UpdateChecker.attach subscription" + ) + finally: + if checker and checker._job_id: + scheduler.remove(checker._job_id) + loop.close() diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 666ea48e..45d1c9d1 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -34,7 +34,7 @@ from app.library.Utils import ( get_files, get_mime_type, get_possible_images, - get_static_ytdlp, + get_ytdlp, init_class, is_private_address, list_folders, @@ -137,9 +137,8 @@ class TestTimedLruCache: def test_function(x): return x * 2 - # Test that methods exist - assert hasattr(test_function, "cache_clear") - assert hasattr(test_function, "cache_info") + assert hasattr(test_function, "cache_clear"), "Cached function should have cache_clear method" + assert hasattr(test_function, "cache_info"), "Cached function should have cache_info method" # Call function to populate cache test_function(5) @@ -256,9 +255,8 @@ class TestAsyncTimedLruCache: async def async_method_test(x): return x + 1 - # Test that cache methods exist - assert hasattr(async_method_test, "cache_clear") - assert hasattr(async_method_test, "cache_info") + assert hasattr(async_method_test, "cache_clear"), "Async cached function should have cache_clear method" + assert hasattr(async_method_test, "cache_info"), "Async cached function should have cache_info method" # Test cache_info info = async_method_test.cache_info() @@ -286,12 +284,11 @@ class TestAsyncTimedLruCache: # Fill cache beyond max_size result1 = await async_limited_func(1) result2 = await async_limited_func(2) - result3 = await async_limited_func(3) # Should evict oldest entry + result3 = await async_limited_func(3) - # Verify results - assert result1 == 4 - assert result2 == 8 - assert result3 == 12 + assert result1 == 4, "async_limited_func(1) should return 4" + assert result2 == 8, "async_limited_func(2) should return 8" + assert result3 == 12, "async_limited_func(3) should return 12 (should evict oldest entry)" # Check cache size is limited info = async_limited_func.cache_info() @@ -664,9 +661,8 @@ class TestMergeDict: source = {"nested": {"a": 1}} destination = {"nested": {"b": 2}, "other": 3} result = merge_dict(source, destination) - # Should merge nested dictionaries - assert "nested" in result - assert "other" in result + assert "nested" in result, "Should merge nested dictionaries" + assert "other" in result, "Should preserve other keys" def test_merge_dict_empty_source(self): """Test merging with empty source.""" @@ -695,7 +691,7 @@ class TestMergeDict: destination = {"existing": "data"} result = merge_dict(source, destination) - assert "__class__" not in result, "__class__ should be filtered out" + assert "__class__" not in result, "__class__ attribute pollution should be blocked" assert result["safe"] == "value", "Safe values should be preserved" assert result["existing"] == "data", "Existing data should be preserved" @@ -742,11 +738,10 @@ class TestMergeDict: # All dangerous attributes should be filtered out dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"] for key in dangerous_keys: - assert key not in result, f"{key} should be filtered out" + assert key not in result, f"{key} should be filtered out (all dangerous attributes)" - # Safe data should be preserved - assert result["safe_key"] == "safe_value" - assert result["existing"] == "data" + assert result["safe_key"] == "safe_value", "Safe data should be preserved" + assert result["existing"] == "data", "Existing data should be preserved" def test_merge_dict_nested_dunder_pollution(self): """Test that nested dangerous attributes are handled correctly.""" @@ -754,11 +749,9 @@ class TestMergeDict: destination = {"nested": {"existing_nested": "original"}} result = merge_dict(source, destination) - # Nested dangerous attributes should be filtered out - assert "__class__" not in result["nested"], "Nested __class__ should be filtered" - # Safe nested data should be preserved - assert result["nested"]["safe_nested"] == "value" - assert result["nested"]["existing_nested"] == "original" + assert "__class__" not in result["nested"], "Nested dangerous attributes should be filtered out" + assert result["nested"]["safe_nested"] == "value", "Safe nested data should be preserved" + assert result["nested"]["existing_nested"] == "original", "Existing nested data should be preserved" def test_merge_dict_prototype_pollution_attempt(self): """Test protection against prototype pollution attempts.""" @@ -766,10 +759,10 @@ class TestMergeDict: destination = {"existing": "value"} result = merge_dict(source, destination) - # These should be treated as regular keys (not filtered unless explicitly in the filter list) - # The function filters Python-specific dangerous attributes, not JavaScript ones - assert result["safe"] == "data" - assert result["existing"] == "value" + assert result["safe"] == "data", ( + "Function filters Python-specific dangerous attributes, not JS ones like __proto__" + ) + assert result["existing"] == "value", "Existing data should be preserved" def test_merge_dict_special_method_pollution(self): """Test with various Python special methods.""" @@ -784,10 +777,10 @@ class TestMergeDict: destination = {"target": "data"} result = merge_dict(source, destination) - # These are not in the current filter list, so they should pass through - # This test documents current behavior - may need updating if more filters are added - assert result["safe"] == "value" - assert result["target"] == "data" + assert result["safe"] == "value", ( + "Safe data should be preserved (special methods not in filter list, documents current behavior)" + ) + assert result["target"] == "data", "Target data should be preserved" def test_merge_dict_list_pollution_safe(self): """Test that list merging doesn't allow dangerous manipulation.""" @@ -795,8 +788,9 @@ class TestMergeDict: destination = {"items": ["old1", "old2"]} result = merge_dict(source, destination) - # Lists should be concatenated safely (destination + source) - assert result["items"] == ["old1", "old2", "new1", "new2"] + assert result["items"] == ["old1", "old2", "new1", "new2"], ( + "Lists should be concatenated safely (destination + source)" + ) def test_merge_dict_deep_nested_pollution(self): """Test with deeply nested dangerous attributes.""" @@ -812,13 +806,13 @@ class TestMergeDict: destination = {"level1": {"level2": {"existing": "data"}}} result = merge_dict(source, destination) - # The function should now properly filter all dangerous keys recursively - assert "__class__" not in result["level1"]["level2"], "Deep __class__ should be filtered" - assert "__globals__" not in result["level1"]["level2"]["level3"], "Very deep __globals__ should be filtered" + assert "__class__" not in result["level1"]["level2"], ( + "Function should properly filter all dangerous keys recursively (deep __class__)" + ) + assert "__globals__" not in result["level1"]["level2"]["level3"], "Function should filter very deep __globals__" - # Safe data should be preserved - assert result["level1"]["level2"]["safe_deep"] == "value" - assert result["level1"]["level2"]["existing"] == "data" + assert result["level1"]["level2"]["safe_deep"] == "value", "Safe nested data should be preserved" + assert result["level1"]["level2"]["existing"] == "data", "Existing nested data should be preserved" def test_merge_dict_type_validation(self): """Test that non-dict parameters are properly rejected.""" @@ -845,13 +839,13 @@ class TestMergeDict: result = merge_dict(original_source, original_destination) - # Original dictionaries should be unchanged - assert original_source == source_copy, "Source should not be modified" - assert original_destination == destination_copy, "Destination should not be modified" + assert original_source == source_copy, "Original source dictionary should be unchanged (immutability)" + assert original_destination == destination_copy, ( + "Original destination dictionary should be unchanged (immutability)" + ) - # Result should be different from both originals - assert result != original_source - assert result != original_destination + assert result != original_source, "Result should be different from source original" + assert result != original_destination, "Result should be different from destination original" def test_merge_dict_custom_max_depth(self): """Test custom max_depth parameter.""" @@ -900,16 +894,14 @@ class TestMergeDict: source = {"items": list(range(3000))} destination = {"items": list(range(2000, 5000))} # 3000 items - # Total would be 6000 items, but limit is 4000 result = merge_dict(source, destination, max_list_size=4000) - # Should have original destination (3000) + truncated source (1000) = 4000 - assert len(result["items"]) == 4000 + assert len(result["items"]) == 4000, ( + "Total would be 6000 items, but limit is 4000: destination (3000) + truncated source (1000)" + ) - # First 3000 should be from destination - assert result["items"][:3000] == list(range(2000, 5000)) - # Next 1000 should be from source (truncated) - assert result["items"][3000:] == list(range(1000)) + assert result["items"][:3000] == list(range(2000, 5000)), "First 3000 items should be from destination" + assert result["items"][3000:] == list(range(1000)), "Next 1000 items should be from source (truncated)" def test_merge_dict_nested_with_limits(self): """Test nested merging with both depth and size limits.""" @@ -2018,8 +2010,7 @@ class TestDecryptDataCornerCases: encrypted = encrypt_data("test data", correct_key) result = decrypt_data(encrypted, wrong_key) - # Should return None on decryption failure - assert result is None + assert result is None, "Should return None on decryption failure with wrong key" def test_decrypt_data_truncated(self): """Test decrypting truncated encrypted data.""" @@ -2175,11 +2166,10 @@ class TestRenameFile: # Rename file new_path, sidecars = rename_file(test_file, "renamed_video.mp4") - # Assertions - assert new_path.exists() - assert "renamed_video.mp4" == new_path.name - assert not test_file.exists() - assert 0 == len(sidecars) + assert new_path.exists(), "Renamed file should exist" + assert "renamed_video.mp4" == new_path.name, "File should have new name" + assert not test_file.exists(), "Original file should not exist" + assert 0 == len(sidecars), "Should have no sidecar files" def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path): """Test renaming a file with subtitle sidecar.""" @@ -2193,12 +2183,11 @@ class TestRenameFile: # Rename file new_path, sidecars = rename_file(test_file, "renamed_video.mp4") - # Assertions - assert new_path.exists() - assert "renamed_video.mp4" == new_path.name - assert not test_file.exists() + assert new_path.exists(), "Renamed file should exist" + assert "renamed_video.mp4" == new_path.name, "File should have new name" + assert not test_file.exists(), "Original file should not exist after rename" - assert 1 == len(sidecars) + assert 1 == len(sidecars), "Should have renamed 1 sidecar file" old_sidecar, new_sidecar = sidecars[0] assert new_sidecar.exists() assert "renamed_video.en.srt" == new_sidecar.name @@ -2223,12 +2212,11 @@ class TestRenameFile: # Rename file new_path, sidecars = rename_file(test_file, "renamed_video.mp4") - # Assertions - assert new_path.exists() - assert "renamed_video.mp4" == new_path.name - assert not test_file.exists() + assert new_path.exists(), "Renamed file should exist" + assert "renamed_video.mp4" == new_path.name, "File should have new name" + assert not test_file.exists(), "Original file should not exist after rename" - assert 3 == len(sidecars) + assert 3 == len(sidecars), "Should have renamed 3 sidecar files" # Check all sidecars were renamed sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} @@ -2236,10 +2224,9 @@ class TestRenameFile: assert "renamed_video.fr.srt" in sidecar_names assert "renamed_video.info.json" in sidecar_names - # Check old files don't exist - assert not subtitle_en.exists() - assert not subtitle_fr.exists() - assert not info_file.exists() + assert not subtitle_en.exists(), "Old subtitle file should not exist after rename" + assert not subtitle_fr.exists(), "Old subtitle file should not exist after rename" + assert not info_file.exists(), "Old info file should not exist after rename" def test_rename_file_destination_exists(self, tmp_path: Path): """Test renaming a file when destination already exists.""" @@ -2254,9 +2241,8 @@ class TestRenameFile: with pytest.raises(ValueError, match="already exists"): rename_file(test_file, "renamed_video.mp4") - # Original files should still exist - assert test_file.exists() - assert existing_file.exists() + assert test_file.exists(), "Original file should still exist when rename fails" + assert existing_file.exists(), "Existing file should still exist when rename fails" def test_rename_file_sidecar_destination_exists(self, tmp_path: Path): """Test renaming when sidecar destination already exists.""" @@ -2295,11 +2281,10 @@ class TestRenameFile: # Rename file new_path, sidecars = rename_file(test_file, "renamed.mp4") - # Assertions - assert new_path.exists() - assert "renamed.mp4" == new_path.name + assert new_path.exists(), "Renamed file should exist" + assert "renamed.mp4" == new_path.name, "File should have new name" - assert 2 == len(sidecars) + assert 2 == len(sidecars), "Should have renamed 2 sidecar files" sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} assert "renamed.en-US.ass" in sidecar_names assert "renamed.thumb.jpg" in sidecar_names @@ -2322,12 +2307,11 @@ class TestMoveFile: # Move file new_path, sidecars = move_file(test_file, target_dir) - # Assertions - assert new_path.exists() - assert "video.mp4" == new_path.name - assert new_path.parent == target_dir - assert not test_file.exists() - assert 0 == len(sidecars) + assert new_path.exists(), "Moved file should exist at destination" + assert "video.mp4" == new_path.name, "File should keep same name" + assert new_path.parent == target_dir, "File should be in target directory" + assert not test_file.exists(), "Original file should not exist after move" + assert 0 == len(sidecars), "Should have no sidecar files" def test_move_file_with_subtitle_sidecar(self, tmp_path: Path): """Test moving a file with subtitle sidecar.""" @@ -2346,13 +2330,12 @@ class TestMoveFile: # Move file new_path, sidecars = move_file(test_file, target_dir) - # Assertions - assert new_path.exists() - assert "video.mp4" == new_path.name - assert new_path.parent == target_dir - assert not test_file.exists() + assert new_path.exists(), "Moved file should exist at destination" + assert "video.mp4" == new_path.name, "File should keep same name" + assert new_path.parent == target_dir, "File should be in target directory" + assert not test_file.exists(), "Original file should not exist after move" - assert 1 == len(sidecars) + assert 1 == len(sidecars), "Should have moved 1 sidecar file" old_sidecar, new_sidecar = sidecars[0] assert new_sidecar.exists() assert "video.en.srt" == new_sidecar.name @@ -2383,13 +2366,12 @@ class TestMoveFile: # Move file new_path, sidecars = move_file(test_file, target_dir) - # Assertions - assert new_path.exists() - assert "video.mp4" == new_path.name - assert new_path.parent == target_dir - assert not test_file.exists() + assert new_path.exists(), "Moved file should exist at destination" + assert "video.mp4" == new_path.name, "File should keep same name" + assert new_path.parent == target_dir, "File should be in target directory" + assert not test_file.exists(), "Original file should not exist after move" - assert 3 == len(sidecars) + assert 3 == len(sidecars), "Should have moved 3 sidecar files" # Check all sidecars were moved sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} @@ -2399,12 +2381,11 @@ class TestMoveFile: # Check all are in target directory for _old_sidecar, new_sidecar in sidecars: - assert new_sidecar.parent == target_dir + assert new_sidecar.parent == target_dir, "All sidecars should be in target directory" - # Check old files don't exist - assert not subtitle_en.exists() - assert not subtitle_fr.exists() - assert not info_file.exists() + assert not subtitle_en.exists(), "Old subtitle file should not exist after move" + assert not subtitle_fr.exists(), "Old subtitle file should not exist after move" + assert not info_file.exists(), "Old info file should not exist after move" def test_move_file_destination_exists(self, tmp_path: Path): """Test moving a file when destination already exists.""" @@ -2423,9 +2404,8 @@ class TestMoveFile: with pytest.raises(ValueError, match="already exists"): move_file(test_file, target_dir) - # Original files should still exist - assert test_file.exists() - assert existing_file.exists() + assert test_file.exists(), "Original file should still exist when move fails" + assert existing_file.exists(), "Existing file should still exist when move fails" def test_move_file_sidecar_destination_exists(self, tmp_path: Path): """Test moving when sidecar destination already exists.""" @@ -2682,41 +2662,43 @@ class TestGetExtras: class TestGetStaticYtdlp: """Test the get_static_ytdlp function.""" + def setup_method(self): + """Reset YTDLP singleton state before each test.""" + import app.library.Utils as Utils + + Utils.YTDLP_INFO_CLS = None + def test_get_static_ytdlp_returns_instance(self): """Test that get_static_ytdlp returns a YTDLP instance.""" - from app.library.Utils import get_static_ytdlp from app.library.ytdlp import YTDLP - # Force reload to ensure we get a real instance, not a mock - instance = get_static_ytdlp(reload=True) + # Get the cached instance + instance = get_ytdlp() assert instance is not None assert isinstance(instance, YTDLP) def test_get_static_ytdlp_returns_same_instance(self): """Test that get_static_ytdlp returns the same cached instance.""" - from app.library.Utils import get_static_ytdlp - instance1 = get_static_ytdlp() - instance2 = get_static_ytdlp() + instance1 = get_ytdlp() + instance2 = get_ytdlp() assert instance1 is instance2 - def test_get_static_ytdlp_reload(self): - """Test that get_static_ytdlp can reload and return a new instance.""" - from app.library.Utils import get_static_ytdlp + def test_get_static_ytdlp_with_params(self): + """Test that get_static_ytdlp returns a new instance when params are provided.""" - instance1 = get_static_ytdlp() - instance2 = get_static_ytdlp(reload=True) + instance1 = get_ytdlp() + instance2 = get_ytdlp(params={"quiet": False}) assert instance1 is not instance2 assert instance2 is not None def test_get_static_ytdlp_has_correct_params(self): """Test that get_static_ytdlp initializes with correct parameters.""" - from app.library.Utils import get_static_ytdlp - instance = get_static_ytdlp(reload=True) + instance = get_ytdlp() # Access the internal params params = instance.params @@ -2732,9 +2714,14 @@ class TestGetStaticYtdlp: class TestParseOuttmpl: """Test the parse_outtmpl function.""" + def setup_method(self): + """Reset YTDLP singleton state before each test.""" + import app.library.Utils as Utils + + Utils.YTDLP_INFO_CLS = None + def test_parse_outtmpl_basic(self): """Test basic template parsing with simple placeholders.""" - from app.library.Utils import parse_outtmpl template = "%(title)s.%(ext)s" info_dict = { @@ -2748,7 +2735,6 @@ class TestParseOuttmpl: def test_parse_outtmpl_with_id(self): """Test template parsing with video ID.""" - from app.library.Utils import parse_outtmpl template = "[%(id)s] %(title)s.%(ext)s" info_dict = { @@ -2763,7 +2749,6 @@ class TestParseOuttmpl: def test_parse_outtmpl_with_uploader(self): """Test template parsing with uploader information.""" - from app.library.Utils import parse_outtmpl template = "%(uploader)s - %(title)s.%(ext)s" info_dict = { @@ -2778,7 +2763,6 @@ class TestParseOuttmpl: def test_parse_outtmpl_with_nested_path(self): """Test template parsing with nested directory structure.""" - from app.library.Utils import parse_outtmpl template = "%(uploader)s/%(title)s.%(ext)s" info_dict = { @@ -2793,22 +2777,19 @@ class TestParseOuttmpl: def test_parse_outtmpl_with_missing_field(self): """Test template parsing with missing field defaults to NA.""" - from app.library.Utils import parse_outtmpl template = "%(title)s - %(upload_date)s.%(ext)s" info_dict = { "title": "Test Video", "ext": "mp4", - # upload_date is missing } result = parse_outtmpl(template, info_dict) - assert result == "Test Video - NA.mp4" + assert result == "Test Video - NA.mp4", "Missing field upload_date should default to NA" def test_parse_outtmpl_complex(self): """Test complex template with multiple fields.""" - from app.library.Utils import parse_outtmpl template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s" info_dict = { @@ -2826,7 +2807,6 @@ class TestParseOuttmpl: def test_parse_outtmpl_with_special_characters(self): """Test template parsing handles special characters in values.""" - from app.library.Utils import parse_outtmpl template = "%(title)s.%(ext)s" info_dict = { @@ -2836,13 +2816,11 @@ class TestParseOuttmpl: result = parse_outtmpl(template, info_dict) - # yt-dlp sanitizes special characters in filenames - assert ".mp4" in result - assert "Test" in result + assert ".mp4" in result, "yt-dlp should sanitize special characters but preserve extension" + assert "Test" in result, "yt-dlp should preserve safe parts of title" def test_parse_outtmpl_with_playlist_info(self): """Test template parsing with playlist information.""" - from app.library.Utils import parse_outtmpl template = "%(playlist)s/%(title)s.%(ext)s" info_dict = { @@ -2854,3 +2832,19 @@ class TestParseOuttmpl: result = parse_outtmpl(template, info_dict) assert result == "My Playlist/Video Title.webm" + + def test_parse_outtmpl_with_restrict_filename(self): + """Test template parsing with restrict_filename parameter.""" + + template = "%(uploader)s/%(title)s.%(ext)s" + info_dict = { + "uploader": "Foobar's Workshop", + "title": "Test Video", + "ext": "mp4", + } + + result_unrestricted: str = parse_outtmpl(template, info_dict) + assert result_unrestricted == "Foobar's Workshop/Test Video.mp4" + + result_restricted: str = parse_outtmpl(template, info_dict, params={"restrictfilenames": True}) + assert result_restricted == "Foobar_s_Workshop/Test_Video.mp4" diff --git a/app/tests/test_ytdlp_module.py b/app/tests/test_ytdlp_module.py index eec4568c..880a929c 100644 --- a/app/tests/test_ytdlp_module.py +++ b/app/tests/test_ytdlp_module.py @@ -75,8 +75,7 @@ class TestYtDlpOptions: # For any ignored flag that actually exists in yt-dlp parser, ensure it is marked ignored present_ignored_flags = [f for f in ignored_flags if f in flag_to_ignored] - # We expect at least one to be present (e.g., -P / --paths, etc.) - assert len(present_ignored_flags) > 0 + assert len(present_ignored_flags) > 0, "We expect at least one to be present (e.g., -P / --paths, etc.)" assert all(flag_to_ignored[f] is True for f in present_ignored_flags) @@ -111,8 +110,7 @@ class TestYTDLP: # Our __init__ code manually sets these after super() ytdlp.params["download_archive"] = "/tmp/archive.txt" - # Verify download_archive was restored to params - assert ytdlp.params["download_archive"] == "/tmp/archive.txt" + assert ytdlp.params["download_archive"] == "/tmp/archive.txt", "Verify download_archive was restored to params" # Verify archive proxy was set up assert isinstance(ytdlp.archive, _ArchiveProxy) @@ -133,8 +131,7 @@ class TestYTDLP: assert call_kwargs["params"]["quiet"] is True assert call_kwargs["auto_init"] is False - # Verify archive proxy is falsey - assert isinstance(ytdlp.archive, _ArchiveProxy) + assert isinstance(ytdlp.archive, _ArchiveProxy), "Verify archive proxy is falsey" assert not ytdlp.archive @patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__") @@ -230,8 +227,7 @@ class TestYTDLP: assert ytdlp.archive.add.call_count == 3 calls = [call[0][0] for call in ytdlp.archive.add.call_args_list] - # First call is main archive_id - assert calls[0] == "youtube new123" + assert calls[0] == "youtube new123", "First call is main archive_id" # Should add old IDs except the duplicate assert "youtube old123" in calls diff --git a/app/tests/test_ytdlpopts.py b/app/tests/test_ytdlpopts.py index a7e6d5e3..916aad42 100644 --- a/app/tests/test_ytdlpopts.py +++ b/app/tests/test_ytdlpopts.py @@ -474,8 +474,7 @@ class TestYTDLPOpts: assert "Failed to load" in error_args assert "Cookie Preset" in error_args - # cookiefile should not be set - assert "cookiefile" not in opts._preset_opts + assert "cookiefile" not in opts._preset_opts, "cookiefile should not be set" def test_replacer_substitution_in_cli(self): """Test that CLI arguments get replacer substitution.""" @@ -540,13 +539,11 @@ class TestARGSMerger: merger.add(cli_with_comments) - # Comments should be filtered out - assert "#" not in merger.as_string() + assert "#" not in merger.as_string(), "Comments should be filtered out" assert "This is a comment" not in merger.as_string() assert "Another comment" not in merger.as_string() - # Valid options should remain - assert "--format" in merger.args + assert "--format" in merger.args, "Valid options should remain" assert "best" in merger.args assert "--output" in merger.args assert "test.mp4" in merger.args @@ -566,14 +563,12 @@ class TestARGSMerger: merger.add(cli_with_indented_comments) - # Comments should be filtered out result = merger.as_string() - assert "# Indented comment with spaces" not in result + assert "# Indented comment with spaces" not in result, "Comments should be filtered out" assert "# Indented comment with tabs" not in result assert "# Another indented comment" not in result - # Valid options should remain - assert "--format" in merger.args + assert "--format" in merger.args, "Valid options should remain" assert "--output" in merger.args assert "--socket-timeout" in merger.args @@ -591,13 +586,14 @@ class TestARGSMerger: result = merger.as_string() - # Commented lines should be filtered out completely - assert "player_js_version=actual" not in result - # Check the specific commented variant (with comma before web_safari, not dash) - assert "mweb,web_safari;formats=incomplete" not in result + assert "player_js_version=actual" not in result, "Commented lines should be filtered out completely" + assert "mweb,web_safari;formats=incomplete" not in result, ( + "Check the specific commented variant (with comma before web_safari, not dash)" + ) - # Valid extractor-args should remain (with -web_safari, note the dash) - assert "youtube:player-client=default,tv,mweb,-web_safari;formats=incomplete" in result + assert "youtube:player-client=default,tv,mweb,-web_safari;formats=incomplete" in result, ( + "Valid extractor-args should remain (with -web_safari, note the dash)" + ) assert "--socket-timeout" in merger.args assert "60" in merger.args @@ -839,8 +835,7 @@ class TestYTDLPCli: cli = YTDLPCli(item=item) command, info = cli.build() - # Should use preset values - assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s" + assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s", "Should use preset values" assert info["merged"]["save_path"] == "/downloads/preset_folder" assert "--format 720p" in command @@ -879,11 +874,9 @@ class TestYTDLPCli: cli = YTDLPCli(item=item) command, info = cli.build() - # Should use user values, not preset - assert info["merged"]["template"] == "%(id)s.%(ext)s" + assert info["merged"]["template"] == "%(id)s.%(ext)s", "Should use user values, not preset" assert info["merged"]["save_path"] == "/downloads/user_folder" - # User CLI should appear after preset CLI in command - assert "--format best" in command + assert "--format best" in command, "User CLI should appear after preset CLI in command" @patch("app.library.Presets.Presets") @patch("app.library.YTDLPOpts.Config") @@ -998,8 +991,9 @@ class TestYTDLPCli: cli = YTDLPCli(item=item) command, info = cli.build() - # The implementation strips leading slash and joins with download_path - assert info["merged"]["save_path"] == "/downloads/absolute/path" + assert info["merged"]["save_path"] == "/downloads/absolute/path", ( + "The implementation strips leading slash and joins with download_path" + ) assert "--paths" in command @patch("app.library.Presets.Presets") @@ -1053,5 +1047,4 @@ class TestYTDLPCli: preset_format_idx = args_list.index("720p") if "720p" in args_list else -1 user_format_idx = args_list.index("best") if "best" in args_list else -1 - # User's 'best' should appear after preset's '720p' - assert user_format_idx > preset_format_idx + assert user_format_idx > preset_format_idx, "User's 'best' should appear after preset's '720p'" diff --git a/ui/app/layouts/default.vue b/ui/app/layouts/default.vue index fcf977f2..41e5845e 100644 --- a/ui/app/layouts/default.vue +++ b/ui/app/layouts/default.vue @@ -157,34 +157,113 @@ -
-
-
- © {{ Year }} - YTPTube - ( - {{ config?.app?.app_version || 'unknown' }}) - - yt-dlp -  ({{ config?.app?.ytdlp_version || 'unknown' }}) - - CHANGELOG - - FAQ - - README - - API - - - - Top - +
+
+
+
+ + + + YTPTube + + + + {{ config?.app?.app_version || 'unknown' }} + +
+ + + +

+ + + + {{ moment.unix(config.app?.started).fromNow() }} + + +

+
+ +
+
+

Powered by

+ + + + yt-dlp + + + {{ config?.app?.ytdlp_version || 'unknown' }} +
+
+ +
+

Quick Links

+
+ + + + Changelog + + + + + + FAQ + + + + + + Docs + + + + + + API + + + +
-
-
- - {{ moment.unix(config.app?.started).fromNow() }} - -
-
-
+
@@ -204,7 +283,6 @@ import Shutdown from '~/components/shutdown.vue' import Markdown from '~/components/Markdown.vue' import Connection from '~/components/Connection.vue' -const Year = new Date().getFullYear() const selectedTheme = useStorage('theme', 'auto') const socket = useSocketStore() const config = useConfigStore() @@ -218,6 +296,53 @@ const app_shutdown = ref(false) const simpleMode = useStorage('simple_mode', config.app.simple_mode || false) const show_settings = ref(false) const doc = ref<{ file: string }>({ file: '' }) +const checkingUpdates = ref(false) +const updateCheckMessage = ref('Up to date - Click to check') + +const checkForUpdates = async () => { + if (checkingUpdates.value) { + return + } + + const msg = 'Up to date - Click to check' + + try { + checkingUpdates.value = true + updateCheckMessage.value = 'Checking...' + + const response = await fetch('/api/system/check-updates', { method: 'POST' }) + + if (!response.ok) { + await response.json() + updateCheckMessage.value = 'Check failed' + setTimeout(() => updateCheckMessage.value = msg, 3000) + return + } + + const data = await response.json() + + switch (data.status) { + case 'update_available': + updateCheckMessage.value = 'Update found!' + config.app.new_version = data.new_version + break + case 'up_to_date': + updateCheckMessage.value = 'Up to date ✓' + setTimeout(() => updateCheckMessage.value = msg, 3000) + break + case 'error': + updateCheckMessage.value = 'Check failed' + setTimeout(() => updateCheckMessage.value = msg, 3000) + break + } + } catch (e) { + console.error('Update check failed:', e) + updateCheckMessage.value = 'Check failed' + setTimeout(() => updateCheckMessage.value = msg, 3000) + } finally { + checkingUpdates.value = false + } +} const applyPreferredColorScheme = (scheme: string) => { if (!scheme || scheme === 'auto') { @@ -515,5 +640,14 @@ const scrollToTop = () => document.getElementById('top')?.scrollIntoView({ behav .basic-wrapper.settings-open { transform: translateX(0); } + + .footer .footer-divider { + border-left: none !important; + border-right: none !important; + padding: 0 !important; + border-top: 1px solid rgba(128, 128, 128, 0.2); + padding-top: 1.5rem !important; + margin-top: 1.5rem !important; + } } diff --git a/ui/app/pages/presets.vue b/ui/app/pages/presets.vue index a1ef8b65..2966c463 100644 --- a/ui/app/pages/presets.vue +++ b/ui/app/pages/presets.vue @@ -408,25 +408,13 @@ const editItem = (item: Preset) => { onMounted(async () => (socket.isConnected ? await reloadContent(true) : '')) const exportItem = (item: Preset) => { - const keys = ['id', 'default', 'raw', 'cookies', 'toggle_description'] - const data = JSON.parse(JSON.stringify(item)) - - for (const key of keys) { - if (key in data) { - const { [key]: _, ...rest } = data - Object.assign(data, rest) - } - } - - const userData: Record = {} - for (const key of Object.keys(data)) { - if (data[key]) { - userData[key] = data[key] - } - } + const excludedKeys = ['id', 'default', 'raw', 'cookies', 'toggle_description'] + const userData = Object.fromEntries( + Object.entries(JSON.parse(JSON.stringify(item))).filter(([key, value]) => !excludedKeys.includes(key) && value) + ) userData['_type'] = 'preset' - userData['_version'] = '2.5' + userData['_version'] = '2.6' copyText(encode(userData)) } diff --git a/ui/app/stores/ConfigStore.ts b/ui/app/stores/ConfigStore.ts index 44b1b64d..2e874e47 100644 --- a/ui/app/stores/ConfigStore.ts +++ b/ui/app/stores/ConfigStore.ts @@ -26,6 +26,8 @@ export const useConfigStore = defineStore('config', () => { app_env: 'production', simple_mode: false, default_pagination: 50, + check_for_updates: true, + new_version: '', }, presets: [ { diff --git a/ui/app/types/config.d.ts b/ui/app/types/config.d.ts index 6da002b3..90be9755 100644 --- a/ui/app/types/config.d.ts +++ b/ui/app/types/config.d.ts @@ -45,6 +45,10 @@ type AppConfig = { app_env: "production" | "development" /** Default number of items per page for pagination */ default_pagination: number + /** Indicates if the app should check for updates */ + check_for_updates: boolean + /** New version available, empty string if none */ + new_version: string } type ConfigState = {